diff --git a/mobile/lib/shared/read_state/read_state_manager.dart b/mobile/lib/shared/read_state/read_state_manager.dart index d95b0793af..65bae53328 100644 --- a/mobile/lib/shared/read_state/read_state_manager.dart +++ b/mobile/lib/shared/read_state/read_state_manager.dart @@ -81,6 +81,10 @@ class ReadStateManager { Completer? _publishCompleter; bool _remoteUnsupported = false; int _maxFetchedCreatedAt = 0; + // When the last successful history fetch was sent. A later fetch and the + // live subscription ask only for blobs stored since then. + int? _lastFetchSentAt; + final int Function() _now; final Map _contextSourceCreatedAt = {}; final Set _pendingSyncedAdvances = {}; @@ -93,7 +97,9 @@ class ReadStateManager { required SignedEventRelay? signedEventRelay, required bool remoteEnabled, required VoidCallback onChanged, - }) : _crypto = crypto, + @visibleForTesting int Function() now = currentUnixSeconds, + }) : _now = now, + _crypto = crypto, _storage = ReadStateStorage(prefs), _relaySession = relaySession, _signedEventRelay = signedEventRelay, @@ -142,6 +148,7 @@ class ReadStateManager { void markContextRead(String contextId, int unixTimestamp) { if (_disposed || isOverrideContext(contextId)) return; + _resyncIfClockMovedBack(); // Set first: retention keeps the most recently written marks, so the // save that follows must already see this read as the newest. _contextSourceCreatedAt[contextId] = max( @@ -274,6 +281,7 @@ class ReadStateManager { } Future _fetchAndMerge() async { + final sentAt = _now(); try { final events = await _relaySession!.fetchHistory( NostrFilter( @@ -282,11 +290,15 @@ class ReadStateManager { tags: const { '#t': ['read-state'], }, - since: currentUnixSeconds() - readStateHorizonSeconds, + since: max( + sentAt - readStateHorizonSeconds, + _storedSinceLastFetch ?? 0, + ), limit: readStateFetchLimit, ), ); _mergeEvents(events); + _lastFetchSentAt = sentAt; _persistLocalState(); _onChanged(); return true; @@ -364,6 +376,32 @@ class ReadStateManager { return changed; } + /// The earliest `created_at` a blob stored since the last successful fetch + /// was sent can carry, or null before one. Merging takes the newest mark per + /// context, so blobs from before it add nothing. Blobs come from other + /// devices, so this reaches back [relayCrossDeviceSkewSeconds]. + int? get _storedSinceLastFetch { + final sentAt = _lastFetchSentAt; + // A clock set back since then can't place blobs relative to it, so the + // next fetch covers the whole horizon again. + if (sentAt == null || sentAt > _now()) return null; + return sentAt - relayCrossDeviceSkewSeconds; + } + + /// The live subscription keeps the `since` it was opened with. When the + /// clock has since moved back behind the last fetch, that `since` may be in + /// the relay's future and hide other devices' blobs until a reconnect, so + /// resync now: the fetch covers the whole horizon and the subscription is + /// reopened from the corrected clock. The last fetch's time is dropped + /// first, so a resync that fails isn't retried on every read; the next + /// fetch covers the whole horizon either way. + void _resyncIfClockMovedBack() { + final sentAt = _lastFetchSentAt; + if (_unsubscribeLive == null || sentAt == null || sentAt <= _now()) return; + _lastFetchSentAt = null; + unawaited(reinitializeRemote()); + } + Future _startLiveSubscription() async { try { final unsub = await _relaySession!.subscribe( @@ -373,6 +411,8 @@ class ReadStateManager { tags: const { '#t': ['read-state'], }, + // The fetch just merged everything stored before this. + since: _storedSinceLastFetch, limit: readStateFetchLimit, ), _handleIncomingEvent, diff --git a/mobile/lib/shared/relay/relay_reconnect_policy.dart b/mobile/lib/shared/relay/relay_reconnect_policy.dart index 3d33b3d20b..5d2020ff97 100644 --- a/mobile/lib/shared/relay/relay_reconnect_policy.dart +++ b/mobile/lib/shared/relay/relay_reconnect_policy.dart @@ -27,6 +27,13 @@ const relayReplaySkewSeconds = 5; /// narrower window would miss it for a quiet subscription. const relayLiveWindowSkewSeconds = 960; +/// Seconds a repeated read of events other devices wrote reaches behind the +/// moment the previous read was sent. The relay accepts a `created_at` up to +/// 900s either side of its own clock, so a writer 900s slow and a reader 900s +/// fast are 1800s apart, plus 60s of margin. [relayLiveWindowSkewSeconds] +/// covers only the reader's own skew. +const relayCrossDeviceSkewSeconds = 1860; + /// Whether [closeCode] says the relay is restarting or going away. bool isRelayServiceRestartClose(int? closeCode) => closeCode == 1012 || closeCode == 1001; 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 0c2389b5c3..8bd9b78eea 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 @@ -685,6 +685,192 @@ void main() { expect(relay.liveSubscriptions, 0); }); + test('a re-sync asks only for blobs stored since the last fetch', () 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: () {}, + ); + + final before = currentUnixSeconds(); + await manager.initialize(); + final after = currentUnixSeconds(); + // The first fetch covers the whole horizon. + expect( + relay.fetchFilters.single.since, + inInclusiveRange( + before - readStateHorizonSeconds, + after - readStateHorizonSeconds, + ), + ); + // Blobs come from other devices: a writer 900s slow and this reader 900s + // fast are 1800s apart, plus a margin. + const overlap = 2 * 900 + 60; + final firstSince = inInclusiveRange(before - overlap, after - overlap); + expect(relay.subscribeFilters.single.since, firstSince); + + // A failed re-sync fetch leaves the next one starting from the last + // successful fetch. + relay.failNextFetch = true; + await manager.reinitializeRemote(); + await manager.reinitializeRemote(); + expect(relay.fetchFilters, hasLength(3)); + expect(relay.fetchFilters[1].since, firstSince); + expect(relay.fetchFilters[2].since, firstSince); + expect(relay.subscribeFilters.last.since, isNotNull); + + manager.dispose(); + }); + + test('a clock set back after a fetch resyncs the whole horizon', () 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(); + var now = 1000000 + 7200; + final manager = ReadStateManager( + pubkey: keychain.public, + relayUrl: _relayUrl, + prefs: prefs, + crypto: crypto, + relaySession: relay, + signedEventRelay: _FakeSignedEventRelay(), + remoteEnabled: true, + onChanged: () {}, + now: () => now, + ); + + // The first fetch ran while the clock was two hours fast. + await manager.initialize(); + now = 1000000; + await manager.reinitializeRemote(); + expect(relay.fetchFilters.last.since, now - readStateHorizonSeconds); + expect( + relay.subscribeFilters.last.since, + now - relayCrossDeviceSkewSeconds, + ); + + manager.dispose(); + }); + + test( + 'a read after the clock moves back reopens the 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(); + var now = 1000000 + 7200; + final manager = ReadStateManager( + pubkey: keychain.public, + relayUrl: _relayUrl, + prefs: prefs, + crypto: crypto, + relaySession: relay, + signedEventRelay: _FakeSignedEventRelay(), + remoteEnabled: true, + onChanged: () {}, + now: () => now, + ); + + // Initialized while the clock was two hours fast, so the live + // subscription's since is in the relay's future. + await manager.initialize(); + expect( + relay.subscribeFilters.last.since, + now - relayCrossDeviceSkewSeconds, + ); + + // The clock is corrected while connected. The next read resyncs. + now = 1000000; + manager.markContextRead('channel-a', now); + await pumpEventQueue(); + expect(relay.fetchFilters.last.since, now - readStateHorizonSeconds); + expect( + relay.subscribeFilters.last.since, + now - relayCrossDeviceSkewSeconds, + ); + expect(relay.subscribeCount, 2); + expect(relay.liveSubscriptions, 1); + + // Once resynced, reads don't resync again. + final fetches = relay.fetchCount; + manager.markContextRead('channel-b', now); + await pumpEventQueue(); + expect(relay.fetchCount, fetches); + expect(relay.subscribeCount, 2); + + manager.dispose(); + }, + ); + + test( + 'a failed resync after the clock moves back is not retried on every read', + () 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(); + var now = 1000000 + 7200; + final manager = ReadStateManager( + pubkey: keychain.public, + relayUrl: _relayUrl, + prefs: prefs, + crypto: crypto, + relaySession: relay, + signedEventRelay: _FakeSignedEventRelay(), + remoteEnabled: true, + onChanged: () {}, + now: () => now, + ); + await manager.initialize(); + + // The clock is corrected, and the resync's fetch fails. + now = 1000000; + relay.failNextFetch = true; + manager.markContextRead('channel-a', now); + await pumpEventQueue(); + final fetches = relay.fetchCount; + expect(relay.liveSubscriptions, 1); + expect(relay.subscribeFilters.last.since, isNull); + + // Later reads don't fetch again; the next reconnect covers the horizon. + manager.markContextRead('channel-b', now); + manager.markContextRead('channel-c', now); + await pumpEventQueue(); + expect(relay.fetchCount, fetches); + await manager.reinitializeRemote(); + expect(relay.fetchFilters.last.since, now - readStateHorizonSeconds); + + manager.dispose(); + }, + ); + test('remote read-state rollback is ignored', () async { SharedPreferences.setMockInitialValues({}); final prefs = await SharedPreferences.getInstance(); @@ -874,6 +1060,8 @@ class _FakeRelaySession extends RelaySessionNotifier { int fetchCount = 0; int subscribeCount = 0; int liveSubscriptions = 0; + final List fetchFilters = []; + final List subscribeFilters = []; @override Future> fetchHistory( @@ -881,6 +1069,7 @@ class _FakeRelaySession extends RelaySessionNotifier { Duration timeout = const Duration(seconds: 8), }) async { fetchCount++; + fetchFilters.add(filter); await fetchGate; if (failNextFetch) { failNextFetch = false; @@ -896,6 +1085,7 @@ class _FakeRelaySession extends RelaySessionNotifier { void Function(String message)? onClosed, }) async { subscribeCount++; + subscribeFilters.add(filter); liveSubscriptions++; var open = true; return () {