KB8PMY
/
meshcore-client
mirror of https://github.com/OffbandMesh/meshcore-client.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat(#763): recover a short contact sync with paced, capped re-syncs and by-key checks

Browse Source 
Replaces the single retry with the owner-approved recovery (#707). After a
short full sync the app waits until the radio has been quiet for 10 s and
no other startup sync is running, syncs again, and stops at 3 full syncs
per connection. Contacts delivered by any of them count toward the
declared total. Saved contacts still missing are then checked one at a
time by key, 250 ms apart.

Only a contact the radio answers "not found" for is removed, and only
once every declared contact is accounted for: a union of several syncs
can reach the total while a different contact is still missing, so
"absent from the union" is never proof. Unanswered contacts are kept.
The banner appears only if the list is still short after all of this.

Recovery state belongs to one connection and one radio; a disconnect or
radio switch bumps a run counter that stops any pending step.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
fix/793-log-composer-budget
diffwireauto 1 month ago committed by Benjamin Wiechel
parent 6f221b1136
commit 0158424da4
2 changed files with 356 additions and 60 deletions
Show Stats Download Patch File Download Diff File

187
lib/connector/meshcore_connector.dart
Unescape View File

@ -453,8 +453,28 @@ class MeshCoreConnector extends ChangeNotifier {
bool _contactSyncIsFull = false;
int? _contactSyncDeclaredTotal;
final Set<String> _contactSyncDeliveredKeys = {};
bool _contactSyncRetryUsed = false;
bool _contactSyncDecisionPending = false;
// Forced resync (#703): after a short full sync the app waits for the link to
// settle, syncs again (at most [contactSyncMaxFullStreams] full syncs per
// connection, owner decision), then asks the radio by key for each saved
// contact still missing. Only a contact the radio answers "not found" for is
// ever removed by it. All of this is per connection: a disconnect or radio
// switch bumps [_recoveryRun] and drops it.
static const int contactSyncMaxFullStreams = 3;
static const Duration defaultContactRecoverySettle = Duration(seconds: 10);
static const Duration defaultContactKeyCheckGap = Duration(milliseconds: 250);
@visibleForTesting
Duration contactRecoverySettle = defaultContactRecoverySettle;
@visibleForTesting
Duration contactKeyCheckGap = defaultContactKeyCheckGap;
int _contactFullStreams = 0;
int _recoveryRun = 0;
String? _recoveryRadioKey;
final Set<String> _recoveryDelivered = {};
final Set<String> _recoveryGone = {};
bool _recoveryCheckingKeys = false;
Timer? _recoverySettleTimer;
DateTime _lastRadioFrameAt = DateTime.fromMillisecondsSinceEpoch(0);
ContactSyncShortfall? _contactSyncShortfall;
// One contact request at a time: the firmware refuses a second
// CMD_GET_CONTACTS mid-stream with ERR_CODE_BAD_STATE. A request made while
@ -3164,7 +3184,14 @@ class MeshCoreConnector extends ChangeNotifier {
_contactSyncIsFull = false;
_contactSyncDeclaredTotal = null;
_contactSyncDeliveredKeys.clear();
_contactSyncRetryUsed = false;
_contactFullStreams = 0;
_recoveryRun++;
_recoveryRadioKey = null;
_recoveryDelivered.clear();
_recoveryGone.clear();
_recoveryCheckingKeys = false;
_recoverySettleTimer?.cancel();
_recoverySettleTimer = null;
_contactSyncDecisionPending = false;
_contactRequestWatchdog?.cancel();
_contactRequestWatchdog = null;
@ -3758,6 +3785,7 @@ class MeshCoreConnector extends ChangeNotifier {
_contactSyncTotal = null;
_contactSyncReceived = 0;
_contactSyncUsesSinceFilter = since != null;
if (since == null) _recoverySettleTimer?.cancel();
_beginContactSync(full: since == null);
if (!preserveExisting) {
_hasLoadedContacts = false;
@ -3870,9 +3898,12 @@ class MeshCoreConnector extends ChangeNotifier {
/// Contacts the app holds that the last full sync did not deliver, i.e. what
/// "use the radio's list" would remove.
int get contactSyncUndeliveredCount => _contacts
.where((c) => !_contactSyncDeliveredKeys.contains(c.publicKeyHex))
.where((c) => !_recoveryDelivered.contains(c.publicKeyHex))
.length;
/// Saved contacts the radio answered "not found" for during recovery.
int get contactSyncConfirmedGoneCount => _recoveryGone.length;
/// A full sync is complete only when the radio declared a total and at
/// least that many contacts arrived. No declared total (older firmware) is
/// unverifiable, so it is never treated as complete.
@ -3892,6 +3923,13 @@ class MeshCoreConnector extends ChangeNotifier {
void _finishFullContactSync() {
final declared = _contactSyncDeclaredTotal;
final received = _contactSyncReceived;
_contactFullStreams++;
if (_recoveryRadioKey != selfPublicKeyHex) {
_recoveryRadioKey = selfPublicKeyHex;
_recoveryDelivered.clear();
_recoveryGone.clear();
}
_recoveryDelivered.addAll(_contactSyncDeliveredKeys);
if (isContactSyncComplete(declared: declared, received: received)) {
final before = _contacts.length;
_contacts.removeWhere(
@ -3907,39 +3945,128 @@ class MeshCoreConnector extends ChangeNotifier {
}
_contactSyncShortfall = null;
_contactSyncDecisionPending = false;
_recoveryDelivered.clear();
_recoveryGone.clear();
return;
}
_appDebugLogService?.warn(
'Contact sync short: radio declared ${declared ?? 'no total'}, sent '
'$received (full sync $_contactFullStreams of '
'$contactSyncMaxFullStreams). Nothing removed.',
tag: 'ContactSync',
);
// A user refresh during the by-key checks only adds what it delivered;
// the checks finish the run.
if (_recoveryCheckingKeys) return;
final unionComplete =
declared != null && _recoveryDelivered.length >= declared;
if (!unionComplete && _contactFullStreams < contactSyncMaxFullStreams) {
_scheduleRecoveryStream(_recoveryRun);
return;
}
unawaited(_checkMissingContactsByKey(_recoveryRun, declared));
}
void _scheduleRecoveryStream(int run) {
_recoverySettleTimer?.cancel();
_recoverySettleTimer = Timer(
contactRecoverySettle,
() => _runRecoveryStreamWhenSettled(run),
);
}
/// Waits until the radio has been quiet for [contactRecoverySettle] and no
/// other startup sync is running, so recovery never hammers the link.
void _runRecoveryStreamWhenSettled(int run) {
if (run != _recoveryRun || !isConnected) return;
final busy =
_contactRequestInFlight ||
_isSyncingChannels ||
_channelSyncInFlight ||
isSyncingQueuedMessages;
final quietFor = contactSyncClock().difference(_lastRadioFrameAt);
if (busy || quietFor < contactRecoverySettle) {
final wait = busy
? contactRecoverySettle
: contactRecoverySettle - quietFor;
_recoverySettleTimer = Timer(
wait < const Duration(seconds: 1) ? const Duration(seconds: 1) : wait,
() => _runRecoveryStreamWhenSettled(run),
);
return;
}
_appDebugLogService?.info(
'Contact recovery: full sync ${_contactFullStreams + 1} of '
'$contactSyncMaxFullStreams',
tag: 'ContactSync',
);
unawaited(
getContacts().catchError((Object e) {
_appDebugLogService?.error(
'Contact recovery sync failed to send: $e',
tag: 'ContactSync',
);
}),
);
}
/// After the last full sync: asks the radio, one key at a time and spaced
/// apart, about each saved contact no sync delivered. Removes only the ones
/// it answers "not found" for, and only when every declared contact has
/// been accounted for; otherwise the owner decides.
Future<void> _checkMissingContactsByKey(int run, int? declared) async {
final missing = _contacts
.where((c) => !_recoveryDelivered.contains(c.publicKeyHex))
.map((c) => c.publicKey)
.toList();
_recoveryCheckingKeys = true;
for (final key in missing) {
await Future<void>.delayed(contactKeyCheckGap);
if (run != _recoveryRun || !isConnected) return;
final result = await checkContactOnRadio(key);
if (run != _recoveryRun) return;
final hex = pubKeyToHex(key);
if (result == ContactKeyCheckResult.onRadio) {
_recoveryDelivered.add(hex);
} else if (result == ContactKeyCheckResult.notFound) {
_recoveryGone.add(hex);
}
}
_recoveryCheckingKeys = false;
final accounted = _recoveryDelivered.length;
if (declared != null && accounted >= declared) {
final gone = Set<String>.of(_recoveryGone);
_contacts.removeWhere((c) => gone.contains(c.publicKeyHex));
_appDebugLogService?.info(
'Contact recovery complete ($accounted of $declared): removed '
'${gone.length} contact(s) the radio no longer holds',
tag: 'ContactSync',
);
_contactSyncShortfall = null;
_contactSyncDecisionPending = false;
notifyListeners();
await _persistContacts();
return;
}
final kept = contactSyncUndeliveredCount;
_contactSyncShortfall = ContactSyncShortfall(
declared: declared,
received: received,
received: accounted,
keptLocally: kept,
);
_contactSyncDecisionPending = true;
final message =
'Contact sync incomplete: radio declared ${declared ?? 'no total'}, '
'sent $received. Kept $kept saved contact(s) it did not send; '
'nothing removed.';
'Contact sync incomplete after $_contactFullStreams full sync(s): '
'radio declared ${declared ?? 'no total'}, app has $accounted. Kept '
'$kept saved contact(s) it did not send (${_recoveryGone.length} '
'confirmed gone); nothing removed.';
debugPrint(message);
_appDebugLogService?.error(message, tag: 'ContactSync');
if (!_contactSyncRetryUsed) {
_contactSyncRetryUsed = true;
_appDebugLogService?.warn(
'Retrying the contact sync once',
tag: 'ContactSync',
);
unawaited(
getContacts().catchError((Object e) {
_appDebugLogService?.error(
'Contact sync retry failed to send: $e',
tag: 'ContactSync',
);
}),
);
} else {
_contactSyncDecisionPending = true;
}
notifyListeners();
}
/// The owner chose to keep the local list after a short sync.
@ -3952,13 +4079,11 @@ class MeshCoreConnector extends ChangeNotifier {
notifyListeners();
}
/// The owner chose the radio's list after a short sync: drop what it did not
/// deliver.
/// The owner chose the radio's list after a short sync: drop what no sync on
/// this connection delivered (confirmed gone, or never sent by the radio).
Future<void> resolveContactSyncUseRadio() async {
final removed = contactSyncUndeliveredCount;
_contacts.removeWhere(
(c) => !_contactSyncDeliveredKeys.contains(c.publicKeyHex),
);
_contacts.removeWhere((c) => !_recoveryDelivered.contains(c.publicKeyHex));
_contactSyncDecisionPending = false;
_contactSyncShortfall = null;
_appDebugLogService?.warn(
@ -5834,6 +5959,7 @@ class MeshCoreConnector extends ChangeNotifier {
void _handleFrameInner(List<int> data) {
if (data.isEmpty) return;
_lastRxTime = DateTime.now();
_lastRadioFrameAt = contactSyncClock();
final frame = Uint8List.fromList(data);
_receivedFramesController.add(frame);
@ -8935,6 +9061,7 @@ class MeshCoreConnector extends ChangeNotifier {
_channelsChangedDebounce?.cancel();
_contactRequestWatchdog?.cancel();
_pendingKeyCheck?.timer?.cancel();
_recoverySettleTimer?.cancel();
radioStatsNotifier.dispose();
_receivedFramesController.close();
_usbManager.dispose();

229
test/connector/contact_sync_test.dart
Unescape View File

@ -65,6 +65,9 @@ void main() {
connector = MeshCoreConnector();
connector.contactsForTest.clear();
connector.contactRecoverySettle = Duration.zero;
connector.contactKeyCheckGap = Duration.zero;
connector.contactKeyCheckTimeout = const Duration(milliseconds: 20);
sent = [];
connector.sendFrameOverrideForTest = sent.add;
connector.setConnectedForTest();
@ -136,65 +139,89 @@ void main() {
expect(contactRequests(), 1, reason: 'no retry after a complete sync');
});
test('a short sync removes nothing, merges, reports, and retries', () async {
Future<void> settle() =>
Future<void>.delayed(const Duration(milliseconds: 5));
int byKeyRequests() =>
sent.where((f) => f.isNotEmpty && f[0] == cmdGetContactByKey).length;
/// Answers the recovery request the app just sent with [streamed].
Future<void> answerStream(int? declared, List<Uint8List> streamed) async {
connector.handleFrameForTest(
declared == null
? Uint8List.fromList([respCodeContactsStart])
: _start(declared),
);
for (final f in streamed) {
connector.handleFrameForTest(f);
}
connector.handleFrameForTest(_end);
await settle();
}
/// Three short full syncs that each deliver only Alpha; the by-key checks
/// that follow get no reply, so every missing contact stays unresolved.
Future<void> threeShortSyncs({int? declared = 350}) async {
seedSaved();
final alpha = [_contactFrame(0x11, 'Alpha')];
await connector.getContacts();
await answerStream(declared, alpha);
await answerStream(declared, alpha);
await answerStream(declared, alpha);
await Future<void>.delayed(const Duration(milliseconds: 100));
}
test('a short sync removes nothing, merges, and recovers quietly', () async {
seedSaved();
await fullSync(
declared: 350,
streamed: [_contactFrame(0x11, 'Alpha'), _contactFrame(0x44, 'Delta')],
);
await settle();
expect(names(), {'Alpha', 'Bravo', 'Charlie', 'Delta'});
final shortfall = connector.contactSyncShortfall!;
expect(shortfall.declared, 350);
expect(shortfall.received, 2);
expect(shortfall.keptLocally, 2);
expect(
connector.contactSyncShortfall,
isNull,
reason: 'no banner while recovery is still running',
);
expect(connector.contactSyncDecisionPending, isFalse);
expect(contactRequests(), 2, reason: 'one automatic retry');
expect(contactRequests(), 2, reason: 'recovery sync after the settle');
});
test('no declared total is never treated as complete', () async {
seedSaved();
await connector.getContacts();
connector.handleFrameForTest(Uint8List.fromList([respCodeContactsStart]));
connector.handleFrameForTest(_contactFrame(0x11, 'Alpha'));
connector.handleFrameForTest(_end);
await Future<void>.delayed(Duration.zero);
await threeShortSyncs(declared: null);
expect(names(), {'Alpha', 'Bravo', 'Charlie'});
expect(connector.contactSyncShortfall!.declared, isNull);
expect(connector.contactSyncDecisionPending, isTrue);
});
group('after the retry is also short', () {
Future<void> twoShortSyncs() async {
seedSaved();
final streamed = [_contactFrame(0x11, 'Alpha')];
await fullSync(declared: 350, streamed: streamed);
// The retry was sent by the first END; the radio answers it short too.
connector.handleFrameForTest(_start(350));
connector.handleFrameForTest(streamed.first);
connector.handleFrameForTest(_end);
await Future<void>.delayed(Duration.zero);
}
test('the owner decision is raised and nothing is removed', () async {
await twoShortSyncs();
group('still short after three full syncs', () {
test('stops at three, checks the rest by key, removes nothing', () async {
await threeShortSyncs();
expect(contactRequests(), 3, reason: 'owner cap: 3 per connection');
expect(byKeyRequests(), 2, reason: 'Bravo and Charlie, one each');
expect(connector.contactSyncDecisionPending, isTrue);
expect(connector.contactSyncUndeliveredCount, 2);
expect(names(), {'Alpha', 'Bravo', 'Charlie'});
expect(contactRequests(), 2, reason: 'retry is bounded to one');
final shortfall = connector.contactSyncShortfall!;
expect(shortfall.declared, 350);
expect(shortfall.received, 1);
expect(shortfall.keptLocally, 2);
});
test('keep my contacts leaves the list as is', () async {
await twoShortSyncs();
await threeShortSyncs();
connector.resolveContactSyncKeepLocal();
expect(connector.contactSyncDecisionPending, isFalse);
expect(names(), {'Alpha', 'Bravo', 'Charlie'});
});
test('use the radio list drops what it did not send', () async {
await twoShortSyncs();
test('use the radio list drops what no sync delivered', () async {
await threeShortSyncs();
await connector.resolveContactSyncUseRadio();
expect(connector.contactSyncDecisionPending, isFalse);
@ -203,6 +230,148 @@ void main() {
});
});
group('forced resync (#703)', () {
test('the cap and pacing defaults are pinned', () {
expect(MeshCoreConnector.contactSyncMaxFullStreams, 3);
expect(
MeshCoreConnector.defaultContactRecoverySettle,
const Duration(seconds: 10),
);
expect(
MeshCoreConnector.defaultContactKeyCheckGap,
const Duration(milliseconds: 250),
);
expect(
MeshCoreConnector().contactKeyCheckTimeout,
const Duration(seconds: 5),
);
});
test(
'syncs that each miss different contacts add up to complete',
() async {
seedSaved();
await connector.getContacts();
await answerStream(3, [
_contactFrame(0x11, 'Alpha'),
_contactFrame(0x22, 'Bravo'),
]);
await answerStream(3, [
_contactFrame(0x22, 'Bravo'),
_contactFrame(0x33, 'Charlie'),
]);
await settle();
expect(contactRequests(), 2, reason: 'no third sync once complete');
expect(byKeyRequests(), 0);
expect(connector.contactSyncShortfall, isNull);
expect(connector.contactSyncDecisionPending, isFalse);
expect(names(), {'Alpha', 'Bravo', 'Charlie'});
},
);
test('only a contact the radio says is gone is removed', () async {
seedSaved();
connector.contactsForTest.add(_saved(0x44, 'Delta'));
await connector.getContacts();
await answerStream(3, [
_contactFrame(0x11, 'Alpha'),
_contactFrame(0x22, 'Bravo'),
]);
await answerStream(3, [
_contactFrame(0x11, 'Alpha'),
_contactFrame(0x33, 'Charlie'),
]);
expect(byKeyRequests(), 1, reason: 'Delta, the only one missing');
connector.handleFrameForTest(
Uint8List.fromList([respCodeErr, errCodeNotFound]),
);
await settle();
expect(names(), {'Alpha', 'Bravo', 'Charlie'});
expect(connector.contactSyncShortfall, isNull);
});
test(
'a contact with no answer is kept even when the rest add up',
() async {
seedSaved();
connector.contactsForTest.add(_saved(0x44, 'Delta'));
await connector.getContacts();
await answerStream(3, [
_contactFrame(0x11, 'Alpha'),
_contactFrame(0x22, 'Bravo'),
]);
await answerStream(3, [_contactFrame(0x33, 'Charlie')]);
await Future<void>.delayed(const Duration(milliseconds: 100));
expect(names(), {'Alpha', 'Bravo', 'Charlie', 'Delta'});
},
);
test('a by-key answer counts toward complete', () async {
seedSaved();
await connector.getContacts();
final alpha = [_contactFrame(0x11, 'Alpha')];
await answerStream(3, alpha);
await answerStream(3, alpha);
await answerStream(3, alpha);
connector.handleFrameForTest(_contactFrame(0x22, 'Bravo'));
await settle();
connector.handleFrameForTest(_contactFrame(0x33, 'Charlie'));
await settle();
expect(connector.contactSyncShortfall, isNull);
expect(connector.contactSyncDecisionPending, isFalse);
expect(names(), {'Alpha', 'Bravo', 'Charlie'});
});
test('a disconnect mid-recovery stops it and removes nothing', () async {
seedSaved();
connector.contactRecoverySettle = const Duration(milliseconds: 30);
await fullSync(declared: 350, streamed: [_contactFrame(0x11, 'Alpha')]);
connector.resetConnectionHandshakeStateForTest();
await Future<void>.delayed(const Duration(milliseconds: 80));
expect(contactRequests(), 1, reason: 'no recovery after disconnect');
expect(byKeyRequests(), 0);
expect(names(), {'Alpha', 'Bravo', 'Charlie'});
});
test('a new connection starts with no recovery state', () async {
seedSaved();
await connector.getContacts();
await answerStream(3, [
_contactFrame(0x11, 'Alpha'),
_contactFrame(0x22, 'Bravo'),
]);
connector.resetConnectionHandshakeStateForTest();
await connector.getContacts();
await answerStream(3, [_contactFrame(0x11, 'Alpha')]);
expect(
connector.contactSyncUndeliveredCount,
2,
reason: 'Bravo from the old connection must not count',
);
});
test('recovery waits for the radio to go quiet', () async {
var now = DateTime.utc(2026, 10, 2, 6);
connector.contactSyncClock = () => now;
connector.contactRecoverySettle = const Duration(milliseconds: 10);
seedSaved();
await fullSync(declared: 350, streamed: [_contactFrame(0x11, 'Alpha')]);
await Future<void>.delayed(const Duration(milliseconds: 50));
expect(contactRequests(), 1, reason: 'radio not quiet yet');
now = now.add(const Duration(seconds: 1));
await Future<void>.delayed(const Duration(milliseconds: 1100));
expect(contactRequests(), 2);
});
});
test('an advert heard mid-sync survives a complete sync', () async {
seedSaved();
await connector.getContacts();

Write Preview
Loading…
Cancel
Save

Powered by TurnKey Linux.