From f71c6c76002e6d148836b7a29cb87dd602813a61 Mon Sep 17 00:00:00 2001 From: Tom Brow Date: Sun, 1 Nov 2026 14:01:54 -0800 Subject: [PATCH] fix(mobile): start a repeated catch-up before the relay's drift window A repeated unread catch-up started 60s before the newest message an earlier one returned. A message that reaches the relay late with an older created_at, such as one composed offline, could fall before that start and never count as unread. The relay rejects events whose created_at is more than 900s from its clock, so the next catch-up now starts relayLiveWindowSkewSeconds (960s, as reconnect replay uses) before the previous one was sent. That covers every message stored since, however late, and needs no event timestamps. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Tom Brow --- .../channels/channel_unread_catch_up.dart | 45 ++++++----------- .../features/channels/channels_provider.dart | 10 ++-- .../channels/channels_provider_test.dart | 49 ++++++++++++------- 3 files changed, 49 insertions(+), 55 deletions(-) diff --git a/mobile/lib/features/channels/channel_unread_catch_up.dart b/mobile/lib/features/channels/channel_unread_catch_up.dart index d062671bc5..9ccf6f29ef 100644 --- a/mobile/lib/features/channels/channel_unread_catch_up.dart +++ b/mobile/lib/features/channels/channel_unread_catch_up.dart @@ -73,7 +73,7 @@ extension _UnreadCatchUp on ChannelsNotifier { return; } - _recordUnreadCaughtUp(events, readAtByChannel, requestedAt); + _recordUnreadCaughtUp(readAtByChannel, requestedAt); // Keep this response's deletion evidence independently of the bounded // cross-request tombstones: a large batch can evict its own early IDs. @@ -192,44 +192,27 @@ extension _UnreadCatchUp on ChannelsNotifier { /// refresh path catches up again (resume, reconnect, join), and on a slow /// link re-downloading up to [_unreadCatchUpLimit] messages per channel each /// time keeps the batch from ever finishing. A later catch-up therefore - /// starts from the newest message an earlier one returned, less - /// [_unreadCatchUpOverlap] for events that reach the relay out of order. - /// Messages before that point were already recorded or skipped. A read - /// timestamp that moved back since then makes the full window apply again. + /// starts [relayLiveWindowSkewSeconds] before an earlier one was sent. The + /// earlier one returned every message stored before it was sent, and the + /// relay stores no event with a `created_at` more than 900s behind its own + /// clock, so anything stored since then is newer than that start, however + /// late it arrived. A read timestamp that moved back since then makes the + /// full window apply again. int _unreadCatchUpSince(String channelId, int? readAt) { final afterRead = (readAt ?? -1) + 1; final caughtUp = _unreadCaughtUpByChannel[channelId]; if (caughtUp == null || (readAt ?? -1) < (caughtUp.readAt ?? -1)) { return afterRead; } - return max(afterRead, caughtUp.through - _unreadCatchUpOverlap); + return max(afterRead, caughtUp.sentAt - relayLiveWindowSkewSeconds); } - /// Remembers, per channel, the newest event [events] returned, capped at - /// [requestedAt] so a client clock running ahead cannot skip messages. - void _recordUnreadCaughtUp( - List events, - Map readAtByChannel, - int requestedAt, - ) { - final newest = {}; - for (final event in events) { - final channelId = event.channelId; - if (channelId == null || !readAtByChannel.containsKey(channelId)) { - continue; - } - newest.update( - channelId, - (at) => max(at, event.createdAt), - ifAbsent: () => event.createdAt, - ); - } - for (final MapEntry(key: channelId, value: at) in newest.entries) { - final previous = _unreadCaughtUpByChannel[channelId]?.through; - _unreadCaughtUpByChannel[channelId] = ( - through: max(previous ?? 0, min(at, requestedAt)), - readAt: readAtByChannel[channelId], - ); + /// Remembers when a catch-up that covered every channel in + /// [readAtByChannel] was sent, and the read timestamps it used. + void _recordUnreadCaughtUp(Map readAtByChannel, int sentAt) { + for (final MapEntry(key: channelId, value: readAt) + in readAtByChannel.entries) { + _unreadCaughtUpByChannel[channelId] = (sentAt: sentAt, readAt: readAt); } } } diff --git a/mobile/lib/features/channels/channels_provider.dart b/mobile/lib/features/channels/channels_provider.dart index 1fb2bd86a1..af76df06d4 100644 --- a/mobile/lib/features/channels/channels_provider.dart +++ b/mobile/lib/features/channels/channels_provider.dart @@ -34,10 +34,6 @@ part 'channels_provider_lifecycle.dart'; const _channelTypeOrder = {'stream': 0, 'forum': 1, 'dm': 2}; const _unreadCatchUpLimit = 1000; -/// How far before the newest message it already returned a later unread -/// catch-up starts. See `_unreadCatchUpSince`. -const _unreadCatchUpOverlap = 60; - /// Relay requests one channel-history batch keeps in flight at most. const _historyRequestConcurrency = 4; const _participatedRootIdsPrefix = 'buzz-thread-participation.v1'; @@ -68,9 +64,9 @@ class ChannelsNotifier extends AsyncNotifier> { String? _subscriptionRelayBaseUrl; Timer? _backstopTimer; final Map _latestObservedByChannel = {}; - // Per channel, the newest message an earlier unread catch-up in this session - // returned and the read timestamp it used. See [_UnreadCatchUp]. - final Map _unreadCaughtUpByChannel = {}; + // Per channel, when an earlier unread catch-up in this session was sent and + // the read timestamp it used. See [_UnreadCatchUp]. + final Map _unreadCaughtUpByChannel = {}; final Map> _observedUnreadEventsByChannel = {}; final Map> _unreadDeletedIdsByChannel = {}; diff --git a/mobile/test/features/channels/channels_provider_test.dart b/mobile/test/features/channels/channels_provider_test.dart index 4282425414..e7218a29ae 100644 --- a/mobile/test/features/channels/channels_provider_test.dart +++ b/mobile/test/features/channels/channels_provider_test.dart @@ -2163,12 +2163,13 @@ void main() { }); test( - 'a repeated catch-up starts near the newest message it already has', + 'a repeated catch-up starts before the last one by the relay drift', () async { + int now() => DateTime.now().millisecondsSinceEpoch ~/ 1000; final session = _FakeRelaySession( memberships: [_membership(_channelA, myPk)], metadata: [_meta(id: _channelA, name: 'general')], - recentMessages: [message('seen', 10000, mention: true)], + recentMessages: [message('seen', now() - 3600, mention: true)], ); final container = _buildContainer(session: session); addTearDown(container.dispose); @@ -2177,34 +2178,48 @@ void main() { for (final filter in session.queryBatches.lastWhere(isUnread)) filter.since, ]; + Future<(int, int)> refreshed() async { + final before = session.queryBatches.where(isUnread).length; + final from = now(); + await notifier.refresh(); + await _waitUntil( + () => session.queryBatches.where(isUnread).length > before, + ); + await _settle(); + return (from, now()); + } + final firstFrom = now(); await container.read(channelsProvider.future); await _waitUntil(() => session.queryBatches.any(isUnread)); await _settle(); + final firstTo = now(); // Never read: the first catch-up asks for the whole window. expect(lastUnreadSince(), [0, 0]); - // A message lands while the app is away, then a refresh catches up. - session.recentMessages.add(message('missed', 10030, mention: true)); - final before = session.queryBatches.where(isUnread).length; - await notifier.refresh(); - await _waitUntil( - () => session.queryBatches.where(isUnread).length > before, + // While the app is away, a message composed offline reaches the relay + // with a created_at well before the newest one the app has seen. + session.recentMessages.add( + message('late', firstFrom - 600, mention: true), ); - await _settle(); - expect(lastUnreadSince(), [10000 - 60, 10000 - 60]); + final (secondFrom, secondTo) = await refreshed(); + for (final since in lastUnreadSince()) { + expect( + since, + inInclusiveRange(firstFrom - 960, firstTo - 960), + reason: 'starts 960s before the previous catch-up was sent', + ); + } expect( notifier.observedUnreadEventsByChannel[_channelA]?.keys, - unorderedEquals(['seen', 'missed']), + unorderedEquals(['seen', 'late']), ); // Each catch-up moves the start forward. - final again = session.queryBatches.where(isUnread).length; - await notifier.refresh(); - await _waitUntil( - () => session.queryBatches.where(isUnread).length > again, - ); - expect(lastUnreadSince(), [10030 - 60, 10030 - 60]); + await refreshed(); + for (final since in lastUnreadSince()) { + expect(since, inInclusiveRange(secondFrom - 960, secondTo - 960)); + } }, );