From d090ac24bb7c78eb8a9db68f422ffaf7f8c54bfc Mon Sep 17 00:00:00 2001 From: Wes Date: Mon, 26 Oct 2026 10:24:43 -0700 Subject: [PATCH] perf(mobile): cut idle and resume sync churn (#8490) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Mobile does background work nobody sees, and it waits on round trips it doesn't need. This PR fixes four things: - **Chats you've left let go.** A chat with no listener for 30 s drops its live and typing subscriptions. Coming back shows the cached timeline right away, then resubscribes and catches up. - **No requests before AUTH.** The relay rejects any REQ that arrives before NIP-42 AUTH completes. History requests are now held until the socket authenticates, and they flush after the rate-limit gate. Live subscriptions skip their REQ, and the replay on connect sends it. `auth-required` is terminal for a live subscription, so a pre-AUTH live REQ could be dropped for good. - **Resume reconciles the channel list once.** `resumed` is published before the session hears about it. So the channel list now decides one microtask later: if the session is reconnecting, the connect does the reconcile. - **The 60 s backstop only runs in the foreground on a connected session.** After an OS suspension, the missed tick fires at unfreeze, about 450 ms before the app hears `resumed`, while the session still looks connected. Without this, that tick triggered a second full sync. Scope note: while the app is in the foreground, the backstop still re-downloads membership every 60 s. That matches desktop, which refetches every 60 s while focused. To drop it, mobile would need to subscribe to the live membership notices (44100/44101), as desktop does. That would be a follow-up. ### Related issue N/A. This is follow-up work from the mobile perf dive. None found. ## Before and after Simulator setup: iPhone 18 Pro simulator, debug build, a local relay behind a proxy that adds ~120 ms RTT, and a seeded account with 120 channels. I scripted one identical run against each build: 1. Open 5 channels. 2. Visit 20 more channels. 3. Wait 40 s. 4. Stay idle in the foreground for 3 min. 5. Go to the background for 3 min. 6. Resume. `main` is `ca3be5501`; the branch is `d77c4e86f`. The last amend changed only test names. Since then, chat history waits for the live EOSE again (see the note below), so channel open is back to `main`'s order. | | `main` | this PR | |---|---|---| | Live subscriptions after visiting 20 chats | 61 | 11 | | Resume: membership reads | 2 (244 kB) | 1 (122 kB) | | Resume: websocket REQs | 36 | 31 | | Resume: REQs sent before AUTH | 9 | 0 | | Resume: REQs rejected as `auth-required` | 2 | 0 | | Idle foreground, 3 min | 3 membership reads, 12 REQs | same | Notes: - **Channel open:** an earlier revision requested history without waiting for EOSE, which cut the median open from 297 ms to 197 ms. Review showed it can lose a message. The relay registers the live REQ only after its access checks, so a message committed in between reached neither the window read nor the subscription. A timestamp overlap only covers messages signed within it. Keeping that speedup safely would mean a second window read on every open, so it's left out. - **Idle foreground is unchanged by design.** See the scope note above. - **Background traffic:** the simulator suspends the app shortly after it leaves the foreground, so both builds sent almost nothing during the 3 minutes in the background. The backstop fix shows up at unfreeze instead, as the missing second membership read. - **Resume inbound events:** in this one run, events received over the websocket on resume were 247 kB on `main` and 338 kB on the branch. I haven't attributed the difference. ## Test These extend the existing journeys, plus one new journey: - **`channel_messages_journey_test`:** - The window read waits for the live REQ's admission, and a message signed a minute earlier but committed before admission shows up. - After the listener leaves, the chat keeps following for 29 s and stops at 31 s. - Coming back shows the cached rows immediately, then catches up on a message sent while away. - **`relay_session_test`:** a stale-connected resume sends nothing before AUTH. Then exactly one history REQ and one live REQ go out after AUTH. - **New `channels_sync_journey_test`:** drives the real lifecycle, session and channel notifier against `FakeRelay`: - A backgrounding just before a backstop tick sends nothing. - A resume after the session paused syncs once. - An OS-suspension resume with AUTH held syncs once, after AUTH, and sends no REQ before it. - **`FakeRelay`:** gained `holdReq`, `holdAuth`, `before_id` pagination and an injectable clock. Each guard was mutation-checked, and every mutation failed its test: - live REQ sent before AUTH - history sent before AUTH - resume refresh not deferred - backstop without the lifecycle check - history read alongside the live REQ - a 30-day detach delay Follow-up, not in this PR: a history REQ already in flight on a socket that gets replaced still waits for its 8 s timeout. Princess Donut found this in review; it predates this change. ## Validation - `flutter analyze`: clean - Full mobile suite at `678617b`: 3,696 passed - Court review (Princess Donut on the relay session, Mongo on chat subscriptions, Mordecai on the channel list plus a challenge pass): no blocking findings. The rate-limit flush ordering came from that review. **Human test:** Wes ran `da5077d` on the simulator with his real account and confirmed it works. `460dc35` then fixed a review blocker: a chat rebuilt while detached stayed detached. A retest of that path is pending: leave a chat for more than 30 s, background the app for more than 5 s, resume, then reopen the chat. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Signed-off-by: Wes Co-authored-by: Carl <32a2e2c9d428ee08902cab75d956da2c1d235a22d4766b0dd4138bf6e2e5db1d@buzz.block.builderlab.xyz> Co-authored-by: Claude Opus 5.5 --- .../channels/channel_messages_provider.dart | 40 ++++++ .../channels/channel_typing_provider.dart | 54 ++++++-- .../features/channels/channels_provider.dart | 13 +- .../channels/channels_provider_lifecycle.dart | 15 ++- mobile/lib/shared/relay/relay_session.dart | 33 ++++- .../channel_messages_journey_test.dart | 58 ++++++++- .../channels/channels_sync_journey_test.dart | 119 ++++++++++++++++++ mobile/test/helpers/fake_relay.dart | 85 +++++++++++-- .../test/shared/relay/relay_session_test.dart | 22 +++- 9 files changed, 402 insertions(+), 37 deletions(-) create mode 100644 mobile/test/features/channels/channels_sync_journey_test.dart diff --git a/mobile/lib/features/channels/channel_messages_provider.dart b/mobile/lib/features/channels/channel_messages_provider.dart index 42d6fd66e7..f4009f060a 100644 --- a/mobile/lib/features/channels/channel_messages_provider.dart +++ b/mobile/lib/features/channels/channel_messages_provider.dart @@ -1,3 +1,5 @@ +import 'dart:async'; + import 'package:flutter/foundation.dart'; import 'package:hooks_riverpod/hooks_riverpod.dart'; @@ -17,6 +19,10 @@ part 'channel_messages_provider/pagination.dart'; const _maxCachedRepliesPerRoot = 256; const _maxCachedReplies = 2048; +/// How long a chat nobody shows keeps its live subscription. A brief cover, +/// such as a profile pushed over the chat, returns without a catch-up fetch. +const _liveDetachDelay = Duration(seconds: 30); + const _channelLiveEventKinds = [ ...EventKind.channelEventKinds, EventKind.channelThreadSummary, @@ -59,6 +65,8 @@ class ChannelMessagesNotifier extends Notifier>> { ref.read(relaySessionProvider).status == SessionStatus.connected, ); bool _hasListeners = true; + Timer? _liveDetachTimer; + bool _liveDetached = false; late final _summaryRefreshes = ThreadSummaryRefreshQueue( refresh: _refreshOverflowSummary, canRun: () => @@ -115,18 +123,38 @@ class ChannelMessagesNotifier extends Notifier>> { @override AsyncValue> build() { final sessionState = ref.watch(relaySessionProvider); + // Riverpod defers rebuilding a provider nobody is listening to until it + // is listened to or read again, and then skips onResume. So a build + // always means the chat is wanted. + _hasListeners = true; + _liveDetached = false; ref.onCancel(() { _hasListeners = false; _deletionOwnerQueue.pause(); _summaryRefreshes.pause(); + _liveDetachTimer?.cancel(); + _liveDetachTimer = Timer(_liveDetachDelay, _detachLive); }); ref.onResume(() { _hasListeners = true; + _liveDetachTimer?.cancel(); + _liveDetachTimer = null; + if (_liveDetached) { + // Keep the cached timeline on screen while a rebuild resubscribes + // and catches up, as it does after a reconnect. Life-cycle callbacks + // can't use ref, so rebuild just after. + _liveDetached = false; + Future.microtask(() { + if (ref.mounted) ref.invalidateSelf(); + }); + } Future.microtask(_summaryRefreshes.resume); Future.microtask(_deletionOwnerQueue.resume); }); ref.onDispose(() { _initVersion++; + _liveDetachTimer?.cancel(); + _liveDetachTimer = null; _deletionOwnerQueue.pause(); _summaryRefreshes.pause(); _clearSubscription(); @@ -868,6 +896,18 @@ class ChannelMessagesNotifier extends Notifier>> { bool _isCurrentInit(int initVersion) => initVersion == _initVersion; + /// Stops live work for a chat that stayed hidden; resume catches it up. + void _detachLive() { + _liveDetachTimer = null; + _liveDetached = true; + // Retire an unfinished init so its subscription can't install later. + _initVersion++; + _initInFlight = false; + _initialWindowQueryInFlight = false; + _liveSummaryRootsDuringInitialWindowQuery.clear(); + _clearSubscription(); + } + void _clearSubscription() { _unsubscribe?.call(); _unsubscribe = null; diff --git a/mobile/lib/features/channels/channel_typing_provider.dart b/mobile/lib/features/channels/channel_typing_provider.dart index 1d04360c08..cf299eaaa6 100644 --- a/mobile/lib/features/channels/channel_typing_provider.dart +++ b/mobile/lib/features/channels/channel_typing_provider.dart @@ -27,11 +27,16 @@ class ChannelTypingNotifier extends Notifier> { static const _ttlMs = 8000; static const _pruneIntervalMs = 1000; + /// How long a channel nobody shows keeps its typing subscription. + static const _detachDelay = Duration(seconds: 30); + final String channelId; void Function()? _unsubscribe; Timer? _pruneTimer; Timer? _retryTimer; Timer? _healthyTimer; + Timer? _detachTimer; + bool _detached = false; int _generation = 0; final _activity = <(String, String?), _TypingActivity>{}; @@ -42,18 +47,34 @@ class ChannelTypingNotifier extends Notifier> { final sessionState = ref.watch(relaySessionProvider); final generation = ++_generation; _activity.clear(); + // Riverpod defers rebuilding a provider nobody is listening to until it + // is listened to or read again, and then skips onResume. So a build + // always means the channel is wanted. + _detached = false; + ref.onCancel(() { + _detachTimer?.cancel(); + _detachTimer = Timer(_detachDelay, () { + _detachTimer = null; + _detached = true; + _stop(); + state = []; + }); + }); + ref.onResume(() { + _detachTimer?.cancel(); + _detachTimer = null; + if (_detached) { + _detached = false; + Future.microtask(() { + if (ref.mounted) ref.invalidateSelf(); + }); + } + }); ref.onDispose(() { - _generation++; - _activity.clear(); - _unsubscribe?.call(); - _unsubscribe = null; - _pruneTimer?.cancel(); - _pruneTimer = null; - _retryTimer?.cancel(); - _retryTimer = null; - _healthyTimer?.cancel(); - _healthyTimer = null; + _detachTimer?.cancel(); + _detachTimer = null; + _stop(); }); if (sessionState.status == SessionStatus.connected) { @@ -63,6 +84,19 @@ class ChannelTypingNotifier extends Notifier> { return []; } + void _stop() { + _generation++; + _activity.clear(); + _unsubscribe?.call(); + _unsubscribe = null; + _pruneTimer?.cancel(); + _pruneTimer = null; + _retryTimer?.cancel(); + _retryTimer = null; + _healthyTimer?.cancel(); + _healthyTimer = null; + } + Future _subscribeLive(int generation, [int attempt = 0]) async { final session = ref.read(relaySessionProvider.notifier); var closed = false; diff --git a/mobile/lib/features/channels/channels_provider.dart b/mobile/lib/features/channels/channels_provider.dart index 1119a9a8f9..28a734f208 100644 --- a/mobile/lib/features/channels/channels_provider.dart +++ b/mobile/lib/features/channels/channels_provider.dart @@ -167,9 +167,16 @@ class ChannelsNotifier extends AsyncNotifier> { // Re-fetch when the app returns to foreground so channels created on // another device while mobile was backgrounded appear immediately. ref.listen(appLifecycleProvider, (prev, next) { - if (next == AppLifecycleState.resumed) { - refresh(); - } + if (next != AppLifecycleState.resumed) return; + // Resume is published before the session hears it. Let the session + // decide first: one that reconnects reconciles when it connects, so + // refreshing here too would fetch the same list twice. + scheduleMicrotask(() { + if (!ref.mounted) return; + if (ref.read(relaySessionProvider).status == SessionStatus.connected) { + refresh(); + } + }); }); ref.onDispose(() { diff --git a/mobile/lib/features/channels/channels_provider_lifecycle.dart b/mobile/lib/features/channels/channels_provider_lifecycle.dart index 9078d08dad..4be4412a86 100644 --- a/mobile/lib/features/channels/channels_provider_lifecycle.dart +++ b/mobile/lib/features/channels/channels_provider_lifecycle.dart @@ -166,10 +166,17 @@ extension _ChannelsNotifierLiveSubscriptions on ChannelsNotifier { unawaited(_unreadCatchUp); _backstopTimer?.cancel(); - _backstopTimer = Timer.periodic( - ChannelsNotifier._backstopInterval, - (_) => _backstopRefresh(), - ); + _backstopTimer = Timer.periodic(ChannelsNotifier._backstopInterval, (_) { + // A backgrounded app polls nothing, and a reconnecting session + // reconciles when it connects. After an OS suspension the missed tick + // fires before the app hears it has resumed, so check both. + if (_lifecycleRef.read(appLifecycleProvider) == + AppLifecycleState.resumed && + _lifecycleRef.read(relaySessionProvider).status == + SessionStatus.connected) { + unawaited(_backstopRefresh()); + } + }); } void _handleLiveChunkClosed(String chunkKey, int generation, String message) { diff --git a/mobile/lib/shared/relay/relay_session.dart b/mobile/lib/shared/relay/relay_session.dart index f955640a3e..8af77553d6 100644 --- a/mobile/lib/shared/relay/relay_session.dart +++ b/mobile/lib/shared/relay/relay_session.dart @@ -33,6 +33,9 @@ class _HistorySubscription { final Completer> completer; final Timer timeout; + /// The filter of a request held until the socket authenticates. + NostrFilter? heldFilter; + _HistorySubscription({required this.completer, required this.timeout}); } @@ -328,15 +331,22 @@ class RelaySessionNotifier extends Notifier { TimeoutException('Relay history request timed out after $timeout'), ); } - _sendClose(subId); + if (sub?.heldFilter == null) _sendClose(subId); }); - _historySubscriptions[subId] = _HistorySubscription( + final subscription = _HistorySubscription( completer: completer, timeout: timer, ); + _historySubscriptions[subId] = subscription; - _sendReq(subId, filter); + // The relay rejects a REQ sent before NIP-42 AUTH completes. Hold it for + // the authenticated connection instead of paying a rejection and retry. + if (_socketConnected) { + _sendReq(subId, filter); + } else { + subscription.heldFilter = filter; + } return completer.future; } @@ -385,7 +395,9 @@ class RelaySessionNotifier extends Notifier { readyCompleter: readyCompleter, ); - _sendReq(subId, filter); + // Before AUTH the relay closes it as auth-required, which is terminal. + // The replay on connect sends it instead. + if (_socketConnected) _sendReq(subId, filter); try { await readyCompleter.future.timeout( @@ -573,6 +585,8 @@ class RelaySessionNotifier extends Notifier { if (_disposed || _ageRestricted) return; final generation = ++_connectionGeneration; + // A replacement socket is unauthenticated until its AUTH completes. + _socketConnected = false; state = SessionState( status: _hasConnectedOnce ? SessionStatus.reconnecting @@ -668,6 +682,7 @@ class RelaySessionNotifier extends Notifier { Future _replayLiveSubscriptions(int generation) async { if (_rateLimitGate.isActive) await _rateLimitGate.wait(); if (!_isActiveConnection(generation)) return; + _sendHeldHistory(); final entries = _liveSubscriptions.entries.toList(); final visibleChannelId = _visibleChannelsByOwner.isEmpty @@ -1022,6 +1037,16 @@ class RelaySessionNotifier extends Notifier { _socket?.send(['REQ', subId, filter.toJson()]); } + void _sendHeldHistory() { + for (final MapEntry(key: subId, value: sub) + in _historySubscriptions.entries) { + final filter = sub.heldFilter; + if (filter == null) continue; + sub.heldFilter = null; + _sendReq(subId, filter); + } + } + void _sendClose(String subId) { _socket?.send(['CLOSE', subId]); } diff --git a/mobile/test/features/channels/channel_messages_journey_test.dart b/mobile/test/features/channels/channel_messages_journey_test.dart index 8b778f5795..cb0bac6439 100644 --- a/mobile/test/features/channels/channel_messages_journey_test.dart +++ b/mobile/test/features/channels/channel_messages_journey_test.dart @@ -1,4 +1,5 @@ import 'package:buzz/features/channels/channel_messages_provider.dart'; +import 'package:buzz/features/channels/channel_typing_provider.dart'; import 'package:buzz/features/channels/pending_local_messages_provider.dart'; import 'package:buzz/features/channels/send_message_provider.dart'; import 'package:buzz/shared/relay/relay.dart'; @@ -12,7 +13,7 @@ const _channel = '11111111-1111-4111-8111-111111111111'; void main() { test( - 'a channel loads, follows live messages, settles sends and recovers', + 'a channel loads, follows, settles sends, recovers and detaches once left', () { fakeAsync((async) { final relay = FakeRelay(); @@ -27,9 +28,11 @@ void main() { ); relay.seed([message('first', now - 20), message('second', now - 10)]); + // The relay admits the live REQ only after its access checks. + final admitted = relay.holdReq(); final container = ProviderContainer(overrides: relay.overrides); addTearDown(container.dispose); - final messages = container.listen( + var messages = container.listen( channelMessagesProvider(_channel), (_, _) {}, ); @@ -60,7 +63,13 @@ void main() { container.read(relaySessionProvider).status, SessionStatus.connected, ); - expect(timeline(), ['first', 'second']); + // History waits for the live EOSE, so a message committed before the + // relay admits the REQ is read, however long ago it was signed. + expect(relay.queries, isEmpty); + relay.seed([message('committed as it opened', now - 60)]); + admitted.complete(); + settle(); + expect(timeline(), ['committed as it opened', 'first', 'second']); expect(relay.queries, hasLength(1), reason: 'read the channel window'); expect( relay.reqs.where((req) => (req[1] as String).startsWith('h-')), @@ -123,6 +132,49 @@ void main() { settle(); expect(timeline(), contains('after reconnect')); expect(timeline().where((text) => text == 'hello'), hasLength(1)); + + // A chat left for a while stops following live, typing included; + // returning shows the cached rows at once and catches up on what + // arrived meanwhile. + bool follows({required bool typing}) => relay.liveFilters.any( + (filters) => filters.any( + (filter) => + ((filter['#h'] as List?)?.contains(_channel) ?? false) && + (filter['kinds'] as List).contains(EventKind.typingIndicator) == + typing, + ), + ); + bool followsChannel() => + follows(typing: false) && follows(typing: true); + bool followsNothing() => + !follows(typing: false) && !follows(typing: true); + var typing = container.listen( + channelTypingProvider(_channel), + (_, _) {}, + ); + settle(); + expect(followsChannel(), isTrue); + messages.close(); + typing.close(); + async.elapse(const Duration(seconds: 29)); + expect(followsChannel(), isTrue, reason: 'a brief cover keeps it'); + async.elapse(const Duration(seconds: 2)); + expect(followsNothing(), isTrue); + // A reconnect while it's hidden must not strand it detached. + relay.drop(); + async.elapse(const Duration(seconds: 2)); + expect(followsNothing(), isTrue); + relay.deliver(message('while away', liveSince() + 1)); + messages = container.listen( + channelMessagesProvider(_channel), + (_, _) {}, + ); + typing = container.listen(channelTypingProvider(_channel), (_, _) {}); + expect(timeline(), contains('after reconnect')); + settle(); + expect(followsChannel(), isTrue); + expect(timeline(), contains('while away')); + expect(timeline().where((text) => text == 'hello'), hasLength(1)); }); }, ); diff --git a/mobile/test/features/channels/channels_sync_journey_test.dart b/mobile/test/features/channels/channels_sync_journey_test.dart new file mode 100644 index 0000000000..b6cf055dee --- /dev/null +++ b/mobile/test/features/channels/channels_sync_journey_test.dart @@ -0,0 +1,119 @@ +import 'package:buzz/features/channels/channels_provider.dart'; +import 'package:buzz/shared/relay/relay.dart'; +import 'package:buzz/shared/theme/theme_provider.dart'; +import 'package:fake_async/fake_async.dart'; +import 'package:flutter/widgets.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:hooks_riverpod/hooks_riverpod.dart'; +import 'package:shared_preferences/shared_preferences.dart'; + +import '../../helpers/fake_relay.dart'; + +const _channel = '11111111-1111-4111-8111-111111111111'; + +void main() { + final binding = TestWidgetsFlutterBinding.ensureInitialized(); + + test( + 'the channel list reconciles once per resume and stays quiet while away', + () async { + SharedPreferences.setMockInitialValues({}); + final prefs = await SharedPreferences.getInstance(); + fakeAsync((async) { + var now = DateTime(2026, 10, 20, 12); + final relay = FakeRelay(now: () => now); + relay.seed([ + relay.sign( + kind: 39002, + tags: [ + ['d', _channel], + ['p', relay.pubkey], + ], + ), + relay.sign( + kind: 39000, + tags: [ + ['d', _channel], + ['name', 'general'], + ['t', 'stream'], + ['public'], + ], + ), + ]); + final container = ProviderContainer( + overrides: [ + ...relay.overrides, + savedPrefsProvider.overrideWithValue(prefs), + ], + ); + addTearDown(container.dispose); + final channels = container.listen(channelsProvider, (_, _) {}); + // A membership read is paginated; count each read's first page. + int membershipReads() => relay.queries + .expand((filters) => filters) + .where( + (f) => + f.containsKey('#p') && + f['kinds'].contains(39002) && + f['until'] == null, + ) + .length; + void lifecycle(AppLifecycleState state) => + binding.handleAppLifecycleStateChanged(state); + void settle() => async.elapse(const Duration(milliseconds: 50)); + + settle(); + expect(channels.read().requireValue.single.name, 'general'); + expect(membershipReads(), 1); + + // Backgrounded just before a backstop tick, while the session is + // still connected through its grace period: the 60s backstop sends + // nothing until the app returns. + async.elapse(const Duration(seconds: 58)); + expect(membershipReads(), 1); + lifecycle(AppLifecycleState.inactive); + lifecycle(AppLifecycleState.hidden); + lifecycle(AppLifecycleState.paused); + async.elapse(const Duration(minutes: 3)); + expect( + container.read(relaySessionProvider).status, + isNot(SessionStatus.connected), + ); + expect(membershipReads(), 1); + + lifecycle(AppLifecycleState.hidden); + lifecycle(AppLifecycleState.inactive); + lifecycle(AppLifecycleState.resumed); + settle(); + expect( + container.read(relaySessionProvider).status, + SessionStatus.connected, + ); + expect(membershipReads(), 2); + + // The OS suspends the app, so the session's pause timer never fires + // and it still believes it is connected when the app resumes. It + // reconnects; the list reconciles once, after AUTH, not before too. + lifecycle(AppLifecycleState.inactive); + lifecycle(AppLifecycleState.hidden); + lifecycle(AppLifecycleState.paused); + now = now.add(const Duration(minutes: 10)); + final auth = relay.holdAuth(); + lifecycle(AppLifecycleState.hidden); + lifecycle(AppLifecycleState.inactive); + lifecycle(AppLifecycleState.resumed); + settle(); + expect(membershipReads(), 2); + auth.complete(); + settle(); + expect( + container.read(relaySessionProvider).status, + SessionStatus.connected, + ); + expect(membershipReads(), 3); + expect(relay.reqsBeforeAuth, isEmpty); + expect(channels.read().requireValue.single.name, 'general'); + }); + }, + ); +} diff --git a/mobile/test/helpers/fake_relay.dart b/mobile/test/helpers/fake_relay.dart index 286fb90f02..eb44c4e954 100644 --- a/mobile/test/helpers/fake_relay.dart +++ b/mobile/test/helpers/fake_relay.dart @@ -22,7 +22,9 @@ import 'package:nostr/nostr.dart' as nostr; /// recorded and fails the test at teardown, because app code often catches /// relay errors and falls back. class FakeRelay { - FakeRelay() { + /// [now] is the session's wall clock, so a test can model OS suspension: + /// time passes while no timer runs. + FakeRelay({DateTime Function()? now}) : _now = now { addTearDown(() { expect(unanswerable, isEmpty, reason: 'requests FakeRelay cannot answer'); }); @@ -38,12 +40,16 @@ class FakeRelay { final _rejections = <({bool Function(NostrEvent) match, String message})>[]; Completer? _okHold; Completer? _echoHold; + Completer? _reqHold; + Completer? _authHold; + final DateTime Function()? _now; String get pubkey => keys.public; late final session = RelaySessionNotifier( socketFactory: _openSocket, httpClient: MockClient(_handleHttp), + now: _now, ); List get overrides => [ @@ -92,6 +98,22 @@ class FakeRelay { /// The relay dispatches after commit, so echo and OK have no fixed order. Completer holdEcho() => _echoHold = Completer(); + /// Holds new REQs until the completer completes, as a relay's access checks + /// would: they read stored events, start live fan-out and reach EOSE only + /// once admitted. Events committed meanwhile reach them only through replay. + Completer holdReq() => _reqHold = Completer(); + + /// Keeps the next connection unauthenticated until the completer completes. + /// Like the relay, it answers a REQ sent before AUTH with CLOSED. + Completer holdAuth() => _authHold = Completer(); + + /// REQs the relay rejected because they arrived before AUTH. + final reqsBeforeAuth = >[]; + + /// Filters of the live subscriptions open on the current connection. + Iterable>> get liveFilters => + sockets.isEmpty ? const [] : sockets.last._live.values; + /// Drops the current connection, as a network loss would. Its live /// subscriptions and queued replies die with it; stored history survives. void drop([Object? error]) => sockets.last.drop(error); @@ -116,18 +138,35 @@ class FakeRelay { void _receive(FakeRelaySocket socket, List message) { switch (message.first) { case 'REQ': - reqs.add(message); final subId = message[1] as String; + if (!socket._authenticated) { + reqsBeforeAuth.add(message); + scheduleMicrotask( + () => socket._reply([ + 'CLOSED', + subId, + 'auth-required: not authenticated', + ]), + ); + return; + } + reqs.add(message); final filters = message.skip(2).cast>().toList(); - final history = _query(filters); - if (history == null) return; - socket._live[subId] = filters; - scheduleMicrotask(() { - for (final event in history) { - socket._reply(['EVENT', subId, event.toJson()]); - } - socket._reply(['EOSE', subId]); - }); + void admit() { + if (!socket._authenticated) return; + final history = _query(filters); + if (history == null) return; + socket._live[subId] = filters; + scheduleMicrotask(() { + for (final event in history) { + socket._reply(['EVENT', subId, event.toJson()]); + } + socket._reply(['EOSE', subId]); + }); + } + + final hold = _reqHold; + hold == null ? admit() : hold.future.then((_) => admit()); case 'CLOSE': socket._live.remove(message[1]); case 'EVENT': @@ -224,6 +263,13 @@ class FakeRelay { final until = filter['until'] as int?; if (since != null && event.createdAt < since) return false; if (until != null && event.createdAt > until) return false; + // Buzz pagination: strictly after the cursor in `created_at DESC, id ASC`. + final beforeId = filter['before_id'] as String?; + if (beforeId != null && + event.createdAt == until && + event.id.compareTo(beforeId) <= 0) { + return false; + } for (final MapEntry(:key, :value) in filter.entries) { if (!key.startsWith('#')) continue; final name = key.substring(1); @@ -327,10 +373,16 @@ class FakeRelaySocket extends RelaySocket { final void Function(Object? error) _onDisconnected; final _live = >>{}; bool _open = false; + bool _authenticated = false; @override Future connect() async { _open = true; + final hold = _relay._authHold; + _relay._authHold = null; + if (hold != null) await hold.future; + if (!_open) return; + _authenticated = true; _onConnected(); } @@ -345,6 +397,7 @@ class FakeRelaySocket extends RelaySocket { @override void dispose() { _open = false; + _authenticated = false; _live.clear(); } @@ -359,7 +412,15 @@ class FakeRelaySocket extends RelaySocket { } } -const _filterKeys = {'ids', 'authors', 'kinds', 'since', 'until', 'limit'}; +const _filterKeys = { + 'ids', + 'authors', + 'kinds', + 'since', + 'until', + 'limit', + 'before_id', +}; const _windowKeys = { 'top_level', 'include_summaries', diff --git a/mobile/test/shared/relay/relay_session_test.dart b/mobile/test/shared/relay/relay_session_test.dart index 3d57dca120..18310bc1f0 100644 --- a/mobile/test/shared/relay/relay_session_test.dart +++ b/mobile/test/shared/relay/relay_session_test.dart @@ -815,7 +815,7 @@ void main() { }); test( - 'resume reconnects a stale connected session after a long pause', + 'resume reconnects a stale session and holds requests until AUTH', () async { final sockets = <_ControlledRelaySocket>[]; final keychain = nostr.Keys.generate(); @@ -868,6 +868,22 @@ void main() { expect(sockets, hasLength(2)); expect(sockets.first.disposeCalls, 1); expect(session.state.status, SessionStatus.reconnecting); + + // The relay rejects a REQ sent before AUTH, so the session holds it. + final history = session.fetchHistory( + NostrFilter(kinds: const [0], authors: [keychain.public]), + ); + unawaited(session.subscribe(NostrFilter(kinds: const [9]), (_) {})); + await Future.delayed(Duration.zero); + expect(sockets.last.sent, isEmpty); + sockets.last.connectSuccessfully(); + final reqs = sockets.last.sent; + expect(reqs.map((message) => message.first), ['REQ', 'REQ']); + final historyReq = reqs.singleWhere( + (message) => (message[2] as Map)['kinds'].contains(0), + ); + session.debugHandleSocketMessageForTest(['EOSE', historyReq[1]]); + expect(await history, isEmpty); }, ); @@ -1961,6 +1977,7 @@ class _ControlledRelaySocket extends RelaySocket { final void Function() _connected; final void Function(Object? error) _disconnected; int disposeCalls = 0; + final sent = >[]; _ControlledRelaySocket({ required super.wsUrl, @@ -1974,6 +1991,9 @@ class _ControlledRelaySocket extends RelaySocket { @override Future connect() async {} + @override + void send(List payload) => sent.add(payload); + @override void dispose() { disposeCalls++;