From a8738761e044ab8d07151349486337bb950c87bb Mon Sep 17 00:00:00 2001 From: Tom Brow Date: Sun, 1 Nov 2026 13:44:05 -0800 Subject: [PATCH] fix(mobile): catch up unread from the newest message already seen Every channel refresh path (resume, reconnect backstop, join) re-ran the unread catch-up from each channel's read timestamp, with up to 1000 full messages per channel. On a measured cold start that batch was 3.2MB of gzip on the wire, and it repeated on each refresh. On an Edge-class link it cannot finish inside the 8s history timeout, so it never completes. A later catch-up in the same session now starts 60s before the newest message an earlier one returned for that channel (capped at the request time). Unread results are unchanged: messages before that point were already recorded or skipped. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Tom Brow --- .../channels/channel_unread_catch_up.dart | 55 +++++++++++++- .../features/channels/channels_provider.dart | 9 +++ .../channels/channels_provider_test.dart | 73 +++++++++++++++++++ 3 files changed, 136 insertions(+), 1 deletion(-) diff --git a/mobile/lib/features/channels/channel_unread_catch_up.dart b/mobile/lib/features/channels/channel_unread_catch_up.dart index 7a8bc74a5e..d062671bc5 100644 --- a/mobile/lib/features/channels/channel_unread_catch_up.dart +++ b/mobile/lib/features/channels/channel_unread_catch_up.dart @@ -38,6 +38,7 @@ extension _UnreadCatchUp on ChannelsNotifier { for (final channel in activeChannels) channel.id: readState.effectiveTimestamp(channel.id), }; + final requestedAt = currentUnixSeconds(); final filters = [ for (final channel in activeChannels) for (final kinds in [ @@ -49,7 +50,10 @@ extension _UnreadCatchUp on ChannelsNotifier { tags: { '#h': [channel.id], }, - since: (readAtByChannel[channel.id] ?? -1) + 1, + since: _unreadCatchUpSince( + channel.id, + readAtByChannel[channel.id], + ), limit: _unreadCatchUpLimit, ), ]; @@ -69,6 +73,8 @@ extension _UnreadCatchUp on ChannelsNotifier { return; } + _recordUnreadCaughtUp(events, readAtByChannel, requestedAt); + // Keep this response's deletion evidence independently of the bounded // cross-request tombstones: a large batch can evict its own early IDs. final batchDeletedIds = {}; @@ -179,4 +185,51 @@ extension _UnreadCatchUp on ChannelsNotifier { } } } + + /// Where a channel's catch-up starts. + /// + /// The first catch-up in a session starts after the read timestamp. Every + /// 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. + 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); + } + + /// 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], + ); + } + } } diff --git a/mobile/lib/features/channels/channels_provider.dart b/mobile/lib/features/channels/channels_provider.dart index b5f3c9980e..1fb2bd86a1 100644 --- a/mobile/lib/features/channels/channels_provider.dart +++ b/mobile/lib/features/channels/channels_provider.dart @@ -16,6 +16,7 @@ import 'channel_management_provider.dart' import 'channel_mutes/channel_mutes_provider.dart'; import 'huddle_channel_filter.dart'; import '../../shared/read_state/read_state_provider.dart'; +import '../../shared/read_state/read_state_time.dart'; import '../../shared/profile/user_cache_provider.dart'; import 'thread_follows/thread_follows_provider.dart'; import 'unread_badge/is_high_priority_event.dart'; @@ -33,6 +34,10 @@ 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'; @@ -63,6 +68,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 = {}; final Map> _observedUnreadEventsByChannel = {}; final Map> _unreadDeletedIdsByChannel = {}; @@ -241,6 +249,7 @@ class ChannelsNotifier extends AsyncNotifier> { _clearLiveSubscriptions(); _latestObservedByChannel.clear(); _observedUnreadEventsByChannel.clear(); + _unreadCaughtUpByChannel.clear(); _unreadDeletedIdsByChannel.clear(); _unscopedUnreadDeletedIds.clear(); _backstopTimer?.cancel(); diff --git a/mobile/test/features/channels/channels_provider_test.dart b/mobile/test/features/channels/channels_provider_test.dart index 7868d46006..4282425414 100644 --- a/mobile/test/features/channels/channels_provider_test.dart +++ b/mobile/test/features/channels/channels_provider_test.dart @@ -2162,6 +2162,79 @@ void main() { ); }); + test( + 'a repeated catch-up starts near the newest message it already has', + () async { + final session = _FakeRelaySession( + memberships: [_membership(_channelA, myPk)], + metadata: [_meta(id: _channelA, name: 'general')], + recentMessages: [message('seen', 10000, mention: true)], + ); + final container = _buildContainer(session: session); + addTearDown(container.dispose); + final notifier = container.read(channelsProvider.notifier); + List lastUnreadSince() => [ + for (final filter in session.queryBatches.lastWhere(isUnread)) + filter.since, + ]; + + await container.read(channelsProvider.future); + await _waitUntil(() => session.queryBatches.any(isUnread)); + await _settle(); + // 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, + ); + await _settle(); + expect(lastUnreadSince(), [10000 - 60, 10000 - 60]); + expect( + notifier.observedUnreadEventsByChannel[_channelA]?.keys, + unorderedEquals(['seen', 'missed']), + ); + + // 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]); + }, + ); + + test('a rebuilt channel list catches up over the whole window', () async { + final session = _FakeRelaySession( + memberships: [_membership(_channelA, myPk)], + metadata: [_meta(id: _channelA, name: 'general')], + recentMessages: [message('seen', 10000, mention: true)], + ); + final container = _buildContainer(session: session); + addTearDown(container.dispose); + await container.read(channelsProvider.future); + await _waitUntil(() => session.queryBatches.any(isUnread)); + await _settle(); + + container.invalidate(channelsProvider); + final before = session.queryBatches.where(isUnread).length; + await container.read(channelsProvider.future); + await _waitUntil( + () => session.queryBatches.where(isUnread).length > before, + ); + expect( + [ + for (final filter in session.queryBatches.lastWhere(isUnread)) + filter.since, + ], + [0, 0], + ); + }); + test('quiet agent messages never count as unread', () async { final session = _FakeRelaySession( memberships: [_membership(_channelA, myPk)],