From bcbcf7c109c55725c1ad757529040cdc28cf64c9 Mon Sep 17 00:00:00 2001 From: jm Date: Wed, 14 Oct 2026 19:10:17 +0000 Subject: [PATCH] Advance relay replay cursor after delivery Signed-off-by: jm Co-authored-by: Codex --- mobile/lib/shared/relay/relay_session.dart | 14 ++- .../relay/relay_session_event_admission.dart | 7 +- .../test/shared/relay/relay_session_test.dart | 102 +++++++++++++----- 3 files changed, 90 insertions(+), 33 deletions(-) diff --git a/mobile/lib/shared/relay/relay_session.dart b/mobile/lib/shared/relay/relay_session.dart index 0bccbb3c37..3867a20880 100644 --- a/mobile/lib/shared/relay/relay_session.dart +++ b/mobile/lib/shared/relay/relay_session.dart @@ -412,6 +412,9 @@ class RelaySessionNotifier extends Notifier { @visibleForTesting int get debugPendingRelayAdmissions => _pendingRelayAdmissions; + @visibleForTesting + int get debugBufferedEventCount => _eventBuffer.length; + @visibleForTesting void debugSupersedeConnection() { _connectionGeneration++; @@ -888,9 +891,18 @@ class RelaySessionNotifier extends Notifier { if (_recentDeliveryKeys.length >= _maxRecentDeliveryKeys) { _recentDeliveryKeys.clear(); } - _recentDeliveryKeys.add(deliveryKey); sub.onEvent(buffered.event); + // A callback that throws has not durably consumed the event, so do not + // let it move the reconnect cursor past data the consumer may still need. + // A callback may also unsubscribe synchronously; that retired owner no + // longer needs dedup or replay state. + if (_liveSubscriptions[buffered.subId] != sub) continue; + _recentDeliveryKeys.add(deliveryKey); + if (sub.lastSeenCreatedAt == null || + buffered.event.createdAt > sub.lastSeenCreatedAt!) { + sub.lastSeenCreatedAt = buffered.event.createdAt; + } } } diff --git a/mobile/lib/shared/relay/relay_session_event_admission.dart b/mobile/lib/shared/relay/relay_session_event_admission.dart index 9942344834..6442f42fda 100644 --- a/mobile/lib/shared/relay/relay_session_event_admission.dart +++ b/mobile/lib/shared/relay/relay_session_event_admission.dart @@ -91,13 +91,10 @@ extension _RelaySessionEventAdmission on RelaySessionNotifier { liveSub.pendingVerificationBytes -= encodedBytes; if (event == null) return; - // Only a verified event can prove activity or advance replay state. + // Only a verified event can prove relay activity. Replay progress is + // recorded later, after the event reaches the subscription callback. _resetClosedRetry(liveSub); liveSub.validEventFrames++; - if (liveSub.lastSeenCreatedAt == null || - event.createdAt > liveSub.lastSeenCreatedAt!) { - liveSub.lastSeenCreatedAt = event.createdAt; - } _eventBuffer.add(_BufferedEvent(subId, event)); _scheduleFlush(); }); diff --git a/mobile/test/shared/relay/relay_session_test.dart b/mobile/test/shared/relay/relay_session_test.dart index 55c4bf3580..e9e7b66f75 100644 --- a/mobile/test/shared/relay/relay_session_test.dart +++ b/mobile/test/shared/relay/relay_session_test.dart @@ -1326,35 +1326,83 @@ void main() { ); } - test('retryable CLOSED drops an already verified buffered event', () async { - final verification = Completer?>>(); - final timers = <_ManualTimer>[]; - final socket = _RecordingRelaySocket(); - final delivered = []; - final session = RelaySessionNotifier( - verifyEventBatch: (_) => verification.future, - retryTimerFactory: (duration, callback) { - final timer = _ManualTimer(duration, callback); - timers.add(timer); - return timer; - }, - ); - session.debugAttachSocketForTest(socket); - final subscribe = session.subscribe(_channelFilter, delivered.add); - session.debugHandleMessage(['EOSE', 'l-1']); - final unsubscribe = await subscribe; - final stale = _signedEvent(createdAt: 30); + test( + 'retryable CLOSED replays all dropped verified buffered events', + () async { + final heldVerification = Completer?>>(); + var verificationCalls = 0; + final timers = <_ManualTimer>[]; + final socket = _RecordingRelaySocket(); + final delivered = []; + final session = RelaySessionNotifier( + verifyEventBatch: (events) { + verificationCalls++; + if (verificationCalls == 2) return heldVerification.future; + return Future.value([ + for (final event in events) event as Map, + ]); + }, + retryTimerFactory: (duration, callback) { + final timer = _ManualTimer(duration, callback); + timers.add(timer); + return timer; + }, + ); + session.debugAttachSocketForTest(socket); + final subscribe = session.subscribe(_channelFilter, delivered.add); + session.debugHandleMessage(['EOSE', 'l-1']); + final unsubscribe = await subscribe; + final alreadyDelivered = _signedEvent( + createdAt: 10, + content: 'already delivered', + ); + final droppedEarlier = _signedEvent( + createdAt: 20, + content: 'dropped earlier', + ); + final droppedLater = _signedEvent( + createdAt: 100, + content: 'dropped later', + ); - session.debugHandleMessage(['EVENT', 'l-1', stale.toJson()]); - verification.complete([stale.toJson()]); - await _waitUntil(() => session.debugPendingRelayAdmissions == 0); - session.debugHandleMessage(['CLOSED', 'l-1', 'error: transient']); - await Future.delayed(const Duration(milliseconds: 30)); + session.debugHandleMessage(['EVENT', 'l-1', alreadyDelivered.toJson()]); + await _waitUntil(() => delivered.length == 1); - expect(delivered, isEmpty); - expect(timers.single.isActive, isTrue); - unsubscribe(); - }); + session.debugHandleMessage(['EVENT', 'l-1', droppedEarlier.toJson()]); + session.debugHandleMessage(['EVENT', 'l-1', droppedLater.toJson()]); + await _waitUntil(() => verificationCalls == 2); + heldVerification.complete([ + droppedEarlier.toJson(), + droppedLater.toJson(), + ]); + await _waitUntil(() => session.debugBufferedEventCount == 2); + + session.debugHandleMessage(['CLOSED', 'l-1', 'error: transient']); + expect(delivered, [alreadyDelivered]); + expect(session.debugBufferedEventCount, 0); + expect(timers.single.isActive, isTrue); + + timers.single.fire(); + await Future.delayed(Duration.zero); + final firstReplayReq = _reqs(socket).where((req) => req[1] == 'l-1').last; + expect(firstReplayReq[2]['since'], 5); + + session.debugHandleMessage(['EVENT', 'l-1', droppedEarlier.toJson()]); + session.debugHandleMessage(['EVENT', 'l-1', droppedLater.toJson()]); + session.debugHandleMessage(['EOSE', 'l-1']); + await _waitUntil(() => delivered.length == 3); + expect(delivered, [alreadyDelivered, droppedEarlier, droppedLater]); + + session.debugHandleMessage(['CLOSED', 'l-1', 'error: transient']); + timers.last.fire(); + await Future.delayed(Duration.zero); + final secondReplayReq = _reqs( + socket, + ).where((req) => req[1] == 'l-1').last; + expect(secondReplayReq[2]['since'], 95); + unsubscribe(); + }, + ); test('local OK acknowledgment bypasses pending verification', () async { final verification = Completer?>>();