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 <noreply@anthropic.com>
Signed-off-by: Tom Brow <tomb@squareup.com>
This commit is contained in:
Tom Brow
2026-11-01 13:44:05 -08:00
parent 93b8b8febc
commit a8738761e0
3 changed files with 136 additions and 1 deletions

View File

@@ -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 = <String>{};
@@ -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<NostrEvent> events,
Map<String, int?> readAtByChannel,
int requestedAt,
) {
final newest = <String, int>{};
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],
);
}
}
}

View File

@@ -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<List<Channel>> {
String? _subscriptionRelayBaseUrl;
Timer? _backstopTimer;
final Map<String, int> _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<String, ({int through, int? readAt})> _unreadCaughtUpByChannel = {};
final Map<String, Map<String, ObservedUnreadEvent>>
_observedUnreadEventsByChannel = {};
final Map<String, Set<String>> _unreadDeletedIdsByChannel = {};
@@ -241,6 +249,7 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
_clearLiveSubscriptions();
_latestObservedByChannel.clear();
_observedUnreadEventsByChannel.clear();
_unreadCaughtUpByChannel.clear();
_unreadDeletedIdsByChannel.clear();
_unscopedUnreadDeletedIds.clear();
_backstopTimer?.cancel();

View File

@@ -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<int?> 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)],