From 8abcf2ea2a416c5cfceba797cb14487c3612b190 Mon Sep 17 00:00:00 2001 From: Tom Brow Date: Thu, 29 Oct 2026 09:12:17 -0700 Subject: [PATCH] fix(mobile): keep read state across reconnects and resume (#8672) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ### Context Every app resume and relay reconnect tore down `ReadStateManager` and built a new one. `ReadStateNotifier.build()` watched the whole `relaySessionProvider` state, so each status change (disconnected, connecting, connected) rebuilt the notifier. On an iPhone 12 mini each resume re-ran the full read-state fetch and decrypt (about 19 to 24 decrypts, 60 to 95 ms). The rebuild also cleared `_forcedUnreadContexts`, so a message or channel the user marked unread lost that flag on any connection change. The watch dates from #422, when the manager was built with `remoteEnabled` tied to the connection status. #802 made `remoteEnabled` constant and added the `ref.listen` that calls `reinitializeRemote()` on connect, but left the watch behind. ### Change - Watch `relaySessionProvider.notifier` instead of its state. Config, nsec, community, pubkey, crypto and a replaced session still rebuild. Status changes don't. - Reconnects re-sync through the existing listener (`reinitializeRemote()`: re-fetch, then re-subscribe, preserving the history and live overlap). - Re-syncs are serialized. One runs at a time after any pending `initialize()`, and reconnects that arrive while one waits share it. Overlapping passes could otherwise each install a live subscription and leak one. - A connect that lands while the first `initialize()` is running (the cold-start connect) skips the re-sync when that initialize fetched and subscribed successfully. Without this, cold start did the read-state work twice. ### Validation - Provider journey: disconnected, connecting and reconnecting cause no refetch, no re-subscribe and no readiness drop, and the forced-unread flag and read mark survive. Connected runs exactly one refetch and one live-subscription swap. A second reconnect behaves the same. Account and community switches still rebuild and clear forced flags. - Manager tests: a connect during a successful `initialize()` does not sync again; one during a failed `initialize()` does, and leaves one live subscription; three reconnects during a running re-sync produce one more shared pass and one live subscription. - Falsified: restoring the old watch fails the reconnect journey (`isReady` false, flag lost). The previous version without serialized re-syncs fails the overlap test (5 passes instead of 3). - `just mobile-check` and `just mobile-test` pass. iPhone 12 mini (Profile builds of base and this branch, real community over the production relay, instrumented manager lifecycle and live-subscription counts): | | base | this branch | | --- | --- | --- | | Resume after 3 s, 10 s (x3), 60 s background | 3 new managers and 3 initializes per resume | no new manager, 1 re-sync | | Live read-state subscriptions after resume | 1 | 1 | | Channel marked unread, then 10 s / 60 s background | flag lost after the first resume | still unread after both | | Read-state work per resume (median) | ~24 decrypts, ~176 ms main isolate | ~18 decrypts, ~125 ms | | Unread badges | same bold set | same bold set | The first device run also found the doubled cold-start sync, fixed above. A re-check on this head: | Cold launch | base | first version | this head | | --- | --- | --- | --- | | Managers built | 3 | 1 | 1 | | Read-state syncs | 2 initializes | 1 initialize + 1 re-sync | 1 initialize | | Decrypts | 19 | 30 | 15 | | Main-isolate ms in read-state code | 124 | 185 to 197 | 94 to 98 | Four resumes on this head: no new manager, 1 re-sync and 1 live subscription each, 17 to 20 decrypts. A foreground socket drop was not exercised on the device; every resume there was a full disconnect and reconnect. buzz-review-completed 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Signed-off-by: Tom Brow Co-authored-by: Claude Opus 5.5 --- .../shared/read_state/read_state_manager.dart | 70 +++++++- .../read_state/read_state_provider.dart | 5 +- .../read_state/read_state_manager_test.dart | 153 +++++++++++++++++- .../read_state/read_state_provider_test.dart | 123 +++++++++++++- 4 files changed, 336 insertions(+), 15 deletions(-) diff --git a/mobile/lib/shared/read_state/read_state_manager.dart b/mobile/lib/shared/read_state/read_state_manager.dart index 4c80c2f697..d95b0793af 100644 --- a/mobile/lib/shared/read_state/read_state_manager.dart +++ b/mobile/lib/shared/read_state/read_state_manager.dart @@ -64,6 +64,15 @@ class ReadStateManager { Timer? _debounceTimer; void Function()? _unsubscribeLive; bool _initialized = false; + // Pending first fetch and subscribe. A reconnect while it runs waits for it + // so two passes cannot each install a live subscription and leak one. + Future? _initializing; + // The reconnect pass running now, and the one waiting to start after it. + Future? _runningResync; + Future? _queuedResync; + // Whether the first fetch and subscribe both succeeded. A connect that + // lands while they run (the cold-start connect) then has nothing to redo. + bool _initialSyncSucceeded = false; bool _disposed = false; bool _isPublishing = false; // Set when a publish is asked for while one is running. The running @@ -111,8 +120,16 @@ class ReadStateManager { return; } - await _fetchAndMerge(); - await _startLiveSubscription(); + final initializing = Completer(); + _initializing = initializing.future; + try { + final fetched = await _fetchAndMerge(); + final subscribed = await _startLiveSubscription(); + _initialSyncSucceeded = fetched && subscribed; + } finally { + _initializing = null; + initializing.complete(); + } if (!_isIdenticalToLastPublished(_currentContexts())) { _schedulePublish(); } @@ -162,8 +179,43 @@ class ReadStateManager { await _publish(); } - Future reinitializeRemote() async { - if (_disposed || !_remoteEnabled || !_initialized) return; + /// Fetch and resubscribe after a reconnect. Passes run one at a time, so + /// each replaces the previous live subscription instead of leaking one. + /// Reconnects while a pass waits to start share that pass. + Future reinitializeRemote() { + if (_disposed || !_remoteEnabled || !_initialized) return Future.value(); + final queued = _queuedResync; + if (queued != null) return queued; + final running = _runningResync; + final initializing = _initializing; + final pass = Completer(); + _queuedResync = pass.future; + unawaited(() async { + try { + if (running != null) { + await running; + } else if (initializing != null) { + await initializing; + // Connected while the first sync ran: it already used this + // connection. + if (_initialSyncSucceeded) return; + } + if (_disposed) return; + if (identical(_queuedResync, pass.future)) _queuedResync = null; + _runningResync = pass.future; + await _resyncRemote(); + } catch (e) { + debugPrint('[ReadStateManager] reinitializeRemote failed: $e'); + } finally { + if (identical(_queuedResync, pass.future)) _queuedResync = null; + if (identical(_runningResync, pass.future)) _runningResync = null; + pass.complete(); + } + }()); + return pass.future; + } + + Future _resyncRemote() async { debugPrint('[ReadStateManager] reinitializeRemote'); if (_isPublishing) { await _publishCompleter?.future; @@ -221,7 +273,7 @@ class ReadStateManager { return true; } - Future _fetchAndMerge() async { + Future _fetchAndMerge() async { try { final events = await _relaySession!.fetchHistory( NostrFilter( @@ -237,8 +289,10 @@ class ReadStateManager { _mergeEvents(events); _persistLocalState(); _onChanged(); + return true; } catch (e) { debugPrint('[ReadStateManager] fetchAndMerge failed: $e'); + return false; } } @@ -310,7 +364,7 @@ class ReadStateManager { return changed; } - Future _startLiveSubscription() async { + Future _startLiveSubscription() async { try { final unsub = await _relaySession!.subscribe( NostrFilter( @@ -325,12 +379,14 @@ class ReadStateManager { ); if (_disposed) { unsub.call(); - return; + return false; } _unsubscribeLive = unsub; debugPrint('[ReadStateManager] live subscription established'); + return true; } catch (e) { debugPrint('[ReadStateManager] live subscription FAILED: $e'); + return false; } } diff --git a/mobile/lib/shared/read_state/read_state_provider.dart b/mobile/lib/shared/read_state/read_state_provider.dart index f47ea707ee..4ff460c325 100644 --- a/mobile/lib/shared/read_state/read_state_provider.dart +++ b/mobile/lib/shared/read_state/read_state_provider.dart @@ -75,7 +75,10 @@ class ReadStateNotifier extends Notifier { _forcedUnreadContexts.clear(); final relayConfig = ref.watch(relayConfigProvider); - ref.watch(relaySessionProvider); + // Watch the session notifier, not its state: a status change (app resume, + // reconnect) must not rebuild the manager and drop session-local forced + // unread flags. The listener below re-syncs on reconnect instead. + ref.watch(relaySessionProvider.notifier); ref.watch(activeCommunityProvider); final nsec = relayConfig.nsec?.trim(); diff --git a/mobile/test/features/channels/read_state/read_state_manager_test.dart b/mobile/test/features/channels/read_state/read_state_manager_test.dart index c021bd9ab0..0c2389b5c3 100644 --- a/mobile/test/features/channels/read_state/read_state_manager_test.dart +++ b/mobile/test/features/channels/read_state/read_state_manager_test.dart @@ -557,6 +557,134 @@ void main() { expect(manager.getEffectiveTimestamp('channel-1'), 300); }); + test( + 'a connect during a successful initialize does not sync again', + () async { + SharedPreferences.setMockInitialValues({}); + final prefs = await SharedPreferences.getInstance(); + final keychain = nostr.Keys.generate(); + final crypto = ReadStateCrypto.tryCreate( + nsec: keychain.nsec, + pubkey: keychain.public, + )!; + // First connect: initialize waits on a history REQ held for the socket + // while the session reports connected and the provider's listener fires. + final gate = Completer(); + final relay = _FakeRelaySession()..fetchGate = gate.future; + final manager = ReadStateManager( + pubkey: keychain.public, + relayUrl: _relayUrl, + prefs: prefs, + crypto: crypto, + relaySession: relay, + signedEventRelay: _FakeSignedEventRelay(), + remoteEnabled: true, + onChanged: () {}, + ); + + final initializing = manager.initialize(); + final reinitializing = manager.reinitializeRemote(); + relay.fetchGate = null; + gate.complete(); + await Future.wait([initializing, reinitializing]); + + expect(relay.fetchCount, 1); + expect(relay.subscribeCount, 1); + expect(relay.liveSubscriptions, 1); + + // A later reconnect syncs again and still leaves one subscription. + await manager.reinitializeRemote(); + expect(relay.fetchCount, 2); + expect(relay.subscribeCount, 2); + expect(relay.liveSubscriptions, 1); + + manager.dispose(); + expect(relay.liveSubscriptions, 0); + }, + ); + + test('a connect during a failed initialize syncs again', () async { + SharedPreferences.setMockInitialValues({}); + final prefs = await SharedPreferences.getInstance(); + final keychain = nostr.Keys.generate(); + final crypto = ReadStateCrypto.tryCreate( + nsec: keychain.nsec, + pubkey: keychain.public, + )!; + final gate = Completer(); + final relay = _FakeRelaySession() + ..fetchGate = gate.future + ..failNextFetch = true; + final manager = ReadStateManager( + pubkey: keychain.public, + relayUrl: _relayUrl, + prefs: prefs, + crypto: crypto, + relaySession: relay, + signedEventRelay: _FakeSignedEventRelay(), + remoteEnabled: true, + onChanged: () {}, + ); + + final initializing = manager.initialize(); + final reinitializing = manager.reinitializeRemote(); + relay.fetchGate = null; + gate.complete(); + await Future.wait([initializing, reinitializing]); + + // Both passes subscribed, but the later one replaced the earlier one. + expect(relay.fetchCount, 2); + expect(relay.subscribeCount, 2); + expect(relay.liveSubscriptions, 1); + + manager.dispose(); + expect(relay.liveSubscriptions, 0); + }); + + test('reconnects during a re-sync leave one live subscription', () async { + SharedPreferences.setMockInitialValues({}); + final prefs = await SharedPreferences.getInstance(); + final keychain = nostr.Keys.generate(); + final crypto = ReadStateCrypto.tryCreate( + nsec: keychain.nsec, + pubkey: keychain.public, + )!; + final relay = _FakeRelaySession(); + final manager = ReadStateManager( + pubkey: keychain.public, + relayUrl: _relayUrl, + prefs: prefs, + crypto: crypto, + relaySession: relay, + signedEventRelay: _FakeSignedEventRelay(), + remoteEnabled: true, + onChanged: () {}, + ); + await manager.initialize(); + + // The first re-sync holds its fetch; three more connects arrive. + final gate = Completer(); + relay.fetchGate = gate.future; + final first = manager.reinitializeRemote(); + await Future.delayed(Duration.zero); + final later = [ + manager.reinitializeRemote(), + manager.reinitializeRemote(), + manager.reinitializeRemote(), + ]; + relay.fetchGate = null; + gate.complete(); + await Future.wait([first, ...later]); + + // One initial pass, the held re-sync, and one shared re-sync after it. + expect(relay.fetchCount, 3); + expect(relay.subscribeCount, 3); + expect(relay.liveSubscriptions, 1); + + manager.dispose(); + expect(relay.liveSubscriptions, 0); + }); + test('remote read-state rollback is ignored', () async { SharedPreferences.setMockInitialValues({}); final prefs = await SharedPreferences.getInstance(); @@ -741,17 +869,38 @@ NostrEvent _readStateEvent({ class _FakeRelaySession extends RelaySessionNotifier { List historyEvents = []; + Future? fetchGate; + bool failNextFetch = false; + int fetchCount = 0; + int subscribeCount = 0; + int liveSubscriptions = 0; @override Future> fetchHistory( NostrFilter filter, { Duration timeout = const Duration(seconds: 8), - }) async => historyEvents; + }) async { + fetchCount++; + await fetchGate; + if (failNextFetch) { + failNextFetch = false; + throw TimeoutException('history timed out'); + } + return historyEvents; + } @override Future subscribe( NostrFilter filter, void Function(NostrEvent) onEvent, { void Function(String message)? onClosed, - }) async => () {}; + }) async { + subscribeCount++; + liveSubscriptions++; + var open = true; + return () { + if (open) liveSubscriptions--; + open = false; + }; + } } diff --git a/mobile/test/features/channels/read_state/read_state_provider_test.dart b/mobile/test/features/channels/read_state/read_state_provider_test.dart index 57430ffa77..65ee76f658 100644 --- a/mobile/test/features/channels/read_state/read_state_provider_test.dart +++ b/mobile/test/features/channels/read_state/read_state_provider_test.dart @@ -24,6 +24,7 @@ void main() { late ProviderContainer container; late _FakeRelayConfig config; + late _FakeRelaySession session; Future pumpNotifier({ Map Function(String pubkey)? stored, @@ -53,11 +54,19 @@ void main() { await Future.delayed(Duration.zero); } expect(container.read(readStateProvider).isReady, isTrue); + session = + container.read(relaySessionProvider.notifier) as _FakeRelaySession; return notifier; } ReadStateState state() => container.read(readStateProvider); + Future flushMicrotasks() async { + for (var i = 0; i < 10; i++) { + await Future.delayed(Duration.zero); + } + } + Future settle() async { for (var i = 0; i < 10 && !state().isReady; i++) { await Future.delayed(Duration.zero); @@ -98,6 +107,86 @@ void main() { expect(state().effectiveTimestamp(channelId), isNull); }); + test('a disconnect and reconnect keeps the manager and forced unread ' + 'flags, and resyncs once on connect', () async { + final notifier = await pumpNotifier(); + // Persists on the same manager, so it survives only if that manager does. + notifier.markContextRead(channelId, 500); + notifier.markContextUnread(msgKey, channelId: channelId); + expect(session.fetchCount, 1); + expect(session.subscribeCount, 1); + final version = state().version; + + for (final status in [ + SessionStatus.disconnected, + SessionStatus.connecting, + SessionStatus.reconnecting, + ]) { + session.setStatus(status); + await flushMicrotasks(); + // No rebuild: still ready, flags kept, and nothing refetched. + expect(state().isReady, isTrue, reason: status.name); + expect(state().isForcedUnread(msgKey), isTrue, reason: status.name); + expect(state().effectiveTimestamp(channelId), 500); + expect(session.fetchCount, 1, reason: status.name); + expect(session.subscribeCount, 1, reason: status.name); + } + + session.setStatus(SessionStatus.connected); + await flushMicrotasks(); + + // The reconnect listener refetches and replaces the live subscription + // on the same manager. + expect(session.fetchCount, 2); + expect(session.subscribeCount, 2); + expect(session.liveSubscriptions, 1); + expect(state().isReady, isTrue); + expect(state().isForcedUnread(msgKey), isTrue); + expect(state().locallyForcedChannelIds, {channelId}); + expect(state().effectiveTimestamp(channelId), 500); + expect(state().version, greaterThanOrEqualTo(version)); + + // A second reconnect resyncs the same way. + session.setStatus(SessionStatus.disconnected); + session.setStatus(SessionStatus.connected); + await flushMicrotasks(); + expect(session.fetchCount, 3); + expect(session.liveSubscriptions, 1); + expect(state().isForcedUnread(msgKey), isTrue); + }); + + test('switching account rebuilds the manager and drops forced unread ' + 'flags', () async { + final notifier = await pumpNotifier(); + notifier.markContextUnread(msgKey, channelId: channelId); + expect(session.fetchCount, 1); + final pubkey = state().pubkey; + + config.switchAccount(nostr.Keys.generate().nsec); + container.read(readStateProvider); + await settle(); + + expect(state().pubkey, isNot(pubkey)); + expect(state().forcedUnreadContexts, isEmpty); + expect(session.fetchCount, 2); + expect(session.subscribeCount, 2); + expect(session.liveSubscriptions, 1); + }); + + test('switching community rebuilds the manager and drops forced unread ' + 'flags', () async { + final notifier = await pumpNotifier(); + notifier.markContextUnread(msgKey, channelId: channelId); + + config.switchTo('http://localhost:2'); + container.read(readStateProvider); + await settle(); + + expect(state().forcedUnreadContexts, isEmpty); + expect(session.fetchCount, 2); + expect(session.liveSubscriptions, 1); + }); + test('message unread → read → unread round-trips through the real ' 'notifier', () async { final notifier = await pumpNotifier(); @@ -189,7 +278,7 @@ void main() { } class _FakeRelayConfig extends RelayConfigNotifier { - final String nsec; + String nsec; _FakeRelayConfig(this.nsec); @@ -198,11 +287,23 @@ class _FakeRelayConfig extends RelayConfigNotifier { void switchTo(String baseUrl) => state = RelayConfig(baseUrl: baseUrl, nsec: nsec); + + void switchAccount(String newNsec) { + nsec = newNsec; + state = RelayConfig(baseUrl: state.baseUrl, nsec: newNsec); + } } -/// Relay session that never connects and returns no history, keeping the -/// manager fully local while exercising its real code paths. +/// Relay session that returns no history and counts the read-state fetches +/// and live subscriptions the manager makes, keeping it fully local while +/// exercising its real code paths. Its status is driven by [setStatus]. class _FakeRelaySession extends RelaySessionNotifier { + int fetchCount = 0; + int subscribeCount = 0; + int liveSubscriptions = 0; + + void setStatus(SessionStatus status) => state = SessionState(status: status); + @override SessionState build() => const SessionState(status: SessionStatus.disconnected); @@ -211,14 +312,26 @@ class _FakeRelaySession extends RelaySessionNotifier { Future> fetchHistory( NostrFilter filter, { Duration timeout = const Duration(seconds: 8), - }) async => []; + }) async { + // The manager's pre-publish own-slot lookup is not a full history fetch. + if (filter.tags.containsKey('#t')) fetchCount++; + return []; + } @override Future subscribe( NostrFilter filter, void Function(NostrEvent) onEvent, { void Function(String message)? onClosed, - }) async => () {}; + }) async { + subscribeCount++; + liveSubscriptions++; + var open = true; + return () { + if (open) liveSubscriptions--; + open = false; + }; + } } class _FakeAppLifecycle extends AppLifecycleNotifier {