Harden relay verification admission lifecycle

Signed-off-by: jm <jm@squareup.com>
Co-authored-by: Codex <noreply@openai.com>

mobile/lib/shared/relay/relay_session.dart

@@ -34,9 +34,8 @@ class _HistorySubscription {
   final Timer timeout;
   Future<void> 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<void> 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<SessionState> {
   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<SessionState> {
       readyCompleter: readyCompleter,
     );
 
-    _sendReq(subId, filter);
+    _sendLiveReq(subId, _liveSubscriptions[subId]!, filter);
 
     try {
       await readyCompleter.future.timeout(
@@ -406,6 +409,9 @@ class RelaySessionNotifier extends Notifier<SessionState> {
   @visibleForTesting
   void debugDispose() => _dispose();
 
+  @visibleForTesting
+  int get debugPendingRelayAdmissions => _pendingRelayAdmissions;
+
   @visibleForTesting
   void debugSupersedeConnection() {
     _connectionGeneration++;
@@ -652,7 +658,7 @@ class RelaySessionNotifier extends Notifier<SessionState> {
           }
           _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<SessionState> {
 
     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<SessionState> {
       _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<SessionState> {
     _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<SessionState> {
     _eventBuffer.clear();
     _eventVerifier?.close();
     _eventVerifier = null;
+    _pendingRelayAdmissions = 0;
+    _pendingRelayAdmissionBytes = 0;
     _socket?.dispose();
     _socket = null;
     _httpQueryClient.close();

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<NostrEvent?> _observeHistoryVerification(
+    String subId,
+    _HistorySubscription subscription,
+    Future<NostrEvent?> 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<NostrEvent?> _observeLiveVerification(
+    String subId,
+    _LiveSubscription subscription,
+    Future<NostrEvent?> 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<dynamic> 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<void> _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;
+    }
   }
 }

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<String, dynamic>;
+    final sections = (decoded['sections'] as List).cast<Map<String, dynamic>>();
+    final renamedIds = <String, String>{};
+    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<String, dynamic>;
+    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<String, dynamic>;
+  final channels = decoded['channels'] as Map<String, dynamic>;
+  for (final entry in channels.values.cast<Map<String, dynamic>>()) {
+    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<String> 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<void>();
-      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<void>();
-      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<void>();
-      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,

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 = <NostrEvent>[];
@@ -44,34 +46,54 @@ class SidebarRelay {
     return NostrEvent.fromJson(event.toMap());
   }
 
-  NostrEvent eventComparedTo(
+  NostrEvent deterministicEvent(
+    String dTag,
+    Object payload,
+    int createdAt, {
+    List<List<String>> 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);

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<List<Map<String, dynamic>?>>();
+      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<List<Map<String, dynamic>?>>();
+      var verificationCalls = 0;
+      final socket = _RecordingRelaySocket();
+      final delivered = <NostrEvent>[];
+      final closed = <String>[];
+      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<void>? 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<void>.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<List<Map<String, dynamic>?>>();
     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<List<Map<String, dynamic>?>>();
+        var verificationCalls = 0;
+        final timers = <_ManualTimer>[];
+        final socket = _RecordingRelaySocket();
+        final delivered = <NostrEvent>[];
+        final statuses = <RelaySubscriptionStatus>[];
+        final session = RelaySessionNotifier(
+          verifyEventBatch: (events) {
+            verificationCalls++;
+            if (verificationCalls == 1) return verification.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.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<void>.delayed(Duration.zero);
+        expect(delivered, isEmpty);
+        expect(statuses, hasLength(2));
+        expect(timers.single.isActive, isTrue);
+
+        timers.single.fire();
+        await Future<void>.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<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);
+
+    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));
+
+    expect(delivered, isEmpty);
+    expect(timers.single.isActive, isTrue);
+    unsubscribe();
+  });
+
   test('local OK acknowledgment bypasses pending verification', () async {
     final verification = Completer<List<Map<String, dynamic>?>>();
     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<void>.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<void>.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 = <String>[];
     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<void>.delayed(Duration.zero);
     session.debugHandleMessage([
       'CLOSED',
       'l-1',
@@ -2316,6 +2531,16 @@ List<String> _replayedChannelIds(_RecordingRelaySocket socket) => _reqs(socket)
 List<List<dynamic>> _reqs(_RecordingRelaySocket socket) =>
     socket.messages.where((message) => message.first == 'REQ').toList();
 
+List<List<dynamic>> _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(