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 {