From b310fea05b2ed2c2d04cc0da4ad16f38108b6a3c Mon Sep 17 00:00:00 2001 From: Tom Brow Date: Wed, 28 Oct 2026 12:00:58 -0700 Subject: [PATCH 1/2] mobile: find DMs opened ahead of the list everywhere (#8646) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary This follows up [#8638](https://github.com/block/buzz/pull/8638) and covers the optional review notes from that PR. - **Shared lookup.** A new `channelByIdProvider` returns the channel from the list, or the copy remembered from a fast DM open when the list doesn't have it yet. The mention picker, mention search failure state, compose bar DM check, thread page, and message unread action all use it. Before this change, those surfaces only searched the list, so a DM opened before the list loaded was treated as an unknown channel. For example, the mention picker skipped directory search and never labeled anyone "not in DM". - **Eviction and scope.** Opened-ahead DMs live in a notifier scoped to one relay and identity, and `channelByIdProvider` watches it. Once the list includes a remembered DM, its opened copy is dropped. Remembering also skips a DM the list already has, since the open's background refresh can list it first. A DM hidden or left later, or one from another community or identity, is no longer found through a stale copy. - **Call-site tests.** Search, the profile Message button, and the New message sheet now assert that they open the DM with `waitForChannelList: false`. ## Validation - `just mobile-check` and `just mobile-test` pass. - The journey test "a DM opened before the channel list loads sends mentions" now also checks the outside-person picker in a DM that isn't in the list yet: a failed directory search shows Retry, then the result appears labeled "not in DM". Mutation checks: reverting the lookup in `mentionCandidatesProvider`, `mentionSearchFailedProvider`, or the compose bar fails the test. - New tests "a DM opened ahead of the list is not found after it is left" (including the refresh-before-remember order) and "a DM opened ahead of the list stays in its community" (a mounted lookup across community and identity switches). Mutation checks: removing the eviction, the listed check in `rememberOpened`, the relay or identity watch, or the lookup's reactive watch each fails a test with `Expected: null, Actual: `. - Call-site assertions. Mutation check: flipping any of the three call sites to `waitForChannelList: true` fails its test with `Expected: [false], Actual: [true]`. - Simulator (iPhone 17 Pro, local relay): I started a fresh DM with Sim Tester from the New message sheet. It opened at once, and typing `@Pat` showed Pat Peer from the directory, labeled "not in DM". Fresh DM with Sim Tester: typing @Pat shows Pat Peer labeled not in DM buzz-review-completed 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Signed-off-by: Tom Brow Co-authored-by: george <973be878edd6d4f79692a3827f20a6f7fe0ed0362f269ff6b6ba8867033d7d8e@sprout-oss.stage.blox.sqprod.co> Co-authored-by: Claude Opus 5.5 --- .../features/channels/channels_provider.dart | 65 +++++++++++++--- .../compose_bar/compose_bar_widget.dart | 3 +- .../mentions/mention_candidates_provider.dart | 8 +- .../features/channels/message_actions.dart | 3 +- .../features/channels/thread_detail_page.dart | 18 +---- .../channels/channel_detail_page_test.dart | 6 +- .../features/channels/channels_page_test.dart | 27 +++++++ .../channels/channels_provider_test.dart | 77 +++++++++++++++++++ .../features/channels/compose_bar_test.dart | 21 +++++ .../user_profile_sheet_message_test.dart | 9 ++- .../features/search/search_page_test.dart | 9 ++- 11 files changed, 204 insertions(+), 42 deletions(-) diff --git a/mobile/lib/features/channels/channels_provider.dart b/mobile/lib/features/channels/channels_provider.dart index d1672ddb0e..c5f7babe51 100644 --- a/mobile/lib/features/channels/channels_provider.dart +++ b/mobile/lib/features/channels/channels_provider.dart @@ -78,11 +78,6 @@ class ChannelsNotifier extends AsyncNotifier> { List _directoryMetas = const []; Set _hiddenDmIds = const {}; - /// Channels opened from their own metadata before this list had them, as a - /// fast DM open does. The list takes precedence once it has them; a DM's - /// participants are fixed, so the opened copy stays valid until then. - Map _openedAhead = const {}; - /// Fences directory responses to the relay and identity that requested them. late final _ChannelRefreshCoordinator _refreshCoordinator = _ChannelRefreshCoordinator.forRef(ref); @@ -104,14 +99,17 @@ class ChannelsNotifier extends AsyncNotifier> { Set get hiddenDmIds => Set.unmodifiable(_hiddenDmIds); /// Keeps [channel], opened before this list has it, findable by - /// [channelById] while the list reloads. - void rememberOpened(Channel channel) => - _openedAhead = {..._openedAhead, channel.id: channel}; + /// [channelById] and [channelByIdProvider] until the list has it. + void rememberOpened(Channel channel) { + // The open's background refresh may have listed it already. + if (state.value?.any((listed) => listed.id == channel.id) ?? false) return; + ref.read(_openedAheadProvider.notifier).remember(channel); + } /// [channelId] from the list, else as it was opened ahead of the list. Channel? channelById(String channelId) => state.value?.where((channel) => channel.id == channelId).firstOrNull ?? - _openedAhead[channelId]; + ref.read(_openedAheadProvider)[channelId]; bool get hasLoaded => _hasLoaded; @@ -176,7 +174,6 @@ class ChannelsNotifier extends AsyncNotifier> { _memberSnapshotsByChannelId = const {}; _directoryMetas = const []; _hiddenDmIds = const {}; - _openedAhead = const {}; // Retire any in-flight directory request: its response describes the // previous relay or identity and must not reach this scope's state. _refreshCoordinator.retireInFlight(); @@ -185,6 +182,11 @@ class ChannelsNotifier extends AsyncNotifier> { (_hasLoaded ? state.value : null) ?? _restoreSnapshot(pubkey, relayBaseUrl); if (snapshot != null) _hasCachedSnapshot = true; + listenSelf((_, next) { + final listed = next.value; + if (listed == null || !ref.exists(_openedAheadProvider)) return; + ref.read(_openedAheadProvider.notifier).dropListed(listed); + }); _saveSnapshot = pubkey == null ? null : _captureSnapshotWriter(pubkey, relayBaseUrl); @@ -1115,3 +1117,46 @@ class ChannelsNotifier extends AsyncNotifier> { final channelsProvider = AsyncNotifierProvider>( ChannelsNotifier.new, ); + +/// The channel with this id from the list, else as it was opened ahead of the +/// list (see [ChannelsNotifier.rememberOpened]). +final channelByIdProvider = Provider.autoDispose.family( + (ref, channelId) => + ref.watch( + channelsProvider.select( + (channels) => channels.value + ?.where((channel) => channel.id == channelId) + .firstOrNull, + ), + ) ?? + ref.watch(_openedAheadProvider.select((opened) => opened[channelId])), +); + +/// Channels opened from their own metadata before the list had them, as a +/// fast DM open does. Scoped to one relay and identity. Each is dropped once +/// the list has it, so a DM later hidden or left is not found through it. +final _openedAheadProvider = + NotifierProvider<_OpenedAheadNotifier, Map>( + _OpenedAheadNotifier.new, + ); + +class _OpenedAheadNotifier extends Notifier> { + @override + Map build() { + ref.watch(relayConfigProvider.select((config) => config.baseUrl)); + ref.watch(myPubkeyProvider); + return const {}; + } + + void remember(Channel channel) => state = {...state, channel.id: channel}; + + void dropListed(List listed) { + if (state.isEmpty) return; + final ids = {for (final channel in listed) channel.id}; + if (!state.keys.any(ids.contains)) return; + state = { + for (final entry in state.entries) + if (!ids.contains(entry.key)) entry.key: entry.value, + }; + } +} diff --git a/mobile/lib/features/channels/compose_bar/compose_bar_widget.dart b/mobile/lib/features/channels/compose_bar/compose_bar_widget.dart index 591bcdd2e1..a950b7ceff 100644 --- a/mobile/lib/features/channels/compose_bar/compose_bar_widget.dart +++ b/mobile/lib/features/channels/compose_bar/compose_bar_widget.dart @@ -246,8 +246,7 @@ class ComposeBar extends HookConsumerWidget { final currentPubkey = ref.watch(currentPubkeyProvider); final userCache = ref.watch(userCacheProvider); final isDmChannel = - channelsAsync.asData?.value.any((c) => c.id == channelId && c.isDm) ?? - false; + ref.watch(channelByIdProvider(channelId))?.isDm ?? false; // Preload profiles for channel members, mentionable agents, and their // owners so @mention suggestions show names ("managed by …" included). diff --git a/mobile/lib/features/channels/mentions/mention_candidates_provider.dart b/mobile/lib/features/channels/mentions/mention_candidates_provider.dart index a5589d08a4..87273980e6 100644 --- a/mobile/lib/features/channels/mentions/mention_candidates_provider.dart +++ b/mobile/lib/features/channels/mentions/mention_candidates_provider.dart @@ -162,9 +162,7 @@ final mentionCandidatesProvider = Provider.family final channels = channelsAsync.asData?.value ?? const []; final userCache = ref.watch(userCacheProvider); final currentPubkey = ref.watch(currentPubkeyProvider); - final channel = channels - .where((candidate) => candidate.id == args.channelId) - .firstOrNull; + final channel = ref.watch(channelByIdProvider(args.channelId)); // Archived channels take no mentions. if (channel?.isArchived == true) return const []; final directory = channel?.mentionsOutsidePeople ?? false; @@ -260,9 +258,7 @@ IdentityNames mentionPickerNames( /// section 2). Archived channels never search. final mentionSearchFailedProvider = Provider.family .autoDispose((ref, args) { - final channel = (ref.watch(channelsProvider).asData?.value ?? const []) - .where((candidate) => candidate.id == args.channelId) - .firstOrNull; + final channel = ref.watch(channelByIdProvider(args.channelId)); if (channel == null || channel.isArchived || !channel.mentionsOutsidePeople) { diff --git a/mobile/lib/features/channels/message_actions.dart b/mobile/lib/features/channels/message_actions.dart index a0af9e27fe..a6e5180bb6 100644 --- a/mobile/lib/features/channels/message_actions.dart +++ b/mobile/lib/features/channels/message_actions.dart @@ -457,8 +457,7 @@ bool messageActionShowsUnread( required TimelineMessage message, }) { final readerPubkey = ref.read(myPubkeyProvider); - final channels = ref.read(channelsProvider).asData?.value; - final channel = channels?.where((c) => c.id == channelId).firstOrNull; + final channel = ref.read(channelsProvider.notifier).channelById(channelId); final channelCatchUp = readerPubkey != null && channel != null && diff --git a/mobile/lib/features/channels/thread_detail_page.dart b/mobile/lib/features/channels/thread_detail_page.dart index d217640c61..8b36284940 100644 --- a/mobile/lib/features/channels/thread_detail_page.dart +++ b/mobile/lib/features/channels/thread_detail_page.dart @@ -745,12 +745,7 @@ class ThreadDetailPage extends HookConsumerWidget { visible.add(replies[replyIndex]); } final isDm = - ref - .read(channelsProvider) - .value - ?.where((candidate) => candidate.id == channelId) - .firstOrNull - ?.isDm ?? + ref.read(channelsProvider.notifier).channelById(channelId)?.isDm ?? false; final marks = readingMarks( readState: readState, @@ -900,16 +895,7 @@ class ThreadDetailPage extends HookConsumerWidget { return null; }, [settledImeBottomInset.value]); - final channel = - ref.watch( - channelsProvider.select( - (channels) => channels.value - ?.where((candidate) => candidate.id == channelId) - .firstOrNull, - ), - ) ?? - // A DM opened ahead of a reloading list. - ref.read(channelsProvider.notifier).channelById(channelId); + final channel = ref.watch(channelByIdProvider(channelId)); final channelNamesMap = ref.watch(channelIdsByNameProvider); return FrostedScaffold( diff --git a/mobile/test/features/channels/channel_detail_page_test.dart b/mobile/test/features/channels/channel_detail_page_test.dart index ebe243714f..55f02451a2 100644 --- a/mobile/test/features/channels/channel_detail_page_test.dart +++ b/mobile/test/features/channels/channel_detail_page_test.dart @@ -10799,12 +10799,12 @@ void main() { final head = formatTimeline([ _textMsg(id: threadIdThreadDashRoot, pubkey: 'pat', content: 'Hi'), ]).single; + final channels = _FakeChannelsNotifier(const []); await tester.pumpWidget( _buildTestable( messages: [], sender: sender, - // The DM was just opened, so the reloading list does not have it yet. - channelsNotifier: _FakeChannelsNotifier(const [])..rememberOpened(dm), + channelsNotifier: channels, home: ThreadDetailPage( threadHead: head, allMessages: [head], @@ -10815,6 +10815,8 @@ void main() { ), ), ); + // The DM was just opened, so the reloading list does not have it yet. + channels.rememberOpened(dm); await tester.pumpAndSettle(); await tester .widget(find.byType(ComposeBar)) diff --git a/mobile/test/features/channels/channels_page_test.dart b/mobile/test/features/channels/channels_page_test.dart index 804a92e0eb..382ba5e12c 100644 --- a/mobile/test/features/channels/channels_page_test.dart +++ b/mobile/test/features/channels/channels_page_test.dart @@ -3123,6 +3123,7 @@ void main() { addTearDown(tester.view.resetPhysicalSize); addTearDown(tester.view.resetDevicePixelRatio); + final waits = []; const directoryUsers = [ DirectoryUser( pubkey: 'alice', @@ -3148,6 +3149,9 @@ void main() { ) .toList(); }), + channelActionsProvider.overrideWith( + (ref) => _PendingDmActions(waits), + ), ], ), ); @@ -3281,6 +3285,11 @@ void main() { ); expect(searchField.textInputAction, TextInputAction.done); expect(searchField.onSubmitted, isNotNull); + + // Done opens the DM at once instead of waiting on the full channel list. + await tester.testTextInput.receiveAction(TextInputAction.done); + await tester.pump(); + expect(waits, [false]); }); testWidgets('wraps many recipients without overflowing above the keyboard', ( @@ -4042,3 +4051,21 @@ class _CompletedPairingNotifier extends PairingNotifier { @override PairingState build() => const PairingState(status: PairingStatus.success); } + +class _PendingDmActions implements ChannelActions { + _PendingDmActions(this.waits); + + final List waits; + + @override + Future openDm({ + required List pubkeys, + bool waitForChannelList = true, + }) { + waits.add(waitForChannelList); + return Completer().future; + } + + @override + dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation); +} diff --git a/mobile/test/features/channels/channels_provider_test.dart b/mobile/test/features/channels/channels_provider_test.dart index 9be1a620df..77afe3fea1 100644 --- a/mobile/test/features/channels/channels_provider_test.dart +++ b/mobile/test/features/channels/channels_provider_test.dart @@ -2320,6 +2320,83 @@ void main() { ); }); + test('a DM opened ahead of the list is not found after it is left', () async { + final session = _FakeRelaySession( + memberships: [_membership(_channelA, myPk)], + metadata: [_meta(id: _channelA, name: 'Alice', channelType: 'dm')], + ); + final container = _buildContainer(session: session); + addTearDown(container.dispose); + final notifier = container.read(channelsProvider.notifier); + final dm = (await container.read(channelsProvider.future)).single; + + // A fast open finds the DM from its own metadata before the list has it. + session.memberships = []; + await notifier.refresh(); + notifier.rememberOpened(dm); + expect(notifier.channelById(_channelA)?.id, _channelA); + + // Once listed, the list owns it, so leaving drops it everywhere. + session.memberships = [_membership(_channelA, myPk)]; + await notifier.refresh(); + session.memberships = []; + await notifier.refresh(); + expect(container.read(channelsProvider).value, isEmpty); + expect(notifier.channelById(_channelA), isNull); + expect(container.read(channelByIdProvider(_channelA)), isNull); + + // The open's background refresh can list the DM before the open + // remembers it. The list still owns it. + session.memberships = [_membership(_channelA, myPk)]; + await notifier.refresh(); + notifier.rememberOpened(dm); + session.memberships = []; + await notifier.refresh(); + expect(notifier.channelById(_channelA), isNull); + }); + + test('a DM opened ahead of the list stays in its community', () async { + final session = _FakeRelaySession( + memberships: [_membership(_channelA, myPk)], + metadata: [_meta(id: _channelA, name: 'Alice', channelType: 'dm')], + ); + final container = _buildContainer(session: session); + addTearDown(container.dispose); + final notifier = container.read(channelsProvider.notifier); + final dm = (await container.read(channelsProvider.future)).single; + session.memberships = []; + await notifier.refresh(); + notifier.rememberOpened(dm); + final found = []; + final lookup = container.listen( + channelByIdProvider(_channelA), + (_, next) => found.add(next?.id), + fireImmediately: true, + ); + addTearDown(lookup.close); + expect(found, [_channelA]); + + // Neither community lists the DM, yet the open lookup lets it go. + container + .read(relayConfigProvider.notifier) + .update(baseUrl: 'https://community-b.example'); + await container.read(channelsProvider.future); + await _settle(); + expect(lookup.read(), isNull); + expect( + container.read(channelsProvider.notifier).channelById(_channelA), + isNull, + ); + + // Likewise across an identity switch. + notifier.rememberOpened(dm); + expect(lookup.read()?.id, _channelA); + container.read(_testPubkeyProvider.notifier).set('other'); + await container.read(channelsProvider.future); + await _settle(); + expect(lookup.read(), isNull); + }); + test( 'archived kind:39000 metadata sets Channel.isArchived (covers TTL auto-archive)', () async { diff --git a/mobile/test/features/channels/compose_bar_test.dart b/mobile/test/features/channels/compose_bar_test.dart index 63c77d5ddc..09f48a5869 100644 --- a/mobile/test/features/channels/compose_bar_test.dart +++ b/mobile/test/features/channels/compose_bar_test.dart @@ -4724,6 +4724,7 @@ void main() { final pat = '1' * 64; List? sentRecipients; List>? sentTags; + var searches = 0; await tester.pumpWidget( _buildComposeBar( uploadService: _testUploadService(signer.nsec), @@ -4744,6 +4745,11 @@ void main() { channelType: 'dm', participantPubkeys: [signer.public, pat], ), + searchPeople: (query) async { + if (query != 'Mary J') return const []; + if (++searches == 1) throw StateError('relay down'); + return [maryJane]; + }, onSend: (_, recipients, {mediaTags = const >[]}) async { sentRecipients = recipients; @@ -4752,6 +4758,21 @@ void main() { ), ); await _expandComposer(tester); + // The picker searches the directory for people outside the DM, as it + // does once the list has the DM, and keeps a failed search's retry. + await tester.enterText(find.byType(TextField), '@Mary J'); + await tester.pump(); + await tester.pump(mentionSearchDebounce * 2); + expect( + find.byKey(const ValueKey('mention-search-error')), + findsOneWidget, + ); + await tester.tap(find.text('Retry')); + await tester.pump(); + await tester.pump(mentionSearchDebounce * 2); + expect(find.text('Mary Jane'), findsOneWidget); + expect(find.text('not in DM'), findsOneWidget); + await tester.enterText(find.byType(TextField), '@Pat'); await tester.pump(); await tester.pumpAndSettle(); diff --git a/mobile/test/features/profile/user_profile_sheet_message_test.dart b/mobile/test/features/profile/user_profile_sheet_message_test.dart index 1ad80bd592..ebdae37a70 100644 --- a/mobile/test/features/profile/user_profile_sheet_message_test.dart +++ b/mobile/test/features/profile/user_profile_sheet_message_test.dart @@ -31,6 +31,7 @@ void main() { tester.view.devicePixelRatio = 1; addTearDown(tester.view.reset); final opened = >[]; + final waits = []; SharedPreferences.setMockInitialValues({}); final prefs = await SharedPreferences.getInstance(); await tester.pumpWidget( @@ -42,7 +43,7 @@ void main() { }), ), channelActionsProvider.overrideWith( - (ref) => _FakeDmActions(dm, opened), + (ref) => _FakeDmActions(dm, opened, waits), ), savedPrefsProvider.overrideWithValue(prefs), // Solo organization is still loading. @@ -70,6 +71,8 @@ void main() { expect(opened, [ [peer], ]); + // The page opens at once instead of waiting on the full channel list. + expect(waits, [false]); expect( tester.widget(find.byType(ChannelDetailPage)).channel, dm, @@ -92,10 +95,11 @@ class _UserCache extends UserCacheNotifier { } class _FakeDmActions implements ChannelActions { - _FakeDmActions(this.dm, this.opened); + _FakeDmActions(this.dm, this.opened, this.waits); final Channel dm; final List> opened; + final List waits; @override Future openDm({ @@ -103,6 +107,7 @@ class _FakeDmActions implements ChannelActions { bool waitForChannelList = true, }) async { opened.add(pubkeys); + waits.add(waitForChannelList); return dm; } diff --git a/mobile/test/features/search/search_page_test.dart b/mobile/test/features/search/search_page_test.dart index 11ee833fcb..4f4aebd03d 100644 --- a/mobile/test/features/search/search_page_test.dart +++ b/mobile/test/features/search/search_page_test.dart @@ -131,6 +131,7 @@ void main() { isMember: true, ); final opened = >[]; + final waits = []; SharedPreferences.setMockInitialValues({}); final prefs = await SharedPreferences.getInstance(); await tester.pumpWidget( @@ -152,7 +153,7 @@ void main() { profileProvider.overrideWith(() => _FakeProfileNotifier()), channelsProvider.overrideWith(() => _FakeChannelsNotifier()), channelActionsProvider.overrideWith( - (ref) => _FakeDmActions(dm, opened), + (ref) => _FakeDmActions(dm, opened, waits), ), savedPrefsProvider.overrideWithValue(prefs), // Solo organization is still loading. @@ -169,6 +170,8 @@ void main() { expect(opened, [ ['maya'], ]); + // The page opens at once instead of waiting on the full channel list. + expect(waits, [false]); final page = tester.widget( find.byType(ChannelDetailPage), ); @@ -999,10 +1002,11 @@ class _FakeChannelsNotifier extends ChannelsNotifier { } class _FakeDmActions implements ChannelActions { - _FakeDmActions(this.dm, this.opened); + _FakeDmActions(this.dm, this.opened, this.waits); final Channel dm; final List> opened; + final List waits; @override Future openDm({ @@ -1010,6 +1014,7 @@ class _FakeDmActions implements ChannelActions { bool waitForChannelList = true, }) async { opened.add(pubkeys); + waits.add(waitForChannelList); return dm; } From 0a362eef150ab16453993d442a8037baad766797 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Wed, 28 Oct 2026 15:59:19 -0400 Subject: [PATCH 2/2] perf(db,relay): bound thread participant scans and coalesce live summaries (#8635) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary A thread summary's `participants` (up to 10 recent authors, used for the facepile) was computed by aggregating **every** reply in the thread. The live `kind:39005` emit runs that aggregate on the writer after every reply insert and every threaded delete. Each summary therefore cost O(thread), recomputed after every reply, so one very large thread could saturate the writer. ### Scale and impact A single long-running thread can grow to hundreds of thousands of replies and take several new replies per second. Each of those replies then pays a whole-thread aggregate on the writer, so writer CPU follows the reply rate of the largest active threads, and the read-side participant queries (channel window, thread root summaries) slow down the same way. Synthetic reproduction (PostgreSQL 17, the repo's `schema/schema.sql`, default `shared_buffers`): one channel with 200k background messages and three threads of 450k, 6k and 300 replies from a few authors with ~800-byte bodies. `EXPLAIN (ANALYZE, BUFFERS)` of the participant query, warm cache (runs 2–3 of 3): | Thread | Before (whole-thread aggregate) | After (newest 1000 replies) | |---|---|---| | 450k replies | ~7.6 s, ~117k buffers read, ~16k temp blocks spilled | ~6.8 ms, ~4k buffers | | 6k replies | ~25 ms, ~24k buffers | ~7 ms, ~4k buffers | | 300 replies | ~2.6 ms, ~1.2k buffers | ~2.6 ms, ~1.2k buffers | Cost now stops growing at 1000 replies, and small threads cost the same as before. With coalescing, a busy thread runs this query at most about once per second per pod instead of once per reply. ### Change 1. **Bound the participant scan to the newest 1000 replies.** All three participant readers now share one SQL fragment (`recent_thread_participants_sql!`): - `get_thread_summary` (the live 39005 emit) - the channel-window batch - `resolve_thread_root_summaries` The fragment walks `idx_thread_metadata_window` newest-first with `LIMIT 1000`, then joins `events` only for those rows, to get `pubkey` and drop deleted replies. Deleted replies keep their `thread_metadata` row, so they count toward the 1000 but contribute no author. It aggregates by author and returns the 10 most recently active. Cost no longer grows with thread size. Desktop and mobile show the 3 most recent participants and the forum view shows up to 10, so an author who hasn't posted in the last 1000 replies is not visible in any client today. `NIP-CW`, `docs/bridge-channel-window.md` and the `ThreadSummary::participants` doc now state the bound, deleted replies included. 2. **Coalesce live summary emits per thread root.** `emit_live_thread_summary` goes through a new `CoalescedRefresh` keyed by `(community, root)`: - A quiet thread emits immediately, as before. - On a busy thread, emits are spaced at least 1s apart per pod. A trailing emit covers the rest of the burst and reads state committed after its last mutation. - Because emits are at least 1s apart, consecutive overlays from one pod always get strictly increasing `created_at`, so the client's newest-wins replacement is never tied. - A panicking or cancelled refresh releases its key, including a task dropped before its first poll: the release guard is built before the spawn. Each run is bounded by a 30s timeout so a hung refresh can't wedge it. ### What got simpler - Three hand-written copies of the participant aggregate are now one fragment. - `get_thread_summary` is one statement instead of two round trips. - The channel-window batch dropped its `ROW_NUMBER()` window function and the `GROUP BY root, pubkey` over every reply of every root on the page. ### Related issue None found. ### Testing - `participant_summaries_read_only_the_newest_replies` (buzz-db, Postgres) seeds a thread with 1001 replies: - an early author, pushed just outside the window - a filler author - a deleted reply - a latest author All three readers must return `[latest, filler]`. Mutant check: removing the `LIMIT` (whole-thread aggregate) turns it red. - `coalesced_refresh` unit tests (paused tokio clock): - A 100-request burst runs once immediately and once trailing, then frees the key. - Separate keys run independently. - A panicking refresh doesn't wedge its key. - A hung refresh times out and releases its key. Mutant check: removing the timeout turns it red. - A request that arrives while a run is in flight gets one trailing run, and that run reads the state written mid-run. Mutant check: dropping the rerun flag turns it red. - A task dropped before its first poll (runtime shut down) still releases its key. Mutant check: building the guard inside the task turns it red. These run in `just test-unit` (added to the buzz-relay `--lib` filter and `scripts/run-tests.sh`). - Blox gate: `cargo fmt --check`, clippy `-D warnings` (buzz-db, buzz-relay), and the full buzz-db and buzz-relay Postgres nextest lanes. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Signed-off-by: Luke Tornquist Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-authored-by: Claude Opus 5.5 --- Justfile | 2 +- crates/buzz-db/src/store/thread.rs | 277 +++++++++++------ crates/buzz-db/tests/observability_source.rs | 1 - .../src/handlers/coalesced_refresh.rs | 287 ++++++++++++++++++ crates/buzz-relay/src/handlers/mod.rs | 2 + .../buzz-relay/src/handlers/side_effects.rs | 170 ++++++----- crates/buzz-relay/src/state.rs | 19 ++ docs/bridge-channel-window.md | 2 +- docs/nips/NIP-CW.md | 2 +- scripts/run-tests.sh | 3 + 10 files changed, 591 insertions(+), 174 deletions(-) create mode 100644 crates/buzz-relay/src/handlers/coalesced_refresh.rs diff --git a/Justfile b/Justfile index d6df05fcaa..18b444f5b4 100644 --- a/Justfile +++ b/Justfile @@ -543,7 +543,7 @@ test-unit: # never connects. # `^config::` is anchored so it does not also select nip_fi_config. cargo nextest run -p buzz-relay --lib --bin buzz-relay \ - -E 'test(/^api::admin::/) + test(/^push_runtime::/) + test(/^operator_listener::tests::/) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^handlers::system_channels::tests::/) + test(/^nip11::tests::/) + test(/^storage_sweep::tests::/) + test(/^nip_fi_core::tests::/) + test(/^nip_fi_http::tests::/) + test(/^nip_fi_config::tests::/) + test(/^nip_fi_resolver::tests::/) + test(/^readiness::tests::/) + test(/^router::tests::/) + test(/^api::parse_query_tests::/) + test(/^api::profiles::/) + test(/^api::git::transport::off_mode_precedence_tests::/) + test(/^audio::join::tests::/) + test(/^audio::handler::tests::/) + test(/^nip_fi_gate::tests::/) + test(/^nip_fi_session::tests::/) + test(/^nip_fi_shadow::tests::/) + test(/^nip_fi_shadow_session::tests::/) + test(/^startup_steps::tests::/) + test(/^mesh_boot::tests::/) + test(/^telemetry::tests::/) + test(/^config::tests::/) + test(=state::tests::neither_a_confirmed_inactive_community_nor_a_failed_lookup_admits_the_socket) + test(=state::tests::periodic_revalidation_disconnects_inside_the_fenced_callback) + test(=state::tests::bare_community_disconnect_fails_closed_when_the_fence_is_unavailable) + test(=connection::tests::send_loop_sends_policy_close_when_community_is_archived) + test(=state::tests::community_disconnect_then_nip_fi_keeps_community_deleted_reason) + test(=state::tests::disconnect_community_wins_reason_losing_nip_fi_does_not_enqueue_frame) + test(=handlers::req::tests::timed_out_historical_read_deregisters_before_closed) + test(=handlers::req::tests::superseded_timeout_leaves_replacement_intact) + test(=handlers::req::tests::search_claim_retires_live_and_yields_to_replacement) + test(=handlers::req::tests::concurrent_claims_and_stale_teardowns_keep_the_last_owner) + test(=handlers::req::tests::timeout_closed_is_emitted_before_a_replacement_can_claim) + test(=handlers::req::tests::revoke_then_replacement_keeps_replacement_whole) + test(=handlers::req::tests::claims_after_connection_cleanup_are_refused) + test(=handlers::req::tests::dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::revoke_dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::count_fallback_covers_filters_that_can_match_private_deletions) + test(=handlers::req::tests::group_state_t_and_creator_tags_reach_sql_only_for_group_state_kinds) + test(=api::bridge::postgres_tests::bridge_search_mode_extension_accepts_prefix_snake_or_camel_case) + test(=api::bridge::postgres_tests::bridge_search_sort_extension_selects_newest_only_when_named) + (kind(bin) & (test(/^tests::/) + test(/^composition_tests::/) + test(/^env_filter_tests::/)) - test(/^tests::postgres_tests::/))' + -E 'test(/^api::admin::/) + test(/^push_runtime::/) + test(/^operator_listener::tests::/) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^handlers::coalesced_refresh::tests::/) + test(/^handlers::system_channels::tests::/) + test(/^nip11::tests::/) + test(/^storage_sweep::tests::/) + test(/^nip_fi_core::tests::/) + test(/^nip_fi_http::tests::/) + test(/^nip_fi_config::tests::/) + test(/^nip_fi_resolver::tests::/) + test(/^readiness::tests::/) + test(/^router::tests::/) + test(/^api::parse_query_tests::/) + test(/^api::profiles::/) + test(/^api::git::transport::off_mode_precedence_tests::/) + test(/^audio::join::tests::/) + test(/^audio::handler::tests::/) + test(/^nip_fi_gate::tests::/) + test(/^nip_fi_session::tests::/) + test(/^nip_fi_shadow::tests::/) + test(/^nip_fi_shadow_session::tests::/) + test(/^startup_steps::tests::/) + test(/^mesh_boot::tests::/) + test(/^telemetry::tests::/) + test(/^config::tests::/) + test(=state::tests::neither_a_confirmed_inactive_community_nor_a_failed_lookup_admits_the_socket) + test(=state::tests::periodic_revalidation_disconnects_inside_the_fenced_callback) + test(=state::tests::bare_community_disconnect_fails_closed_when_the_fence_is_unavailable) + test(=connection::tests::send_loop_sends_policy_close_when_community_is_archived) + test(=state::tests::community_disconnect_then_nip_fi_keeps_community_deleted_reason) + test(=state::tests::disconnect_community_wins_reason_losing_nip_fi_does_not_enqueue_frame) + test(=handlers::req::tests::timed_out_historical_read_deregisters_before_closed) + test(=handlers::req::tests::superseded_timeout_leaves_replacement_intact) + test(=handlers::req::tests::search_claim_retires_live_and_yields_to_replacement) + test(=handlers::req::tests::concurrent_claims_and_stale_teardowns_keep_the_last_owner) + test(=handlers::req::tests::timeout_closed_is_emitted_before_a_replacement_can_claim) + test(=handlers::req::tests::revoke_then_replacement_keeps_replacement_whole) + test(=handlers::req::tests::claims_after_connection_cleanup_are_refused) + test(=handlers::req::tests::dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::revoke_dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::count_fallback_covers_filters_that_can_match_private_deletions) + test(=handlers::req::tests::group_state_t_and_creator_tags_reach_sql_only_for_group_state_kinds) + test(=api::bridge::postgres_tests::bridge_search_mode_extension_accepts_prefix_snake_or_camel_case) + test(=api::bridge::postgres_tests::bridge_search_sort_extension_selects_newest_only_when_named) + (kind(bin) & (test(/^tests::/) + test(/^composition_tests::/) + test(/^env_filter_tests::/)) - test(/^tests::postgres_tests::/))' # Note on audio::join::tests scope: the full suite is infra-free (no DB, # no Redis). The infra-free audio/FI regression witnesses — bootstrap # ordering barrier, CommitConfirmed arm, pending-close invisibility, diff --git a/crates/buzz-db/src/store/thread.rs b/crates/buzz-db/src/store/thread.rs index 0e4c9d9375..7c617aba6a 100644 --- a/crates/buzz-db/src/store/thread.rs +++ b/crates/buzz-db/src/store/thread.rs @@ -18,6 +18,60 @@ use crate::{ ReadSessionInner, RouteDecision, RoutePredicate, }; +/// [`THREAD_PARTICIPANT_SCAN_LIMIT`] as a literal, so SQL can `concat!` it. +macro_rules! thread_participant_scan_limit { + () => { + 1000 + }; +} + +/// How many of a thread's newest replies a participant summary reads. +/// +/// Participants are the thread's most recently active distinct authors, so +/// the newest replies decide them. Aggregating the whole thread instead made +/// every summary cost O(thread), recomputed after every reply, so one very +/// large thread could saturate the writer. An author silent for this many +/// replies drops out of the facepile, which shows at most 10. +pub const THREAD_PARTICIPANT_SCAN_LIMIT: i64 = thread_participant_scan_limit!(); + +/// Correlated `ARRAY(...)` of up to 10 participant pubkeys, most recent +/// first, for the thread rooted at `root.event_id` in community `$1`. +/// +/// Walks `idx_thread_metadata_window` newest-first and stops after +/// [`THREAD_PARTICIPANT_SCAN_LIMIT`] rows, so cost is bounded by the window, +/// not the thread. The one argument is an extra (possibly empty) predicate on +/// the window's `tm` rows. A macro so callers stay `&'static str` via +/// `concat!`. +macro_rules! recent_thread_participants_sql { + ($extra:literal) => { + concat!( + r#"ARRAY( + SELECT e.pubkey + FROM ( + SELECT tm.event_created_at, tm.event_id + FROM thread_metadata tm + WHERE tm.community_id = $1 + AND tm.root_event_id = root.event_id "#, + $extra, + r#" + ORDER BY tm.event_created_at DESC, tm.event_id ASC + LIMIT "#, + thread_participant_scan_limit!(), + r#" + ) recent + JOIN events e + ON e.community_id = $1 + AND e.created_at = recent.event_created_at + AND e.id = recent.event_id + WHERE e.deleted_at IS NULL + GROUP BY e.pubkey + ORDER BY MAX(e.created_at) DESC, e.pubkey ASC + LIMIT 10 + )"# + ) + }; +} + // -- Structs ------------------------------------------------------------------ /// A single reply within a thread, joined with event content. @@ -56,7 +110,9 @@ pub struct ThreadSummary { pub descendant_count: i32, /// Timestamp of the most recent reply in the thread. pub last_reply_at: Option>, - /// Compressed public keys of all participants who have replied. + /// Up to 10 distinct reply authors, most recently active first, drawn + /// from the newest [`THREAD_PARTICIPANT_SCAN_LIMIT`] replies. Deleted + /// replies count toward that window but contribute no author. pub participants: Vec>, } @@ -467,63 +523,29 @@ pub async fn get_thread_summary( crate::observability::WriterOperation::EventWrite, ) .await?; - let row = sqlx::query( - r#" - SELECT reply_count, descendant_count, last_reply_at - FROM thread_metadata - WHERE community_id = $1 AND event_id = $2 + let row = sqlx::query(concat!( + "SELECT root.reply_count, root.descendant_count, root.last_reply_at, ", + recent_thread_participants_sql!(""), + r#" AS participants + FROM thread_metadata root + WHERE root.community_id = $1 AND root.event_id = $2 LIMIT 1 "#, - ) + )) .bind(community_id.as_uuid()) .bind(event_id) .fetch_optional(&mut *connection) .await?; - let row = match row { - Some(r) => r, - None => return Ok(None), - }; - - let reply_count: i32 = row.try_get("reply_count")?; - let descendant_count: i32 = row.try_get("descendant_count")?; - let last_reply_at: Option> = row.try_get("last_reply_at")?; - - // Collect distinct participant pubkeys from the thread, most recent first. - let participant_rows = sqlx::query( - r#" - SELECT pubkey FROM ( - SELECT DISTINCT e.pubkey, MAX(e.created_at) AS last_seen - FROM thread_metadata tm - JOIN events e - ON e.community_id = tm.community_id - AND e.created_at = tm.event_created_at - AND e.id = tm.event_id - WHERE tm.community_id = $1 - AND tm.root_event_id = $2 - AND e.deleted_at IS NULL - GROUP BY e.pubkey - ) sub - ORDER BY last_seen DESC - LIMIT 10 - "#, - ) - .bind(community_id.as_uuid()) - .bind(event_id) - .fetch_all(&mut *connection) - .await?; - - let participants: Vec> = participant_rows - .into_iter() - .map(|r| r.try_get::, _>("pubkey")) - .collect::>()?; - - Ok(Some(ThreadSummary { - reply_count, - descendant_count, - last_reply_at, - participants, - })) + row.map(|row| { + Ok(ThreadSummary { + reply_count: row.try_get("reply_count")?, + descendant_count: row.try_get("descendant_count")?, + last_reply_at: row.try_get("last_reply_at")?, + participants: row.try_get("participants")?, + }) + }) + .transpose() } /// Fetch one channel window: top-level rows (depth = 0, missing metadata, or @@ -692,39 +714,21 @@ pub(crate) async fn get_channel_window_on( } // Batch participants for every row with thread activity — one query for - // the whole window instead of a per-root fan-out. Same shape and 10-cap - // as get_thread_summary. + // the whole window instead of a per-root fan-out, each root bounded the + // same way as get_thread_summary. let roots: Vec> = rows .iter() .filter(|r| r.thread_summary.is_some()) .map(|r| r.stored_event.event.id.as_bytes().to_vec()) .collect(); if !roots.is_empty() { - let participant_rows = sqlx::query( - r#" - SELECT root_event_id, pubkey FROM ( - SELECT - tm.root_event_id, - e.pubkey, - MAX(e.created_at) AS last_seen, - ROW_NUMBER() OVER ( - PARTITION BY tm.root_event_id - ORDER BY MAX(e.created_at) DESC - ) AS rn - FROM thread_metadata tm - JOIN events e - ON e.community_id = tm.community_id - AND e.created_at = tm.event_created_at - AND e.id = tm.event_id - WHERE tm.community_id = $1 - AND tm.root_event_id = ANY($2) - AND e.deleted_at IS NULL - GROUP BY tm.root_event_id, e.pubkey - ) sub - WHERE rn <= 10 - ORDER BY root_event_id, rn + let participant_rows = sqlx::query(concat!( + "SELECT root.event_id, ", + recent_thread_participants_sql!(""), + r#" AS participants + FROM unnest($2::bytea[]) AS root(event_id) "#, - ) + )) .bind(community_id.as_uuid()) .bind(&roots) .fetch_all(&mut *conn) @@ -733,9 +737,7 @@ pub(crate) async fn get_channel_window_on( let mut by_root: std::collections::HashMap, Vec>> = std::collections::HashMap::new(); for row in participant_rows { - let root: Vec = row.try_get("root_event_id")?; - let pubkey: Vec = row.try_get("pubkey")?; - by_root.entry(root).or_default().push(pubkey); + by_root.insert(row.try_get("event_id")?, row.try_get("participants")?); } for row in &mut rows { if let Some(summary) = &mut row.thread_summary { @@ -982,18 +984,10 @@ impl Db { crate::observability::WriterOperation::SubscriptionHistory, ) .await?; - let rows = sqlx::query( - r#" - SELECT root.event_id, root.reply_count, root.descendant_count, root.last_reply_at, - ARRAY( - SELECT e.pubkey - FROM thread_metadata tm - JOIN events e ON e.community_id = tm.community_id - AND e.created_at = tm.event_created_at AND e.id = tm.event_id - WHERE tm.community_id = root.community_id AND tm.root_event_id = root.event_id - AND tm.channel_id = root.channel_id AND e.deleted_at IS NULL - GROUP BY e.pubkey ORDER BY MAX(e.created_at) DESC LIMIT 10 - ) AS participants + let rows = sqlx::query(concat!( + "SELECT root.event_id, root.reply_count, root.descendant_count, root.last_reply_at, ", + recent_thread_participants_sql!("AND tm.channel_id = root.channel_id"), + r#" AS participants FROM thread_metadata root WHERE root.community_id = $1 AND root.channel_id = $2 AND root.event_id IN ( @@ -1003,7 +997,7 @@ impl Db { AND target.event_id = ANY($3) AND target.parent_event_id IS NOT NULL ) "#, - ) + )) .bind(community_id.as_uuid()) .bind(channel_id) .bind(target_ids) @@ -2320,4 +2314,101 @@ mod postgres_tests { "reply-less row carries no summary" ); } + + /// Every participant reader (live summary, channel window, deleted-reply + /// root resolution) aggregates only the newest + /// [`THREAD_PARTICIPANT_SCAN_LIMIT`] replies, most recent author first, + /// skipping deleted replies. An author who last posted before that window + /// drops out; a whole-thread aggregate would still return them. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn participant_summaries_read_only_the_newest_replies() { + let pool = setup_pool().await; + let db = Db::from_pool(pool.clone()); + let early = Keys::generate(); + let steady = Keys::generate(); + let deleted_author = Keys::generate(); + let latest = Keys::generate(); + let (channel, community) = create_test_channel( + &pool, + &format!("participant-window-{}", Uuid::new_v4()), + ChannelType::Stream, + ChannelVisibility::Open, + None, + steady.public_key().to_bytes().as_slice(), + None, + ) + .await + .expect("create channel"); + + let root = make_stream_event(&steady, "root"); + insert_root(&pool, community, channel.id, &root).await; + let base = root.created_at.as_secs(); + let reply_at = |keys: &Keys, offset: u64| { + EventBuilder::new(Kind::Custom(9), format!("reply-{offset}")) + .custom_created_at(nostr::Timestamp::from(base + offset)) + .sign_with_keys(keys) + .expect("sign reply") + }; + + // `early` posts first, then `steady` fills the whole window after it. + let early_reply = reply_at(&early, 1); + insert_reply(&pool, community, channel.id, &root, &early_reply, false).await; + let window = THREAD_PARTICIPANT_SCAN_LIMIT as u64; + for offset in 2..window { + let reply = reply_at(&steady, offset); + insert_reply(&pool, community, channel.id, &root, &reply, false).await; + } + let deleted_reply = reply_at(&deleted_author, window); + insert_reply(&pool, community, channel.id, &root, &deleted_reply, false).await; + let latest_reply = reply_at(&latest, window + 1); + insert_reply(&pool, community, channel.id, &root, &latest_reply, false).await; + db.soft_delete_event_and_update_thread( + community, + deleted_reply.id.as_bytes(), + Some(root.id.as_bytes()), + Some(root.id.as_bytes()), + ) + .await + .expect("delete reply"); + + let expected = vec![ + latest.public_key().to_bytes().to_vec(), + steady.public_key().to_bytes().to_vec(), + ]; + + let summary = get_thread_summary(&pool, community, root.id.as_bytes()) + .await + .expect("thread summary") + .expect("root has thread metadata"); + assert_eq!(summary.participants, expected, "live summary"); + + let window = get_channel_window(&pool, community, channel.id, 50, None, None) + .await + .expect("fetch window"); + let row = window + .rows + .iter() + .find(|r| r.stored_event.event.id == root.id) + .expect("root row present"); + assert_eq!( + row.thread_summary + .as_ref() + .expect("root has a summary") + .participants, + expected, + "channel window" + ); + + let resolved = db + .resolve_thread_root_summaries( + community, + channel.id, + &[deleted_reply.id.as_bytes().to_vec()], + ) + .await + .expect("resolve root"); + assert_eq!(resolved.len(), 1); + assert_eq!(resolved[0].1.participants, expected, "resolved root"); + } } diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index cc11238d4f..fcf4e6ddb1 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -409,7 +409,6 @@ fn p0_pool_acquisitions_use_typed_operation_pairs_without_other() { .0; assert!(thread_summary.contains("WriterOperation::EventWrite")); assert!(thread_summary.contains("fetch_optional(&mut *connection)")); - assert!(thread_summary.contains("fetch_all(&mut *connection)")); assert!(!thread_summary.contains("fetch_optional(pool)")); assert!(!thread_summary.contains("fetch_all(pool)")); diff --git a/crates/buzz-relay/src/handlers/coalesced_refresh.rs b/crates/buzz-relay/src/handlers/coalesced_refresh.rs new file mode 100644 index 0000000000..0110af2fa0 --- /dev/null +++ b/crates/buzz-relay/src/handlers/coalesced_refresh.rs @@ -0,0 +1,287 @@ +//! Per-key single-flight refresh with a minimum interval. +//! +//! A refresh for a key runs at once when none is in flight. Requests that +//! arrive while one is running or cooling down collapse into one trailing +//! rerun, so a key refreshes at most once per interval per process and the +//! last request is always covered by a refresh that started after it. Each +//! run is bounded by a timeout, so a hung refresh can't wedge its key. + +use std::future::Future; +use std::hash::Hash; +use std::sync::Arc; +use std::time::Duration; + +use dashmap::mapref::entry::Entry; +use dashmap::DashMap; + +/// Coalesces refresh requests per key. The map holds only keys with a +/// refresh in flight; the value is "another request arrived since it began". +pub struct CoalescedRefresh { + pending: DashMap, + min_interval: Duration, + run_timeout: Duration, +} + +impl CoalescedRefresh +where + K: Eq + Hash + Clone + Send + Sync + 'static, +{ + /// A coalescer that refreshes each key at most once per `min_interval`, + /// abandoning any single run that takes longer than `run_timeout`. + pub fn new(min_interval: Duration, run_timeout: Duration) -> Self { + Self { + pending: DashMap::new(), + min_interval, + run_timeout, + } + } + + /// Request a refresh of `key`. Runs `refresh` on a spawned task now when + /// `key` is idle; otherwise marks the in-flight refresh to run once more + /// after its interval elapses. + pub fn request(self: &Arc, key: K, refresh: F) + where + F: Fn() -> Fut + Send + 'static, + Fut: Future + Send, + { + match self.pending.entry(key.clone()) { + Entry::Occupied(mut rerun) => { + *rerun.get_mut() = true; + return; + } + Entry::Vacant(slot) => { + slot.insert(false); + } + } + // Frees the key if the task panics or is cancelled, including before + // its first poll, so a failed refresh can't block the key forever. + let mut release = Release { + owner: Arc::clone(self), + key: Some(key.clone()), + }; + tokio::spawn(async move { + let this = Arc::clone(&release.owner); + loop { + if tokio::time::timeout(this.run_timeout, refresh()) + .await + .is_err() + { + tracing::warn!( + timeout_ms = this.run_timeout.as_millis() as u64, + "coalesced refresh timed out" + ); + } + tokio::time::sleep(this.min_interval).await; + if !this.rerun_or_release(&key) { + break; + } + } + // Finished normally: rerun_or_release already freed the key. + release.key.take(); + }); + } + + /// Clear the rerun flag and report it, or free the key when no request + /// arrived. Atomic with [`Self::request`]'s entry check, so a request + /// either sets the flag before this reads it or finds the key free. + fn rerun_or_release(&self, key: &K) -> bool { + match self.pending.entry(key.clone()) { + Entry::Occupied(mut entry) if *entry.get() => { + *entry.get_mut() = false; + true + } + Entry::Occupied(entry) => { + entry.remove(); + false + } + Entry::Vacant(_) => false, + } + } +} + +struct Release { + owner: Arc>, + key: Option, +} + +impl Drop for Release { + fn drop(&mut self) { + if let Some(key) = self.key.take() { + self.owner.pending.remove(&key); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; + + const INTERVAL: Duration = Duration::from_secs(1); + const TIMEOUT: Duration = Duration::from_secs(5); + + fn counting(runs: &Arc) -> impl Fn() -> std::future::Ready<()> + Send + 'static { + let runs = Arc::clone(runs); + move || { + runs.fetch_add(1, Ordering::SeqCst); + std::future::ready(()) + } + } + + /// Let spawned tasks run up to their next timer. + async fn settle() { + for _ in 0..10 { + tokio::task::yield_now().await; + } + } + + #[tokio::test(start_paused = true)] + async fn burst_runs_once_now_and_once_trailing_then_frees_the_key() { + let refresh = Arc::new(CoalescedRefresh::new(INTERVAL, TIMEOUT)); + let runs = Arc::new(AtomicUsize::new(0)); + + for _ in 0..100 { + refresh.request("root", counting(&runs)); + } + settle().await; + assert_eq!(runs.load(Ordering::SeqCst), 1, "first request runs at once"); + + tokio::time::advance(INTERVAL).await; + settle().await; + assert_eq!( + runs.load(Ordering::SeqCst), + 2, + "the burst gets one trailing run" + ); + + tokio::time::advance(INTERVAL).await; + settle().await; + assert_eq!(runs.load(Ordering::SeqCst), 2, "no request, no rerun"); + assert!(refresh.pending.is_empty(), "idle key is released"); + + refresh.request("root", counting(&runs)); + settle().await; + assert_eq!( + runs.load(Ordering::SeqCst), + 3, + "a later request runs at once" + ); + } + + #[tokio::test(start_paused = true)] + async fn request_during_a_run_gets_a_trailing_run_that_sees_its_state() { + let refresh = Arc::new(CoalescedRefresh::new(INTERVAL, TIMEOUT)); + let state = Arc::new(AtomicUsize::new(1)); + let seen = Arc::new(std::sync::Mutex::new(Vec::new())); + let gate = Arc::new(tokio::sync::Semaphore::new(0)); + // Records the state it reads, then holds until the gate opens. + let reader = { + let (state, seen, gate) = (Arc::clone(&state), Arc::clone(&seen), Arc::clone(&gate)); + move || { + let (state, seen, gate) = + (Arc::clone(&state), Arc::clone(&seen), Arc::clone(&gate)); + async move { + seen.lock().unwrap().push(state.load(Ordering::SeqCst)); + let _permit = gate.acquire().await; + } + } + }; + + refresh.request("root", reader.clone()); + settle().await; + assert_eq!(*seen.lock().unwrap(), [1], "the first run read state 1"); + + state.store(2, Ordering::SeqCst); + refresh.request("root", reader.clone()); + settle().await; + assert_eq!( + *seen.lock().unwrap(), + [1], + "no second run while one is in flight" + ); + + gate.add_permits(1); + settle().await; + tokio::time::advance(INTERVAL).await; + settle().await; + assert_eq!( + *seen.lock().unwrap(), + [1, 2], + "the trailing run reads the state written mid-run" + ); + + tokio::time::advance(INTERVAL).await; + settle().await; + assert_eq!(*seen.lock().unwrap(), [1, 2], "no further run"); + assert!(refresh.pending.is_empty(), "idle key is released"); + } + + #[test] + fn a_task_dropped_before_its_first_poll_releases_the_key() { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_time() + .build() + .unwrap(); + let refresh = Arc::new(CoalescedRefresh::new(INTERVAL, TIMEOUT)); + let runs = Arc::new(AtomicUsize::new(0)); + // `block_on` returns without yielding, so the spawned task is never + // polled; dropping the runtime drops it unpolled. + runtime.block_on(async { refresh.request("root", counting(&runs)) }); + assert!(!refresh.pending.is_empty(), "the queued task holds the key"); + + drop(runtime); + assert_eq!(runs.load(Ordering::SeqCst), 0, "the task never ran"); + assert!( + refresh.pending.is_empty(), + "an unpolled task must not wedge the key" + ); + } + + #[tokio::test(start_paused = true)] + async fn keys_refresh_independently() { + let refresh = Arc::new(CoalescedRefresh::new(INTERVAL, TIMEOUT)); + let runs = Arc::new(AtomicUsize::new(0)); + + refresh.request("a", counting(&runs)); + refresh.request("b", counting(&runs)); + settle().await; + assert_eq!(runs.load(Ordering::SeqCst), 2); + } + + #[tokio::test(start_paused = true)] + async fn panicking_refresh_releases_the_key() { + let refresh = Arc::new(CoalescedRefresh::new(INTERVAL, TIMEOUT)); + refresh.request("root", || async { panic!("refresh failed") }); + settle().await; + assert!(refresh.pending.is_empty(), "panic must not wedge the key"); + + let runs = Arc::new(AtomicUsize::new(0)); + refresh.request("root", counting(&runs)); + settle().await; + assert_eq!(runs.load(Ordering::SeqCst), 1); + } + + #[tokio::test(start_paused = true)] + async fn hung_refresh_times_out_and_releases_the_key() { + let refresh = Arc::new(CoalescedRefresh::new(INTERVAL, TIMEOUT)); + refresh.request("root", std::future::pending::<()>); + settle().await; + assert!(!refresh.pending.is_empty(), "the hung run holds the key"); + + // Step one timer at a time: the cooldown sleep is armed only after + // the timeout fires, so a single jump would overshoot it. + tokio::time::advance(TIMEOUT).await; + settle().await; + tokio::time::advance(INTERVAL).await; + settle().await; + assert!( + refresh.pending.is_empty(), + "a hung run must not wedge the key" + ); + + let runs = Arc::new(AtomicUsize::new(0)); + refresh.request("root", counting(&runs)); + settle().await; + assert_eq!(runs.load(Ordering::SeqCst), 1); + } +} diff --git a/crates/buzz-relay/src/handlers/mod.rs b/crates/buzz-relay/src/handlers/mod.rs index 3315ea8eb6..cd61066890 100644 --- a/crates/buzz-relay/src/handlers/mod.rs +++ b/crates/buzz-relay/src/handlers/mod.rs @@ -6,6 +6,8 @@ pub mod auth; pub mod channel_authz; /// Subscription close (CLOSE) handler. pub mod close; +/// Per-key single-flight refresh with a minimum interval. +pub mod coalesced_refresh; /// Command executor — transactional processing for command kinds. pub mod command_executor; /// Relay-operator community provisioning HTTP support. diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 101ac4bc24..add1ddcd9b 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -981,6 +981,11 @@ async fn fan_out_system_message( /// incremented, so the emitted summary is exact even under concurrent /// replies/deletes (newest `created_at` wins client-side). /// +/// Coalesced per root through [`AppState::thread_summary_refresh`]: a quiet +/// thread emits at once, while a busy one emits at most once per +/// [`crate::state::THREAD_SUMMARY_REFRESH_INTERVAL`] per pod, with a trailing +/// emit that reads state committed after the last mutation in the burst. +/// /// Spawned: runs after the triggering write committed and must not add /// latency to the ingest acknowledgement, mirroring /// `dispatch_persistent_event`. @@ -990,88 +995,99 @@ pub fn emit_live_thread_summary( channel_id: Uuid, root_id: Vec, ) { + let key = (tenant.community(), root_id.clone()); let tenant = tenant.clone(); - let state = Arc::clone(state); - tokio::spawn(async move { - let summary = match state - .db - .get_thread_summary(tenant.community(), &root_id) - .await - { - Ok(Some(summary)) => summary, - // Root has no thread row — nothing to summarize (e.g., the root - // itself was just deleted). - Ok(None) => return, - Err(e) => { - warn!( - root = %hex::encode(&root_id), - "live thread summary lookup failed: {e}" - ); - return; - } - }; - - let root_hex = hex::encode(&root_id); - // Same tags/content shape as the channel-window page overlay in - // `api/bridge.rs` — one contract, two delivery doors. - let content = serde_json::json!({ - "reply_count": summary.reply_count, - "descendant_count": summary.descendant_count, - "last_reply_at": summary.last_reply_at.map(|t| t.timestamp()), - "participants": summary.participants.iter().map(hex::encode).collect::>(), - }); - let tags = [ - Tag::parse(["e", &root_hex]), - Tag::parse(["d", &root_hex]), - Tag::parse(["h", &channel_id.to_string()]), - ]; - let mut parsed = Vec::with_capacity(tags.len()); - for tag in tags { - match tag { - Ok(tag) => parsed.push(tag), - Err(e) => { - warn!(root = %root_hex, "live thread summary tag failed: {e}"); - return; - } - } - } - let event = match EventBuilder::new( - Kind::Custom(KIND_THREAD_SUMMARY as u16), - content.to_string(), + let task_state = Arc::clone(state); + state.thread_summary_refresh.request(key, move || { + publish_live_thread_summary( + tenant.clone(), + Arc::clone(&task_state), + channel_id, + root_id.clone(), ) - .tags(parsed) - .sign_with_keys(&state.relay_keypair) - { - Ok(event) => event, - Err(e) => { - warn!(root = %root_hex, "live thread summary sign failed: {e}"); - return; - } - }; - - // Redis before local fan-out so subscribers on other relay pods - // receive it too, matching `dispatch_persistent_event`. - state.mark_local_event(tenant.community(), &event.id); - if let Err(e) = state - .pubsub - .publish_event(&tenant, EventTopic::Channel(channel_id), &event) - .await - { - state - .local_event_ids - .invalidate(&(tenant.community(), event.id.to_bytes())); - warn!(root = %root_hex, "live thread summary Redis publish failed: {e}"); - } - let stored = StoredEvent::new(event, Some(channel_id)); - crate::handlers::event::fan_out_event_to_local_subscribers( - &state, - tenant.community(), - &stored, - ) - .await; }); } +async fn publish_live_thread_summary( + tenant: TenantContext, + state: Arc, + channel_id: Uuid, + root_id: Vec, +) { + let summary = match state + .db + .get_thread_summary(tenant.community(), &root_id) + .await + { + Ok(Some(summary)) => summary, + // Root has no thread row — nothing to summarize (e.g., the root + // itself was just deleted). + Ok(None) => return, + Err(e) => { + warn!( + root = %hex::encode(&root_id), + "live thread summary lookup failed: {e}" + ); + return; + } + }; + + let root_hex = hex::encode(&root_id); + // Same tags/content shape as the channel-window page overlay in + // `api/bridge.rs` — one contract, two delivery doors. + let content = serde_json::json!({ + "reply_count": summary.reply_count, + "descendant_count": summary.descendant_count, + "last_reply_at": summary.last_reply_at.map(|t| t.timestamp()), + "participants": summary.participants.iter().map(hex::encode).collect::>(), + }); + let tags = [ + Tag::parse(["e", &root_hex]), + Tag::parse(["d", &root_hex]), + Tag::parse(["h", &channel_id.to_string()]), + ]; + let mut parsed = Vec::with_capacity(tags.len()); + for tag in tags { + match tag { + Ok(tag) => parsed.push(tag), + Err(e) => { + warn!(root = %root_hex, "live thread summary tag failed: {e}"); + return; + } + } + } + let event = match EventBuilder::new( + Kind::Custom(KIND_THREAD_SUMMARY as u16), + content.to_string(), + ) + .tags(parsed) + .sign_with_keys(&state.relay_keypair) + { + Ok(event) => event, + Err(e) => { + warn!(root = %root_hex, "live thread summary sign failed: {e}"); + return; + } + }; + + // Redis before local fan-out so subscribers on other relay pods + // receive it too, matching `dispatch_persistent_event`. + state.mark_local_event(tenant.community(), &event.id); + if let Err(e) = state + .pubsub + .publish_event(&tenant, EventTopic::Channel(channel_id), &event) + .await + { + state + .local_event_ids + .invalidate(&(tenant.community(), event.id.to_bytes())); + warn!(root = %root_hex, "live thread summary Redis publish failed: {e}"); + } + let stored = StoredEvent::new(event, Some(channel_id)); + crate::handlers::event::fan_out_event_to_local_subscribers(&state, tenant.community(), &stored) + .await; +} + /// Emit a relay-signed membership notification event stored globally (channel_id = None). /// /// kind:44100 = member added, kind:44101 = member removed. Every notice diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 6898f40fe4..4fe9e7b1e7 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -1199,6 +1199,14 @@ impl Default for ConnectionManager { } } +/// Minimum spacing between live thread-summary emits for one thread root on +/// one relay pod. See [`AppState::thread_summary_refresh`]. +pub const THREAD_SUMMARY_REFRESH_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1); + +/// Upper bound on one live thread-summary refresh (DB read, Redis publish, +/// local fan-out). A hung refresh is abandoned so its root keeps emitting. +pub const THREAD_SUMMARY_REFRESH_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); + /// Shared application state, cloned cheaply via inner `Arc` fields. #[derive(Clone)] pub struct AppState { @@ -1335,6 +1343,11 @@ pub struct AppState { Arc>>, /// Current in-flight media uploads per (community, uploader pubkey). pub media_uploads_in_flight: Arc>, + /// Per-(community, thread root) coalescing for live `kind:39005` thread + /// summaries, so a busy thread recomputes its summary at most once per + /// [`THREAD_SUMMARY_REFRESH_INTERVAL`] instead of once per reply. + pub thread_summary_refresh: + Arc)>>, /// Cache for observer agent-owner authorization (kind 24200). /// Key: (community_id, agent_pubkey_bytes, owner_pubkey_bytes). Value: is_owner. /// `agent_owner_pubkey` is immutable inside one community, so a long TTL @@ -1609,6 +1622,12 @@ impl AppState { .build(), ), media_uploads_in_flight: Arc::new(DashMap::new()), + thread_summary_refresh: Arc::new( + crate::handlers::coalesced_refresh::CoalescedRefresh::new( + THREAD_SUMMARY_REFRESH_INTERVAL, + THREAD_SUMMARY_REFRESH_TIMEOUT, + ), + ), observer_owner_cache: Arc::new( moka::sync::Cache::builder() .max_capacity(100_000) diff --git a/docs/bridge-channel-window.md b/docs/bridge-channel-window.md index aec9d19542..1f2470866a 100644 --- a/docs/bridge-channel-window.md +++ b/docs/bridge-channel-window.md @@ -99,7 +99,7 @@ Clients **partition by kind before any cursor math**: - tags: `["e", ]`, `["d", ]`, `["h", ]` - content: `{"reply_count":n,"descendant_count":n,"last_reply_at":ts|null,"participants":["",...]}` - (participants: up to 10, most recent first) + (participants: up to 10 from the newest 1000 replies, deleted ones included, most recent first) - Signed by the relay keypair. Synthesized at query time, **never stored**. Clients treat it as replace-by-target metadata keyed by the `e`/`d` tag (the `d` tag gives parameterized-replaceable semantics natively); it is diff --git a/docs/nips/NIP-CW.md b/docs/nips/NIP-CW.md index 9e6ea14e1b..9016df4e52 100644 --- a/docs/nips/NIP-CW.md +++ b/docs/nips/NIP-CW.md @@ -139,7 +139,7 @@ One per returned row with replies. Tag cardinality is exact: one `e`, one `d`, o - `reply_count` — direct replies to the row. `descendant_count` — all events in the row's thread subtree. - `last_reply_at` — unix seconds of the newest descendant, or `null`. -- `participants` — up to 10 distinct author pubkeys from the thread, most recent first. +- `participants` — up to 10 distinct author pubkeys from the thread's newest 1000 replies, deleted replies included, most recent first. Deleted replies contribute no author, and an author silent for 1000 replies drops out. - The `e` and `d` tags both carry the row's event id: `e` for reference-following, `d` for replaceable addressing. ### `kind:39006` — window bounds diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index 3de473cf62..921dd68488 100755 --- a/scripts/run-tests.sh +++ b/scripts/run-tests.sh @@ -206,6 +206,9 @@ run_unit_tests() { run_test_step "buzz-relay side-effects helper tests" \ cargo test -p buzz-relay --lib handlers::side_effects::tests:: -- --nocapture + run_test_step "buzz-relay coalesced refresh tests" \ + cargo test -p buzz-relay --lib handlers::coalesced_refresh::tests:: -- --nocapture + run_test_step "buzz-relay system channel scope tests" \ cargo test -p buzz-relay --lib handlers::system_channels::tests:: -- --nocapture