jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-27 12:40:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

Advance relay replay cursor after delivery

Browse Source 
Signed-off-by: jm <jm@squareup.com>
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
jm
2026-10-14 19:10:17 +00:00
parent 47c0ff5eed
commit bcbcf7c109
3 changed files with 90 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
mobile/lib/shared/relay/relay_session.dart
View File

@@ -412,6 +412,9 @@ class RelaySessionNotifier extends Notifier<SessionState> {
@visibleForTesting
int get debugPendingRelayAdmissions => _pendingRelayAdmissions;
@visibleForTesting
int get debugBufferedEventCount => _eventBuffer.length;
@visibleForTesting
void debugSupersedeConnection() {
_connectionGeneration++;
@@ -888,9 +891,18 @@ class RelaySessionNotifier extends Notifier<SessionState> {
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;
}
}
}

7
mobile/lib/shared/relay/relay_session_event_admission.dart
View File

@@ -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();
});

102
mobile/test/shared/relay/relay_session_test.dart
View File

@@ -1326,35 +1326,83 @@ void main() {
);
}
test('retryable CLOSED drops an already verified buffered event', () async {
final verification = Completer<List<Map<String, dynamic>?>>();
final timers = <_ManualTimer>[];
final socket = _RecordingRelaySocket();
final delivered = <NostrEvent>[];
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<List<Map<String, dynamic>?>>();
var verificationCalls = 0;
final timers = <_ManualTimer>[];
final socket = _RecordingRelaySocket();
final delivered = <NostrEvent>[];
final session = RelaySessionNotifier(
verifyEventBatch: (events) {
verificationCalls++;
if (verificationCalls == 2) return heldVerification.future;
return Future.value([
for (final event in events) event as Map<String, dynamic>,
]);
},
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<void>.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<void>.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<void>.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<List<Map<String, dynamic>?>>();