fix(mobile): resync read state from the last fetch, not the whole week (#8903)

## Summary

Every read-state resync (each reconnect and each resume from background)
fetches every read-state blob (kind 30078, `#t: read-state`) from the
past week, then opens a live subscription with no `since`, which replays
the same blobs again. Merging keeps the newest mark per context, so
blobs from before the previous fetch add nothing. Each blob is a whole
device's read marks, about 100KB for a heavy reader.

A later fetch and the live subscription now start
`relayCrossDeviceSkewSeconds` (1860s) before the last successful fetch
was sent. The relay rejects a `created_at` more than 900s from its
clock, and blobs come from other devices, so a writer 900s slow and this
phone 900s fast are at most 1800s apart. Every blob stored since that
fetch falls inside the window. A failed fetch keeps the earlier start.
The first fetch of a manager still covers the whole horizon.

Companion to [#8901](https://github.com/block/buzz/pull/8901) and
[#8902](https://github.com/block/buzz/pull/8902), which do the same for
the unread catch-up and the latest-message query.

Measured on an iPhone 12 mini against the internal relay, resuming after
60s in the background (kind 30078 websocket bytes, instrumented profile
build):

| | main | this PR |
| --- | --- | --- |
| Read-state resync on resume | about 1.8MB (91 blobs, fetch plus
subscription replay) | 415KB (4 blobs changed since the last fetch) |

## Testing

- New test: the first fetch covers the horizon, the live subscription
and later fetches start 1860s before the last successful fetch was sent,
and a failed fetch keeps that start. Confirmed it fails without the fix
and with the earlier 960s window.
- New test: after the clock moves back, a resync whose fetch fails isn't
retried on every later read; the next reconnect fetches the whole
horizon. Confirmed it fails without the fix.
- `just mobile-check` and `just mobile-test` pass.
- Device: resumed the app on the 12 mini on Wi-Fi; the resync fetched
only the 4 blobs stored since the previous fetch, and the live
subscription replayed none.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Signed-off-by: Tom Brow <tomb@squareup.com>
Co-authored-by: Tom Brow <tomb@squareup.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

mobile/lib/shared/read_state/read_state_manager.dart

@@ -81,6 +81,10 @@ class ReadStateManager {
   Completer<void>? _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<String, int> _contextSourceCreatedAt = {};
   final Set<String> _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<bool> _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<bool> _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,

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;

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<NostrFilter> fetchFilters = [];
+  final List<NostrFilter> subscribeFilters = [];
 
   @override
   Future<List<NostrEvent>> 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 () {