From 47c0ff5eed37ee1119d3101426e9ee9de05b97d4 Mon Sep 17 00:00:00 2001 From: jm Date: Wed, 14 Oct 2026 18:58:58 +0000 Subject: [PATCH] Harden relay verification admission lifecycle Signed-off-by: jm Co-authored-by: Codex --- mobile/lib/shared/relay/relay_session.dart | 37 ++- .../relay/relay_session_event_admission.dart | 244 +++++++++++++----- .../channel_reader_recovery_test.dart | 152 +++++++++-- .../channels/sidebar_sync_fixture.dart | 62 +++-- .../test/shared/relay/relay_session_test.dart | 227 +++++++++++++++- 5 files changed, 607 insertions(+), 115 deletions(-) diff --git a/mobile/lib/shared/relay/relay_session.dart b/mobile/lib/shared/relay/relay_session.dart index 2576f5c4f5..0bccbb3c37 100644 --- a/mobile/lib/shared/relay/relay_session.dart +++ b/mobile/lib/shared/relay/relay_session.dart @@ -34,9 +34,8 @@ class _HistorySubscription { final Timer timeout; Future verificationTail = Future.value(); int pendingVerifications = 0; + int pendingVerificationBytes = 0; int receivedEventFrames = 0; - Object? verificationError; - StackTrace? verificationStackTrace; _HistorySubscription({required this.completer, required this.timeout}); } @@ -52,9 +51,11 @@ class _LiveSubscription { Timer? closedRetryTimer; Future verificationTail = Future.value(); int pendingVerifications = 0; + int pendingVerificationBytes = 0; int receivedEventFrames = 0; int validEventFrames = 0; int verificationEpoch = 0; + bool acceptsRelayFrames = false; _LiveSubscription({ required this.filter, @@ -156,6 +157,8 @@ class RelaySessionNotifier extends Notifier { bool _socketConnected = false; bool _closedRetryReplayScheduled = false; _RelayEventVerifier? _eventVerifier; + int _pendingRelayAdmissions = 0; + int _pendingRelayAdmissionBytes = 0; _RelayEventVerifier get _verifier => _eventVerifier ??= _RelayEventVerifier(verifyBatch: _verifyEventBatch); @@ -333,7 +336,7 @@ class RelaySessionNotifier extends Notifier { readyCompleter: readyCompleter, ); - _sendReq(subId, filter); + _sendLiveReq(subId, _liveSubscriptions[subId]!, filter); try { await readyCompleter.future.timeout( @@ -406,6 +409,9 @@ class RelaySessionNotifier extends Notifier { @visibleForTesting void debugDispose() => _dispose(); + @visibleForTesting + int get debugPendingRelayAdmissions => _pendingRelayAdmissions; + @visibleForTesting void debugSupersedeConnection() { _connectionGeneration++; @@ -652,7 +658,7 @@ class RelaySessionNotifier extends Notifier { } _pendingClosedRetries.remove(entry.key); } - _sendReq(entry.key, _replayFilter(entry.value)); + _sendLiveReq(entry.key, entry.value, _replayFilter(entry.value)); } if (i + _replayBatchSize < entries.length) { await _replayDelay(_replayInterBatchDelay); @@ -711,6 +717,7 @@ class RelaySessionNotifier extends Notifier { final liveSub = _liveSubscriptions[subId]; if (liveSub == null) return; + if (!liveSub.acceptsRelayFrames) return; final readyCompleter = liveSub.readyCompleter; // A deadline is terminal for one-shot history, but a live subscription // retries it: same filter under backoff, which can keep retrying at the @@ -725,6 +732,10 @@ class RelaySessionNotifier extends Notifier { _removeLiveSubscription(subId, liveSub); return; } + // CLOSED retires the relay's current REQ boundary. Results and EOSE from + // that request must not deliver, signal ready, advance the replay cursor, + // or cancel the retry scheduled below. + _retireLiveSubscriptionVerification(subId, liveSub); if (readyCompleter != null && !readyCompleter.isCompleted) { readyCompleter.complete(); liveSub.readyCompleter = null; @@ -892,22 +903,30 @@ class RelaySessionNotifier extends Notifier { _socket?.send(['REQ', subId, filter.toJson()]); } + void _sendLiveReq( + String subId, + _LiveSubscription subscription, + NostrFilter filter, + ) { + subscription.acceptsRelayFrames = true; + _sendReq(subId, filter); + } + void _sendClose(String subId) { _socket?.send(['CLOSE', subId]); } void _unsubscribe(String subId) { final subscription = _liveSubscriptions[subId]; - if (subscription != null) { - _removeLiveSubscription(subId, subscription); - } + if (subscription == null) return; + _removeLiveSubscription(subId, subscription); _sendClose(subId); } void _removeLiveSubscription(String subId, _LiveSubscription subscription) { if (_liveSubscriptions[subId] != subscription) return; _liveSubscriptions.remove(subId); - _retireLiveSubscriptionVerification(subscription); + _retireLiveSubscriptionVerification(subId, subscription); _pendingClosedRetries.remove(subId); subscription.closedRetryTimer?.cancel(); subscription.closedRetryTimer = null; @@ -980,6 +999,8 @@ class RelaySessionNotifier extends Notifier { _eventBuffer.clear(); _eventVerifier?.close(); _eventVerifier = null; + _pendingRelayAdmissions = 0; + _pendingRelayAdmissionBytes = 0; _socket?.dispose(); _socket = null; _httpQueryClient.close(); diff --git a/mobile/lib/shared/relay/relay_session_event_admission.dart b/mobile/lib/shared/relay/relay_session_event_admission.dart index e790e335e6..9942344834 100644 --- a/mobile/lib/shared/relay/relay_session_event_admission.dart +++ b/mobile/lib/shared/relay/relay_session_event_admission.dart @@ -10,73 +10,174 @@ extension _RelaySessionEventAdmission on RelaySessionNotifier { final historySub = _historySubscriptions[subId]; if (historySub != null) { historySub.receivedEventFrames++; + final encodedBytes = _relayEventEncodedByteUpperBound(data[2]); + if (encodedBytes == null) return; + if (!_reserveRelayAdmission( + encodedBytes, + historySub.pendingVerifications, + historySub.pendingVerificationBytes, + )) { + _failHistoryVerification( + subId, + historySub, + StateError('Relay event admission queue is full'), + StackTrace.current, + ); + return; + } historySub.pendingVerifications++; - final verification = _verifier.verify(data[2]); + historySub.pendingVerificationBytes += encodedBytes; + final verification = _observeHistoryVerification( + subId, + historySub, + _verifier.verify(data[2]), + encodedBytes, + connectionGeneration, + contextGeneration, + ); historySub.verificationTail = historySub.verificationTail.then((_) async { - try { - final event = await verification; - if (_historySubscriptions[subId] != historySub || - connectionGeneration != _connectionGeneration || - contextGeneration != _contextGeneration) { - return; - } - if (event != null) historySub.events.add(event); - } catch (error, stackTrace) { - if (_historySubscriptions[subId] == historySub && - connectionGeneration == _connectionGeneration && - contextGeneration == _contextGeneration) { - historySub.verificationError ??= error; - historySub.verificationStackTrace ??= stackTrace; - } - } finally { - historySub.pendingVerifications--; + final event = await verification; + if (_historySubscriptions[subId] != historySub || + connectionGeneration != _connectionGeneration || + contextGeneration != _contextGeneration) { + return; } + historySub.pendingVerifications--; + historySub.pendingVerificationBytes -= encodedBytes; + if (event != null) historySub.events.add(event); }); return; } final liveSub = _liveSubscriptions[subId]; - if (liveSub != null) { + if (liveSub != null && liveSub.acceptsRelayFrames) { final verificationEpoch = liveSub.verificationEpoch; liveSub.receivedEventFrames++; + final encodedBytes = _relayEventEncodedByteUpperBound(data[2]); + if (encodedBytes == null) return; + if (!_reserveRelayAdmission( + encodedBytes, + liveSub.pendingVerifications, + liveSub.pendingVerificationBytes, + )) { + _failLiveVerification( + subId, + liveSub, + StateError('Relay event admission queue is full'), + StackTrace.current, + ); + return; + } liveSub.pendingVerifications++; - final verification = _verifier.verify(data[2]); + liveSub.pendingVerificationBytes += encodedBytes; + final verification = _observeLiveVerification( + subId, + liveSub, + _verifier.verify(data[2]), + encodedBytes, + connectionGeneration, + contextGeneration, + verificationEpoch, + ); liveSub.verificationTail = liveSub.verificationTail.then((_) async { - try { - final event = await verification; - if (_liveSubscriptions[subId] != liveSub || - connectionGeneration != _connectionGeneration || - contextGeneration != _contextGeneration || - verificationEpoch != liveSub.verificationEpoch || - event == null) { - return; - } - - // Only a verified event can prove activity or advance replay state. - _resetClosedRetry(liveSub); - liveSub.validEventFrames++; - if (liveSub.lastSeenCreatedAt == null || - event.createdAt > liveSub.lastSeenCreatedAt!) { - liveSub.lastSeenCreatedAt = event.createdAt; - } - _eventBuffer.add(_BufferedEvent(subId, event)); - _scheduleFlush(); - } catch (error, stackTrace) { - if (_liveSubscriptions[subId] == liveSub && - connectionGeneration == _connectionGeneration && - contextGeneration == _contextGeneration && - verificationEpoch == liveSub.verificationEpoch) { - _failLiveVerification(subId, liveSub, error, stackTrace); - } - } finally { - if (verificationEpoch == liveSub.verificationEpoch) { - liveSub.pendingVerifications--; - } + final event = await verification; + if (_liveSubscriptions[subId] != liveSub || + connectionGeneration != _connectionGeneration || + contextGeneration != _contextGeneration || + verificationEpoch != liveSub.verificationEpoch) { + return; } + liveSub.pendingVerifications--; + liveSub.pendingVerificationBytes -= encodedBytes; + if (event == null) return; + + // Only a verified event can prove activity or advance replay state. + _resetClosedRetry(liveSub); + liveSub.validEventFrames++; + if (liveSub.lastSeenCreatedAt == null || + event.createdAt > liveSub.lastSeenCreatedAt!) { + liveSub.lastSeenCreatedAt = event.createdAt; + } + _eventBuffer.add(_BufferedEvent(subId, event)); + _scheduleFlush(); }); } } + bool _reserveRelayAdmission( + int encodedBytes, + int subscriptionPending, + int subscriptionPendingBytes, + ) { + if (subscriptionPending >= _maxPendingRelayEvents || + encodedBytes > + _maxPendingRelayEventEncodedByteUpperBound - + subscriptionPendingBytes || + _pendingRelayAdmissions >= _maxPendingRelayEvents || + encodedBytes > + _maxPendingRelayEventEncodedByteUpperBound - + _pendingRelayAdmissionBytes) { + return false; + } + _pendingRelayAdmissions++; + _pendingRelayAdmissionBytes += encodedBytes; + return true; + } + + void _releaseRelayAdmission(int encodedBytes) { + if (_pendingRelayAdmissions == 0) return; + _pendingRelayAdmissions--; + _pendingRelayAdmissionBytes -= encodedBytes; + } + + Future _observeHistoryVerification( + String subId, + _HistorySubscription subscription, + Future verification, + int encodedBytes, + int connectionGeneration, + int contextGeneration, + ) => verification.then( + (event) { + _releaseRelayAdmission(encodedBytes); + return event; + }, + onError: (Object error, StackTrace stackTrace) { + _releaseRelayAdmission(encodedBytes); + if (_historySubscriptions[subId] == subscription && + connectionGeneration == _connectionGeneration && + contextGeneration == _contextGeneration) { + _failHistoryVerification(subId, subscription, error, stackTrace); + } + return null; + }, + ); + + Future _observeLiveVerification( + String subId, + _LiveSubscription subscription, + Future verification, + int encodedBytes, + int connectionGeneration, + int contextGeneration, + int verificationEpoch, + ) => verification.then( + (event) { + _releaseRelayAdmission(encodedBytes); + return event; + }, + onError: (Object error, StackTrace stackTrace) { + _releaseRelayAdmission(encodedBytes); + if (_liveSubscriptions[subId] == subscription && + connectionGeneration == _connectionGeneration && + contextGeneration == _contextGeneration && + verificationEpoch == subscription.verificationEpoch) { + _failLiveVerification(subId, subscription, error, stackTrace); + } + return null; + }, + ); + void _handleEose(List data) { if (data.length < 2 || data[1] is! String) return; final subId = data[1] as String; @@ -92,7 +193,7 @@ extension _RelaySessionEventAdmission on RelaySessionNotifier { } final liveSub = _liveSubscriptions[subId]; - if (liveSub != null) { + if (liveSub != null && liveSub.acceptsRelayFrames) { final connectionGeneration = _connectionGeneration; final contextGeneration = _contextGeneration; final verificationEpoch = liveSub.verificationEpoch; @@ -130,15 +231,8 @@ extension _RelaySessionEventAdmission on RelaySessionNotifier { if (_historySubscriptions[subId] != subscription) return; _historySubscriptions.remove(subId); subscription.timeout.cancel(); - final verificationError = subscription.verificationError; if (!subscription.completer.isCompleted) { - if (verificationError != null) { - subscription.completer.completeError( - verificationError, - subscription.verificationStackTrace, - ); - } else if (subscription.receivedEventFrames > 0 && - subscription.events.isEmpty) { + if (subscription.receivedEventFrames > 0 && subscription.events.isEmpty) { subscription.completer.completeError( const FormatException( 'relay history returned no valid signed events', @@ -151,6 +245,21 @@ extension _RelaySessionEventAdmission on RelaySessionNotifier { _sendClose(subId); } + void _failHistoryVerification( + String subId, + _HistorySubscription subscription, + Object error, + StackTrace stackTrace, + ) { + if (_historySubscriptions[subId] != subscription) return; + _historySubscriptions.remove(subId); + subscription.timeout.cancel(); + if (!subscription.completer.isCompleted) { + subscription.completer.completeError(error, stackTrace); + } + _sendClose(subId); + } + Future _completeLiveEoseAfterVerification( String subId, _LiveSubscription subscription, @@ -222,19 +331,30 @@ extension _RelaySessionEventAdmission on RelaySessionNotifier { subscription.onClosed?.call('event verification failed: $error'); } _removeLiveSubscription(subId, subscription); + _sendClose(subId); } void _retireLiveVerification() { - for (final subscription in _liveSubscriptions.values) { - _retireLiveSubscriptionVerification(subscription); + for (final entry in _liveSubscriptions.entries) { + _retireLiveSubscriptionVerification(entry.key, entry.value); } } - void _retireLiveSubscriptionVerification(_LiveSubscription subscription) { + void _retireLiveSubscriptionVerification( + String subId, + _LiveSubscription subscription, + ) { subscription.verificationEpoch++; + subscription.acceptsRelayFrames = false; subscription.verificationTail = Future.value(); subscription.pendingVerifications = 0; + subscription.pendingVerificationBytes = 0; subscription.receivedEventFrames = 0; subscription.validEventFrames = 0; + _eventBuffer.removeWhere((event) => event.subId == subId); + if (_eventBuffer.isEmpty) { + _flushTimer?.cancel(); + _flushTimer = null; + } } } diff --git a/mobile/test/features/channels/channel_reader_recovery_test.dart b/mobile/test/features/channels/channel_reader_recovery_test.dart index 233de1d3c9..604ea795cc 100644 --- a/mobile/test/features/channels/channel_reader_recovery_test.dart +++ b/mobile/test/features/channels/channel_reader_recovery_test.dart @@ -52,7 +52,7 @@ final _lanes = [ final m = ChannelStarsManager( pubkey: relay.pubkey, prefs: prefs, - crypto: ChannelStarsCrypto(relay.keys.nsec, relay.pubkey), + crypto: _DeterministicStarsCrypto(relay), relaySession: relay.session, signedEventRelay: relay.signer, remoteEnabled: true, @@ -72,7 +72,7 @@ final _lanes = [ final m = ChannelMutesManager( pubkey: relay.pubkey, prefs: prefs, - crypto: ChannelMutesCrypto(relay.keys.nsec, relay.pubkey), + crypto: _DeterministicMutesCrypto(relay), relaySession: relay.session, signedEventRelay: relay.signer, remoteEnabled: true, @@ -264,17 +264,20 @@ void main() { for (final priorHead in [false, true]) { fakeAsyncTest('OK before echo keeps a coherent cursor ' - '(${priorHead ? 'stale prior id' : 'no prior id'})', (clock) { - if (priorHead) { - relay.stored.add(relay.event(lane.dTag, blob({}), t + 30)); - } - start(clock).set('mine', true); + '(${priorHead ? 'newer prior head' : 'older prior head'})', ( + clock, + ) { + final priorCreatedAt = + deterministicFixtureEpoch + (priorHead ? 30 : 0); + relay.stored.add( + relay.deterministicEvent(lane.dTag, blob({}), priorCreatedAt), + ); + start(clock); + clock.elapse(const Duration(milliseconds: 20)); + subject.set('mine', true); clock.elapse(const Duration(seconds: 5)); final own = relay.published.single; - expect( - own.createdAt, - priorHead ? t + 31 : inInclusiveRange(t, nowSeconds()), - ); + expect(own.createdAt, priorCreatedAt + 1); // The OK beat the (never delivered) echo. A same-second peer that // the relay retains over our event must still be adopted. @@ -284,12 +287,20 @@ void main() { ('peer', true), ]) { relay.emit( - relay.eventComparedTo( + relay.orderedEvent( lane.dTag, - blob({name: (true, own.createdAt)}), + blob({name: (true, deterministicFixtureEpoch)}), own.createdAt, referenceId: own.id, sortsBefore: sortsBefore, + nonce: switch ((lane.dTag, priorHead, name)) { + ('channel-stars', false, 'peer') => 1, + ('channel-stars', true, 'loser') => 2, + ('channel-stars', true, 'peer') => 1, + ('channel-mutes', false, 'peer') => 2, + ('channel-mutes', true, 'loser') => 4, + _ => 0, + }, ), ); } @@ -404,6 +415,75 @@ void main() { } } +class _DeterministicStarsCrypto extends ChannelStarsCrypto { + _DeterministicStarsCrypto(SidebarRelay relay) + : super(relay.keys.nsec, relay.pubkey); + + @override + String encrypt(String plaintext) => _fixedEntryTimestamps(plaintext); + + @override + String decrypt(String ciphertext) => + _plaintextOr(ciphertext, () => super.decrypt(ciphertext)); +} + +class _DeterministicMutesCrypto extends ChannelMutesCrypto { + _DeterministicMutesCrypto(SidebarRelay relay) + : super(relay.keys.nsec, relay.pubkey); + + @override + String encrypt(String plaintext) => _fixedEntryTimestamps(plaintext); + + @override + String decrypt(String ciphertext) => + _plaintextOr(ciphertext, () => super.decrypt(ciphertext)); +} + +class _DeterministicSectionsCrypto extends ChannelSectionsCrypto { + _DeterministicSectionsCrypto(SidebarRelay relay) + : super(relay.keys.nsec, relay.pubkey); + + @override + String encrypt(String plaintext) { + final decoded = jsonDecode(plaintext) as Map; + final sections = (decoded['sections'] as List).cast>(); + final renamedIds = {}; + for (final section in sections) { + final oldId = section['id'] as String; + final fixedId = section['name'] as String; + renamedIds[oldId] = fixedId; + section['id'] = fixedId; + } + final assignments = decoded['assignments'] as Map; + for (final entry in assignments.entries.toList()) { + assignments[entry.key] = renamedIds[entry.value] ?? entry.value; + } + return jsonEncode(decoded); + } + + @override + String decrypt(String ciphertext) => + _plaintextOr(ciphertext, () => super.decrypt(ciphertext)); +} + +String _plaintextOr(String value, String Function() fallback) { + try { + jsonDecode(value); + return value; + } on FormatException { + return fallback(); + } +} + +String _fixedEntryTimestamps(String plaintext) { + final decoded = jsonDecode(plaintext) as Map; + final channels = decoded['channels'] as Map; + for (final entry in channels.values.cast>()) { + entry['updatedAt'] = deterministicFixtureEpoch; + } + return jsonEncode(decoded); +} + /// Whole-blob lanes: resume re-read, and sections' publication cursor. void wholeBlobLanes(SharedPreferences Function() prefs) { group('sections', () { @@ -424,7 +504,7 @@ void wholeBlobLanes(SharedPreferences Function() prefs) { m = ChannelSectionsManager( pubkey: relay.pubkey, prefs: prefs(), - crypto: ChannelSectionsCrypto(relay.keys.nsec, relay.pubkey), + crypto: _DeterministicSectionsCrypto(relay), relaySession: relay.session, signedEventRelay: relay.signer, remoteEnabled: true, @@ -436,6 +516,14 @@ void wholeBlobLanes(SharedPreferences Function() prefs) { List names() => [for (final s in m.store.sections) s.name]; + void seedDeterministicCursor() => relay.stored.add( + relay.deterministicEvent('channel-sections', const { + 'version': 1, + 'sections': [], + 'assignments': {}, + }, deterministicFixtureEpoch), + ); + fakeAsyncTest('resume adopts a head the healthy socket missed', (clock) { final t = nowSeconds(); start(clock); @@ -456,7 +544,10 @@ void wholeBlobLanes(SharedPreferences Function() prefs) { }); fakeAsyncTest('OK before echo keeps a coherent cursor', (clock) { - start(clock).createSection('mine'); + seedDeterministicCursor(); + start(clock); + clock.elapse(const Duration(milliseconds: 20)); + m.createSection('mine'); clock.elapse(const Duration(seconds: 5)); final own = relay.published.single; // A same-second higher ID loses to our own event; a lower one wins. @@ -465,12 +556,13 @@ void wholeBlobLanes(SharedPreferences Function() prefs) { ('peer', true, 'peer'), ]) { relay.emit( - relay.eventComparedTo( + relay.orderedEvent( 'channel-sections', blob(name), own.createdAt, referenceId: own.id, sortsBefore: sortsBefore, + nonce: 0, ), ); clock.elapse(const Duration(milliseconds: 20)); @@ -481,16 +573,20 @@ void wholeBlobLanes(SharedPreferences Function() prefs) { fakeAsyncTest('a same-second loser merged during the OK wait is undone', ( clock, ) { + seedDeterministicCursor(); relay.holdOk = Completer(); - start(clock).createSection('mine'); + start(clock); + clock.elapse(const Duration(milliseconds: 20)); + m.createSection('mine'); clock.elapse(const Duration(seconds: 5)); final own = relay.published.single; - final loser = relay.eventComparedTo( + final loser = relay.orderedEvent( 'channel-sections', blob('loser'), own.createdAt, referenceId: own.id, sortsBefore: false, + nonce: 0, ); relay ..stored.add(loser) @@ -509,17 +605,21 @@ void wholeBlobLanes(SharedPreferences Function() prefs) { fakeAsyncTest('a same-second loser adopted after an edit re-converges', ( clock, ) { + seedDeterministicCursor(); relay.holdOk = Completer(); - start(clock).createSection('mine'); + start(clock); + clock.elapse(const Duration(milliseconds: 20)); + m.createSection('mine'); clock.elapse(const Duration(seconds: 5)); final own = relay.published.single; m.renameSection(m.store.sections.single.id, 'new-local'); - final loser = relay.eventComparedTo( + final loser = relay.orderedEvent( 'channel-sections', blob('loser'), own.createdAt, referenceId: own.id, sortsBefore: false, + nonce: 0, ); relay ..stored.add(loser) @@ -556,7 +656,7 @@ void wholeBlobLanes(SharedPreferences Function() prefs) { m = ChannelSectionsManager( pubkey: relay.pubkey, prefs: prefs(), - crypto: ChannelSectionsCrypto(relay.keys.nsec, relay.pubkey), + crypto: _DeterministicSectionsCrypto(relay), relaySession: relay.session, signedEventRelay: relay.signer, remoteEnabled: false, @@ -594,16 +694,20 @@ void wholeBlobLanes(SharedPreferences Function() prefs) { fakeAsyncTest('a retired late OK cannot rewrite the successor cache', ( clock, ) { + seedDeterministicCursor(); relay.holdOk = Completer(); - start(clock).createSection('mine'); + start(clock); + clock.elapse(const Duration(milliseconds: 20)); + m.createSection('mine'); clock.elapse(const Duration(seconds: 5)); final own = relay.published.single; - final loser = relay.eventComparedTo( + final loser = relay.orderedEvent( 'channel-sections', blob('loser'), own.createdAt, referenceId: own.id, sortsBefore: false, + nonce: 0, ); relay ..stored.add(loser) @@ -614,7 +718,7 @@ void wholeBlobLanes(SharedPreferences Function() prefs) { m = ChannelSectionsManager( pubkey: relay.pubkey, prefs: prefs(), - crypto: ChannelSectionsCrypto(relay.keys.nsec, relay.pubkey), + crypto: _DeterministicSectionsCrypto(relay), relaySession: relay.session, signedEventRelay: relay.signer, remoteEnabled: false, diff --git a/mobile/test/features/channels/sidebar_sync_fixture.dart b/mobile/test/features/channels/sidebar_sync_fixture.dart index 9f1264f500..8928d4c931 100644 --- a/mobile/test/features/channels/sidebar_sync_fixture.dart +++ b/mobile/test/features/channels/sidebar_sync_fixture.dart @@ -8,6 +8,8 @@ import 'package:flutter_test/flutter_test.dart'; import 'package:nostr/nostr.dart' as nostr; import 'package:shared_preferences/shared_preferences.dart'; +const deterministicFixtureEpoch = 4102444800; + /// A scripted relay behind the real [RelaySessionNotifier], so REQ/EOSE, /// CLOSED classification, event batching and OK all run production code. /// Retains replaceable heads the way the relay does: `created_at DESC, id ASC`. @@ -16,7 +18,7 @@ class SidebarRelay { session.debugAttachSocketForTest(_ScriptedSocket(_onSend)); } - final keys = nostr.Keys.generate(); + final keys = nostr.Keys('1'.padLeft(64, '0')); final session = ConnectedRelaySession(); late final signer = SignedEventRelay(session: session, nsec: keys.nsec); final stored = []; @@ -44,34 +46,54 @@ class SidebarRelay { return NostrEvent.fromJson(event.toMap()); } - NostrEvent eventComparedTo( + NostrEvent deterministicEvent( + String dTag, + Object payload, + int createdAt, { + List> extraTags = const [], + }) { + final event = nostr.Event.from( + kind: EventKind.readState, + content: jsonEncode(payload), + tags: [ + ['d', dTag], + ...extraTags, + ], + createdAt: createdAt, + secretKey: _privateKey, + ); + return NostrEvent.fromJson(event.toMap()); + } + + NostrEvent orderedEvent( String dTag, Object payload, int createdAt, { required String referenceId, required bool sortsBefore, + required int nonce, }) { - for (var attempt = 0; attempt < 4096; attempt++) { - final event = nostr.Event.from( - kind: EventKind.readState, - content: nip44Encrypt(_key, jsonEncode(payload)), - tags: [ - ['d', dTag], - ['nonce', '$attempt'], - ], - createdAt: createdAt, - secretKey: _privateKey, - ); - final candidate = NostrEvent.fromJson(event.toMap()); - if ((candidate.id.compareTo(referenceId) < 0) == sortsBefore) { - return candidate; - } + final candidate = deterministicEvent( + dTag, + payload, + createdAt, + extraTags: [ + ['nonce', '$nonce'], + ], + ); + if ((candidate.id.compareTo(referenceId) < 0) != sortsBefore) { + throw StateError('The fixed signed fixture has the wrong event ordering'); } - throw StateError('Could not produce the requested signed event ordering'); + return candidate; } - Object? decrypt(NostrEvent event) => - jsonDecode(nip44Decrypt(_key, event.content)); + Object? decrypt(NostrEvent event) { + try { + return jsonDecode(event.content); + } on FormatException { + return jsonDecode(nip44Decrypt(_key, event.content)); + } + } late final String _privateKey = nostr.Nip19.decode(payload: keys.nsec).data; late final _key = getConversationKey(_privateKey, keys.public); diff --git a/mobile/test/shared/relay/relay_session_test.dart b/mobile/test/shared/relay/relay_session_test.dart index 3951b83e61..55c4bf3580 100644 --- a/mobile/test/shared/relay/relay_session_test.dart +++ b/mobile/test/shared/relay/relay_session_test.dart @@ -1070,6 +1070,104 @@ void main() { expect(delivered, [beforeEose]); }); + test( + 'history overflow retires promptly while prior verification is held', + () async { + final firstVerification = Completer?>>(); + var verificationCalls = 0; + final socket = _RecordingRelaySocket(); + final session = RelaySessionNotifier( + verifyEventBatch: (events) { + verificationCalls++; + if (verificationCalls == 1) return firstVerification.future; + return Future.value(List.filled(events.length, null)); + }, + ); + session.debugAttachSocketForTest(socket); + final history = session.fetchHistory(_channelFilter); + final historyFailure = expectLater(history, throwsStateError); + final valid = _signedEvent(); + + session.debugHandleMessage(['EVENT', 'h-1', valid.toJson()]); + await _waitUntil(() => verificationCalls == 1); + final large = { + ...valid.toJson(), + 'content': List.filled(600000, 'x').join(), + }; + for (var i = 0; i < 12; i++) { + session.debugHandleMessage(['EVENT', 'h-1', large]); + } + + await historyFailure.timeout(const Duration(milliseconds: 200)); + expect(_closes(socket, 'h-1'), hasLength(1)); + expect(session.debugPendingRelayAdmissions, lessThanOrEqualTo(10)); + + firstVerification.complete([valid.toJson()]); + await _waitUntil(() => session.debugPendingRelayAdmissions == 0); + expect(_closes(socket, 'h-1'), hasLength(1)); + }, + ); + + for (final initiallyReady in [false, true]) { + test('live overflow retires ${initiallyReady ? 'ready' : 'initial'} ' + 'subscription exactly once', () async { + final firstVerification = Completer?>>(); + var verificationCalls = 0; + final socket = _RecordingRelaySocket(); + final delivered = []; + final closed = []; + final session = RelaySessionNotifier( + verifyEventBatch: (events) { + verificationCalls++; + if (verificationCalls == 1) return firstVerification.future; + return Future.value(List.filled(events.length, null)); + }, + ); + session.debugAttachSocketForTest(socket); + final subscribe = session.subscribe( + _channelFilter, + delivered.add, + onClosed: closed.add, + ); + final Future? subscribeFailure = initiallyReady + ? null + : expectLater(subscribe, throwsStateError); + void Function()? unsubscribe; + if (initiallyReady) { + session.debugHandleMessage(['EOSE', 'l-1']); + unsubscribe = await subscribe; + } + final valid = _signedEvent(); + session.debugHandleMessage(['EVENT', 'l-1', valid.toJson()]); + await _waitUntil(() => verificationCalls == 1); + final large = { + ...valid.toJson(), + 'content': List.filled(600000, 'x').join(), + }; + for (var i = 0; i < 12; i++) { + session.debugHandleMessage(['EVENT', 'l-1', large]); + } + + await _waitUntil(() => _closes(socket, 'l-1').isNotEmpty); + if (initiallyReady) { + expect(closed, hasLength(1)); + } else { + await subscribeFailure; + expect(closed, isEmpty); + } + expect(_closes(socket, 'l-1'), hasLength(1)); + expect(session.debugPendingRelayAdmissions, lessThanOrEqualTo(10)); + + firstVerification.complete([valid.toJson()]); + await _waitUntil(() => session.debugPendingRelayAdmissions == 0); + await Future.delayed(const Duration(milliseconds: 30)); + session.debugFlushEventBuffer(); + expect(delivered, isEmpty); + unsubscribe?.call(); + expect(_closes(socket, 'l-1'), hasLength(1)); + }); + } + test('disconnect immediately retires pending live verification', () async { final verification = Completer?>>(); var verificationStarted = false; @@ -1160,6 +1258,104 @@ void main() { expect(delivered, isEmpty); }); + for (final message in ['error: relay overloaded', 'error: query timed out']) { + test( + 'pending EVENT and EOSE are fenced by retryable CLOSED: $message', + () async { + final verification = Completer?>>(); + var verificationCalls = 0; + final timers = <_ManualTimer>[]; + final socket = _RecordingRelaySocket(); + final delivered = []; + final statuses = []; + final session = RelaySessionNotifier( + verifyEventBatch: (events) { + verificationCalls++; + if (verificationCalls == 1) return verification.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.subscribeWithStatus( + _channelFilter, + delivered.add, + onStatusChanged: statuses.add, + ); + session.debugHandleMessage(['EOSE', 'l-1']); + final unsubscribe = await subscribe; + final stale = _signedEvent(createdAt: 30, content: 'stale'); + + session.debugHandleMessage(['EVENT', 'l-1', stale.toJson()]); + session.debugHandleMessage(['EOSE', 'l-1']); + await _waitUntil(() => verificationCalls == 1); + session.debugHandleMessage(['CLOSED', 'l-1', message]); + expect(statuses, [ + RelaySubscriptionStatus.ready, + RelaySubscriptionStatus.retrying, + ]); + session.debugHandleMessage(['EOSE', 'l-1']); + expect(statuses, hasLength(2)); + + verification.complete([stale.toJson()]); + await _waitUntil(() => session.debugPendingRelayAdmissions == 0); + await Future.delayed(Duration.zero); + expect(delivered, isEmpty); + expect(statuses, hasLength(2)); + expect(timers.single.isActive, isTrue); + + timers.single.fire(); + await Future.delayed(Duration.zero); + final replayReq = _reqs(socket).where((req) => req[1] == 'l-1').last; + expect(replayReq[2], isNot(contains('since'))); + + final replay = _signedEvent(createdAt: 40, content: 'replay'); + session.debugHandleMessage(['EVENT', 'l-1', replay.toJson()]); + session.debugHandleMessage(['EOSE', 'l-1']); + await _waitUntil(() => statuses.length == 3); + expect(delivered, [replay]); + expect(statuses.last, RelaySubscriptionStatus.ready); + unsubscribe(); + }, + ); + } + + 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); + + 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)); + + expect(delivered, isEmpty); + expect(timers.single.isActive, isTrue); + unsubscribe(); + }); + test('local OK acknowledgment bypasses pending verification', () async { final verification = Completer?>>(); var verificationStarted = false; @@ -1240,6 +1436,8 @@ void main() { test('terminal CLOSED fails a live subscribe before ready', () async { final session = RelaySessionNotifier(); + final socket = _RecordingRelaySocket(); + session.debugAttachSocketForTest(socket); const filter = NostrFilter(kinds: [EventKind.agentObserverFrame], limit: 0); final subscribe = session.subscribe(filter, (_) {}); @@ -1259,6 +1457,7 @@ void main() { ), ), ); + expect(_closes(socket, 'l-1'), isEmpty); }); test( @@ -1366,6 +1565,8 @@ void main() { session.debugHandleMessage(['CLOSED', 'l-1', 'error: transient']); expect(timers.last.duration, const Duration(seconds: 2)); + timers.last.fire(); + await Future.delayed(Duration.zero); session.debugHandleMessage(['EOSE', 'l-1']); session.debugHandleMessage(['CLOSED', 'l-1', 'error: transient']); expect(timers.last.duration, const Duration(seconds: 1)); @@ -1424,6 +1625,8 @@ void main() { session.debugHandleMessage(['CLOSED', 'l-1', 'error: transient']); expect(timers.last.duration, const Duration(seconds: 2)); + timers.last.fire(); + await Future.delayed(Duration.zero); session.debugHandleMessage([ 'EVENT', 'l-1', @@ -1459,6 +1662,7 @@ void main() { session.debugResetClosedRetriesForDisconnect(); expect(timers.last.isActive, isFalse); session.debugSetSessionStatus(SessionStatus.connected); + await session.debugReplayLiveSubscriptions(); session.debugHandleMessage(['CLOSED', 'l-1', 'error: transient']); expect(timers.last.duration, const Duration(seconds: 1)); unsubscribe(); @@ -1930,7 +2134,16 @@ void main() { ); test('live onClosed callback runs only for a terminal CLOSED', () async { - final session = RelaySessionNotifier(); + final timers = <_ManualTimer>[]; + final socket = _RecordingRelaySocket(); + final session = RelaySessionNotifier( + retryTimerFactory: (duration, callback) { + final timer = _ManualTimer(duration, callback); + timers.add(timer); + return timer; + }, + ); + session.debugAttachSocketForTest(socket); final closedMessages = []; const filter = NostrFilter(kinds: [EventKind.agentObserverFrame], limit: 0); @@ -1947,6 +2160,8 @@ void main() { 'error: temporarily unavailable', ]); expect(closedMessages, isEmpty); + timers.single.fire(); + await Future.delayed(Duration.zero); session.debugHandleMessage([ 'CLOSED', 'l-1', @@ -2316,6 +2531,16 @@ List _replayedChannelIds(_RecordingRelaySocket socket) => _reqs(socket) List> _reqs(_RecordingRelaySocket socket) => socket.messages.where((message) => message.first == 'REQ').toList(); +List> _closes(_RecordingRelaySocket socket, String subId) => + socket.messages + .where( + (message) => + message.length >= 2 && + message.first == 'CLOSE' && + message[1] == subId, + ) + .toList(); + class _RecordingRelaySocket extends RelaySocket { _RecordingRelaySocket() : super(