jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-03 07:51:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(mobile): start a repeated catch-up before the relay's drift window

Browse Source 
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 <noreply@anthropic.com>
Signed-off-by: Tom Brow <tomb@squareup.com>
This commit is contained in:
Tom Brow
2026-11-01 14:01:54 -08:00
parent a8738761e0
commit f71c6c7600
3 changed files with 49 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
mobile/lib/features/channels/channel_unread_catch_up.dart
View File

@@ -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<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],
);
/// Remembers when a catch-up that covered every channel in
/// [readAtByChannel] was sent, and the read timestamps it used.
void _recordUnreadCaughtUp(Map<String, int?> readAtByChannel, int sentAt) {
for (final MapEntry(key: channelId, value: readAt)
in readAtByChannel.entries) {
_unreadCaughtUpByChannel[channelId] = (sentAt: sentAt, readAt: readAt);
}
}
}

10
mobile/lib/features/channels/channels_provider.dart
View File

@@ -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<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 = {};
// Per channel, when an earlier unread catch-up in this session was sent and
// the read timestamp it used. See [_UnreadCatchUp].
final Map<String, ({int sentAt, int? readAt})> _unreadCaughtUpByChannel = {};
final Map<String, Map<String, ObservedUnreadEvent>>
_observedUnreadEventsByChannel = {};
final Map<String, Set<String>> _unreadDeletedIdsByChannel = {};

49
mobile/test/features/channels/channels_provider_test.dart
View File

@@ -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));
}
},
);