From a39508ccebd98e2094a5bf7058a37f312a2c8f5d Mon Sep 17 00:00:00 2001 From: npub12gtutshhh76rx0jx697f32f9tffd4hhp3hx58fp4x6u4uemkm7sqf8f757 <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@sprout-oss.stage.blox.sqprod.co> Date: Thu, 2 Jul 2026 12:20:53 -0400 Subject: [PATCH] Optimize desktop channel live subscriptions Co-authored-by: npub12gtutshhh76rx0jx697f32f9tffd4hhp3hx58fp4x6u4uemkm7sqf8f757 <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@sprout-oss.stage.blox.sqprod.co> Signed-off-by: npub12gtutshhh76rx0jx697f32f9tffd4hhp3hx58fp4x6u4uemkm7sqf8f757 <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@sprout-oss.stage.blox.sqprod.co> --- desktop/src/features/messages/hooks.ts | 102 ++++++++---- .../shared/api/relayChannelFilters.test.mjs | 7 + desktop/src/shared/api/relayChannelFilters.ts | 6 + desktop/src/shared/api/relayClientSession.ts | 4 +- .../shared/api/relayFrontierBridge.test.mjs | 145 ++++++++++++++++++ desktop/src/shared/api/relayFrontierBridge.ts | 135 ++++++++++++++++ 6 files changed, 371 insertions(+), 28 deletions(-) create mode 100644 desktop/src/shared/api/relayFrontierBridge.test.mjs create mode 100644 desktop/src/shared/api/relayFrontierBridge.ts diff --git a/desktop/src/features/messages/hooks.ts b/desktop/src/features/messages/hooks.ts index 81feccd0c40..dae08d0e105 100644 --- a/desktop/src/features/messages/hooks.ts +++ b/desktop/src/features/messages/hooks.ts @@ -17,6 +17,12 @@ import { } from "@/features/messages/lib/threading"; import { splitOutgoingTags } from "@/features/messages/lib/imetaMediaMarkdown"; import { relayClient } from "@/shared/api/relayClient"; +import { buildChannelLiveFilter } from "@/shared/api/relayChannelFilters"; +import { + frontierBridge, + newestEventFrontier, + type EventFrontier, +} from "@/shared/api/relayFrontierBridge"; import { customEmojiQueryKey } from "@/features/custom-emoji/hooks"; import { reactionEmojiUrl } from "@/shared/api/customEmoji"; import type { CustomEmoji } from "@/shared/lib/remarkCustomEmoji"; @@ -269,23 +275,56 @@ export function useChannelSubscription(channel: Channel | null) { const queryClient = useQueryClient(); const channelId = channel?.id ?? null; const channelType = channel?.channelType ?? null; - const syncLatestHistory = useEffectEvent(async () => { - if (!channelId) { - return; - } + const bridgeFromFrontier = useEffectEvent( + async ( + capturedFrontier?: EventFrontier | null, + isActive: () => boolean = () => true, + ) => { + if (!channelId) { + return; + } - const history = await relayClient.fetchChannelHistory( - channelId, - CHANNEL_HISTORY_LIMIT, - ); + const queryKey = channelMessagesKey(channelId); + const current = queryClient.getQueryData(queryKey) ?? []; + const frontier = capturedFrontier ?? newestEventFrontier(current); - queryClient.setQueryData( - channelMessagesKey(channelId), - (current = []) => mergeTimelineHistoryMessages(current, history), - ); + if (!frontier) { + const history = await relayClient.fetchChannelHistory( + channelId, + CHANNEL_HISTORY_LIMIT, + ); + if (!isActive()) { + return; + } + queryClient.setQueryData(queryKey, (latest = []) => + mergeTimelineHistoryMessages(latest, history), + ); + void backfillAuxForMessages(queryClient, channelId, history); + return; + } - void backfillAuxForMessages(queryClient, channelId, history); - }); + const bridgedEvents: RelayEvent[] = []; + await frontierBridge({ + frontier, + isActive, + knownEventIds: new Set(current.map((event) => event.id)), + targetFilter: buildChannelLiveFilter(channelId), + requestHistory: (filter) => relayClient.fetchEvents(filter), + onEvent: (event) => { + bridgedEvents.push(event); + }, + }); + + if (bridgedEvents.length === 0 || !isActive()) { + return; + } + + queryClient.setQueryData(queryKey, (latest = []) => + mergeTimelineHistoryMessages(latest, bridgedEvents), + ); + void backfillAuxForMessages(queryClient, channelId, bridgedEvents); + }, + ); const appendMessage = useEffectEvent((event: RelayEvent) => { if (!channelId) { @@ -327,10 +366,10 @@ export function useChannelSubscription(channel: Channel | null) { let isDisposed = false; let cleanup: (() => Promise) | undefined; const disposeReconnectListener = relayClient.subscribeToReconnects(() => { - void syncLatestHistory().catch((error) => { + void bridgeFromFrontier(undefined, () => !isDisposed).catch((error) => { if (!isDisposed) { console.error( - "Failed to refresh channel history after reconnecting", + "Failed to bridge channel history after reconnecting", channelId, error, ); @@ -338,6 +377,11 @@ export function useChannelSubscription(channel: Channel | null) { }); }); + const queryKey = channelMessagesKey(channelId); + const subscriptionFrontier = newestEventFrontier( + queryClient.getQueryData(queryKey) ?? [], + ); + relayClient .subscribeToChannel(channelId, (event) => { if (!isDisposed) { @@ -351,15 +395,21 @@ export function useChannelSubscription(channel: Channel | null) { } cleanup = dispose; - // No post-subscribe history refetch: useChannelMessagesQuery already - // loaded the latest CHANNEL_HISTORY_LIMIT events, and the live - // subscription itself backfills up to 50 most-recent events via its - // initial REQ (buildChannelFilter(id, 50)). Both write into the same - // channelMessagesKey cache, so any window between the two REQs is - // covered by the live sub's overlap unless >50 messages land in - // <1s — vanishingly rare in practice. The reconnect listener above - // still bridges gaps from connection drops, where the gap *is* - // unbounded. + // The live sub is registered with limit:0, so channel switches no + // longer replay the same 50 historical events that the query path just + // loaded. Once EOSE confirms the live sub is installed, explicitly + // bridge from the cache frontier to close the history→live race. + void bridgeFromFrontier(subscriptionFrontier, () => !isDisposed).catch( + (error) => { + if (!isDisposed) { + console.error( + "Failed to bridge channel history", + channelId, + error, + ); + } + }, + ); }) .catch((error) => { console.error("Failed to subscribe to channel", channelId, error); @@ -372,7 +422,7 @@ export function useChannelSubscription(channel: Channel | null) { void cleanup(); } }; - }, [channelId, channelType]); + }, [channelId, channelType, queryClient]); } export function useSendMessageMutation( diff --git a/desktop/src/shared/api/relayChannelFilters.test.mjs b/desktop/src/shared/api/relayChannelFilters.test.mjs index 3519c82146f..d8439ee34a5 100644 --- a/desktop/src/shared/api/relayChannelFilters.test.mjs +++ b/desktop/src/shared/api/relayChannelFilters.test.mjs @@ -4,6 +4,7 @@ import test from "node:test"; import { buildChannelAuxDeletionFilter, buildChannelAuxFilter, + buildChannelLiveFilter, buildChannelReactionAuxFilter, buildChannelStructuralAuxFilter, } from "./relayChannelFilters.ts"; @@ -43,3 +44,9 @@ test("buildChannelStructuralAuxFilter excludes reactions", () => { assert.deepEqual(filter["#e"], IDS); assert.equal("#h" in filter, false); }); + +test("buildChannelLiveFilter subscribes with limit 0 to avoid historical replay", () => { + const filter = buildChannelLiveFilter(CHANNEL); + assert.equal(filter.limit, 0); + assert.deepEqual(filter["#h"], [CHANNEL]); +}); diff --git a/desktop/src/shared/api/relayChannelFilters.ts b/desktop/src/shared/api/relayChannelFilters.ts index d0c7e7938e0..356dead1ac3 100644 --- a/desktop/src/shared/api/relayChannelFilters.ts +++ b/desktop/src/shared/api/relayChannelFilters.ts @@ -40,6 +40,12 @@ export function buildChannelFilter( return filter; } +export function buildChannelLiveFilter( + channelId: string, +): RelaySubscriptionFilter { + return buildChannelFilter(channelId, 0); +} + /** * History filter for cold-load and scrollback: message kinds *only*, so the * `limit` budget buys visible message depth. Auxiliary events (reactions, diff --git a/desktop/src/shared/api/relayClientSession.ts b/desktop/src/shared/api/relayClientSession.ts index 1e693ddf22a..f3bada51172 100644 --- a/desktop/src/shared/api/relayClientSession.ts +++ b/desktop/src/shared/api/relayClientSession.ts @@ -22,8 +22,8 @@ import { import { AUX_BACKFILL_CHUNK_SIZE, buildChannelAuxDeletionFilter, - buildChannelFilter, buildChannelHistoryFilter, + buildChannelLiveFilter, buildChannelMentionFilter, buildGlobalStreamFilter, } from "@/shared/api/relayChannelFilters"; @@ -330,7 +330,7 @@ export class RelayClient { channelId: string, onEvent: (event: RelayEvent) => void, ) { - return this.subscribe(buildChannelFilter(channelId, 50), onEvent); + return this.subscribe(buildChannelLiveFilter(channelId), onEvent); } /** diff --git a/desktop/src/shared/api/relayFrontierBridge.test.mjs b/desktop/src/shared/api/relayFrontierBridge.test.mjs new file mode 100644 index 00000000000..19a590e3ab9 --- /dev/null +++ b/desktop/src/shared/api/relayFrontierBridge.test.mjs @@ -0,0 +1,145 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { + buildFrontierBridgeFilter, + frontierBridge, + isEventAfterFrontier, + newestEventFrontier, + shouldApplyBridgeEvent, +} from "./relayFrontierBridge.ts"; +import { buildChannelFilter } from "./relayChannelFilters.ts"; +import { mergeTimelineHistoryMessages } from "../../features/messages/lib/messageQueryKeys.ts"; + +function event(id, createdAt, kind = 9) { + return { + id, + pubkey: "pubkey", + created_at: createdAt, + kind, + tags: [["h", "channel-1"]], + content: "", + sig: "sig", + }; +} + +function verifyBridge({ before, bridgePage, atomic }) { + const frontier = newestEventFrontier(before); + const delivered = []; + + return frontierBridge({ + frontier, + targetFilter: buildChannelFilter("channel-1", 0), + knownEventIds: new Set(before.map((ev) => ev.id)), + requestHistory: async () => bridgePage, + onEvent: (ev) => delivered.push(ev), + }).then(() => { + const bridged = mergeTimelineHistoryMessages(before, delivered); + const fresh = mergeTimelineHistoryMessages(before, atomic); + assert.deepEqual( + bridged.map((ev) => ev.id), + fresh.map((ev) => ev.id), + ); + }); +} + +test("newestEventFrontier uses the composite (created_at, id) frontier", () => { + const frontier = newestEventFrontier([ + event("b", 10), + event("a", 11), + event("c", 11), + ]); + + assert.deepEqual(frontier, { createdAt: 11, eventId: "c" }); +}); + +test("frontier bridge filter keeps since inclusive for same-second ties", () => { + const targetFilter = { ...buildChannelFilter("channel-1", 0), since: 90 }; + const filter = buildFrontierBridgeFilter({ + frontier: { createdAt: 100, eventId: "b" }, + targetFilter, + }); + + assert.equal(filter.limit, 500); + assert.equal(filter.since, 100); + assert.deepEqual(filter["#h"], ["channel-1"]); + assert.deepEqual(filter.kinds, targetFilter.kinds); +}); + +test("frontier comparison orders dense-second events by id", () => { + const frontier = { createdAt: 100, eventId: "b" }; + + assert.equal(isEventAfterFrontier(event("a", 100), frontier), false); + assert.equal(isEventAfterFrontier(event("b", 100), frontier), false); + assert.equal(isEventAfterFrontier(event("c", 100), frontier), true); + assert.equal(isEventAfterFrontier(event("a", 101), frontier), true); +}); + +test("bridge application keeps unknown same-second events that sort before the frontier", () => { + const frontier = { createdAt: 100, eventId: "m" }; + const knownEventIds = new Set(["m"]); + + assert.equal( + shouldApplyBridgeEvent({ + event: event("a", 100), + frontier, + knownEventIds, + }), + true, + ); + assert.equal( + shouldApplyBridgeEvent({ + event: event("m", 100), + frontier, + knownEventIds, + }), + false, + ); +}); + +test("verify_bridge: live-only bridge equals fresh atomic fetch across dense-second and aux events", async () => { + const before = [event("a", 100), event("b", 100), event("m", 101)]; + const afterSameSecondLowerId = event("l", 101); + const afterSameSecond = event("z", 101); + const afterNextSecond = event("n", 102); + const aux = event("aux", 103, 7); + const duplicateAtFrontier = event("m", 101); + + await verifyBridge({ + before, + // Relay history arrives newest-first; the bridge must sort/apply into the + // same timeline cache projection as a fresh atomic read. + bridgePage: [ + aux, + afterNextSecond, + duplicateAtFrontier, + afterSameSecond, + afterSameSecondLowerId, + ], + atomic: [ + duplicateAtFrontier, + afterSameSecondLowerId, + afterSameSecond, + afterNextSecond, + aux, + ], + }); +}); + +test("frontierBridge stops applying when the owner is disposed", async () => { + const delivered = []; + let active = true; + + await frontierBridge({ + frontier: { createdAt: 100, eventId: "a" }, + targetFilter: buildChannelFilter("channel-1", 0), + requestHistory: async () => { + active = false; + return [event("b", 101)]; + }, + isActive: () => active, + onEvent: (ev) => delivered.push(ev), + }); + + assert.deepEqual(delivered, []); +}); diff --git a/desktop/src/shared/api/relayFrontierBridge.ts b/desktop/src/shared/api/relayFrontierBridge.ts new file mode 100644 index 00000000000..9bc99d8ec07 --- /dev/null +++ b/desktop/src/shared/api/relayFrontierBridge.ts @@ -0,0 +1,135 @@ +import { sortEvents } from "@/shared/api/relayClientShared"; +import type { RelaySubscriptionFilter } from "@/shared/api/relayClientShared"; +import type { RelayEvent } from "@/shared/api/types"; + +export type EventFrontier = { + createdAt: number; + eventId: string; +}; + +export const FRONTIER_BRIDGE_PAGE_LIMIT = 500; + +export function compareEventToFrontier( + event: RelayEvent, + frontier: EventFrontier, +) { + if (event.created_at !== frontier.createdAt) { + return event.created_at - frontier.createdAt; + } + return event.id.localeCompare(frontier.eventId); +} + +export function isEventAfterFrontier( + event: RelayEvent, + frontier: EventFrontier, +) { + return compareEventToFrontier(event, frontier) > 0; +} + +export function shouldApplyBridgeEvent({ + event, + frontier, + knownEventIds, +}: { + event: RelayEvent; + frontier: EventFrontier; + knownEventIds: ReadonlySet; +}) { + if (event.created_at > frontier.createdAt) { + return true; + } + + // NIP-01 `since` is second-granularity. Keep unknown events in the frontier + // second even if their id sorts before the frontier id; they may have landed + // in the history→live race and a strict `(created_at,id) > frontier` filter + // would silently drop them. Known ids are harmless duplicates. + return ( + event.created_at === frontier.createdAt && !knownEventIds.has(event.id) + ); +} + +export function newestEventFrontier( + events: RelayEvent[], +): EventFrontier | null { + let frontier: EventFrontier | null = null; + + for (const event of events) { + if (!frontier || isEventAfterFrontier(event, frontier)) { + frontier = { + createdAt: event.created_at, + eventId: event.id, + }; + } + } + + return frontier; +} + +export function buildFrontierBridgeFilter({ + frontier, + limit = FRONTIER_BRIDGE_PAGE_LIMIT, + targetFilter, +}: { + frontier: EventFrontier; + limit?: number; + targetFilter: RelaySubscriptionFilter; +}): RelaySubscriptionFilter { + return { + ...targetFilter, + limit, + since: + targetFilter.since === undefined + ? frontier.createdAt + : Math.max(targetFilter.since, frontier.createdAt), + }; +} + +/** + * Fetch and apply the explicit gap bridge after a live subscription is ready. + * + * The relay's NIP-01 `since` cursor is second-granularity, so the request is + * intentionally inclusive at `frontier.createdAt`; the client then drops known + * duplicates and keeps unknown frontier-second events too. That closes the + * history→live race even when a new same-second event sorts before the current + * frontier id, while normal id-dedupe keeps duplicate delivery harmless. + */ +export async function frontierBridge({ + frontier, + isActive = () => true, + knownEventIds, + limit, + onEvent, + requestHistory, + targetFilter, +}: { + frontier: EventFrontier | null; + isActive?: () => boolean; + limit?: number; + knownEventIds?: ReadonlySet; + onEvent: (event: RelayEvent) => void; + requestHistory: (filter: RelaySubscriptionFilter) => Promise; + targetFilter: RelaySubscriptionFilter; +}) { + if (!frontier) { + return; + } + + const knownIds = knownEventIds ?? new Set(); + + const events = await requestHistory( + buildFrontierBridgeFilter({ frontier, limit, targetFilter }), + ); + + if (!isActive()) { + return; + } + + for (const event of sortEvents(events).filter((event) => + shouldApplyBridgeEvent({ event, frontier, knownEventIds: knownIds }), + )) { + if (!isActive()) { + return; + } + onEvent(event); + } +}