diff --git a/desktop/src/features/messages/hooks.ts b/desktop/src/features/messages/hooks.ts index 062b0ee40b..72225c9dc9 100644 --- a/desktop/src/features/messages/hooks.ts +++ b/desktop/src/features/messages/hooks.ts @@ -262,19 +262,19 @@ export function useChannelSubscription(channel: Channel | null) { }); const appendMessage = useEffectEvent((event: RelayEvent) => { - if (!channelId) return; + if (!channelId) return false; if (event.kind === KIND_CHANNEL_THREAD_SUMMARY) { // Relay-pushed live badge recount — window-store overlay only, never a // timeline row (mirrors the page path, where 39005 is metadata). const parsed = parseLiveThreadSummary(event); - if (!parsed) return; + if (!parsed) return false; const windowKey = channelWindowKey(channelId); const current = queryClient.getQueryData(windowKey) ?? emptyChannelWindowStore(); const next = mergeLiveThreadSummary(current, parsed.rootId, parsed.live); if (next !== current) queryClient.setQueryData(windowKey, next); - return; + return false; } const isTimelineRow = CHANNEL_TIMELINE_KINDS.has(event.kind); const threadReference = isTimelineRow @@ -288,9 +288,9 @@ export function useChannelSubscription(channel: Channel | null) { (current = []) => mergeMessages(current, event), ); } - if (!isBroadcastReply(event.tags)) return; + if (!isBroadcastReply(event.tags)) return false; } - if (!isTimelineRow && !CHANNEL_AUX_KINDS.has(event.kind)) return; + if (!isTimelineRow && !CHANNEL_AUX_KINDS.has(event.kind)) return false; if (!isTimelineRow) { queryClient.setQueriesData( { queryKey: ["thread-replies", channelId] }, @@ -305,7 +305,6 @@ export function useChannelSubscription(channel: Channel | null) { const next = mergeLiveChannelWindowEvent(current, event, isTimelineRow); if (next !== current) { queryClient.setQueryData(windowKey, next); - projectChannelWindowMessages(queryClient, channelId); } if (event.kind === KIND_SYSTEM_MESSAGE) { @@ -328,6 +327,7 @@ export function useChannelSubscription(channel: Channel | null) { // Non-JSON system message — ignore. } } + return next !== current; }); // Notify the relay client which channel is currently visible so its live @@ -348,6 +348,7 @@ export function useChannelSubscription(channel: Channel | null) { let isDisposed = false; let cleanup: (() => Promise) | undefined; + let shouldProjectOnFlush = false; const disposeReconnectListener = relayClient.subscribeToReconnects(() => { void refreshNewestWindow().catch((error) => { if (!isDisposed) { @@ -361,11 +362,22 @@ export function useChannelSubscription(channel: Channel | null) { }); relayClient - .subscribeToChannelLive(channelId, (event) => { - if (!isDisposed) { - appendMessage(event); - } - }) + .subscribeToChannelLive( + channelId, + (event) => { + if (!isDisposed) { + shouldProjectOnFlush = appendMessage(event) || shouldProjectOnFlush; + } + }, + { + onFlush: () => { + if (!isDisposed && shouldProjectOnFlush) { + projectChannelWindowMessages(queryClient, channelId); + } + shouldProjectOnFlush = false; + }, + }, + ) .then((dispose) => { if (isDisposed) { void dispose(); @@ -394,7 +406,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 3519c82146..fdce1c3684 100644 --- a/desktop/src/shared/api/relayChannelFilters.test.mjs +++ b/desktop/src/shared/api/relayChannelFilters.test.mjs @@ -4,9 +4,16 @@ import test from "node:test"; import { buildChannelAuxDeletionFilter, buildChannelAuxFilter, + buildChannelLiveFilter, buildChannelReactionAuxFilter, buildChannelStructuralAuxFilter, } from "./relayChannelFilters.ts"; +import { + CHANNEL_EVENT_KINDS, + KIND_CHANNEL_THREAD_SUMMARY, +} from "../constants/kinds.ts"; +import { shouldPageReconnectReplay } from "./relayReconnectReplay.ts"; +import { handleRelayClosed } from "./relayClosedRecovery.ts"; const CHANNEL = "36411e44-0e2d-4cfe-bd6e-567eb169db9f"; const IDS = [ @@ -43,3 +50,89 @@ test("buildChannelStructuralAuxFilter excludes reactions", () => { assert.deepEqual(filter["#e"], IDS); assert.equal("#h" in filter, false); }); + +test("buildChannelLiveFilter is replay-bounded without a reader-clock since", () => { + const filter = buildChannelLiveFilter(CHANNEL); + + assert.equal("since" in filter, false); + assert.ok(filter.limit > 0); + assert.equal(shouldPageReconnectReplay(filter), true); + assert.deepEqual(filter["#h"], [CHANNEL]); + assert.deepEqual(filter.kinds, [ + ...CHANNEL_EVENT_KINDS, + KIND_CHANNEL_THREAD_SUMMARY, + ]); +}); + +test("CLOSED retry pages a gap larger than the live limit from last-seen author time", async () => { + const originalWindow = globalThis.window; + const originalDateNow = Date.now; + let retry; + globalThis.window = { + setTimeout: (callback) => { + retry = callback; + return 1; + }, + clearTimeout: () => {}, + }; + Date.now = () => 1_100_000; + + try { + const liveFilter = buildChannelLiveFilter(CHANNEL); + const recoveredEvents = Array.from( + { length: liveFilter.limit + 1 }, + (_, index) => ({ + id: String(index).padStart(64, "0"), + pubkey: "a".repeat(64), + created_at: 1_001 + index, + kind: 9, + tags: [["h", CHANNEL]], + content: `recovered ${index}`, + sig: "b".repeat(128), + }), + ); + const delivered = []; + const sentFilters = []; + let requestedFilter; + let resolveReplay; + const replayed = new Promise((resolve) => { + resolveReplay = resolve; + }); + const subscription = { + mode: "live", + filter: liveFilter, + onEvent: (event) => delivered.push(event), + lastSeenCreatedAt: 1_000, + }; + const subscriptions = new Map([["live-closed", subscription]]); + + handleRelayClosed({ + subscriptions, + subId: "live-closed", + message: "error: temporary relay failure", + sendReq: async (_subId, filter) => { + sentFilters.push(filter); + }, + requestHistory: async (filter) => { + requestedFilter = filter; + resolveReplay(); + return recoveredEvents; + }, + replaySubscriptionEvent: (_subId, event) => delivered.push(event), + }); + + assert.equal(typeof retry, "function"); + retry(); + await replayed; + await Promise.resolve(); + + assert.equal(sentFilters.length, 1); + assert.equal(sentFilters[0], liveFilter); + assert.equal(requestedFilter.since, 995); + assert.equal(requestedFilter.limit, 500); + assert.equal(delivered.length, liveFilter.limit + 1); + } finally { + globalThis.window = originalWindow; + Date.now = originalDateNow; + } +}); diff --git a/desktop/src/shared/api/relayChannelFilters.ts b/desktop/src/shared/api/relayChannelFilters.ts index d0c7e7938e..6afe18da69 100644 --- a/desktop/src/shared/api/relayChannelFilters.ts +++ b/desktop/src/shared/api/relayChannelFilters.ts @@ -3,6 +3,7 @@ import { CHANNEL_EVENT_KINDS, CHANNEL_TIMELINE_CONTENT_KINDS, HOME_MENTION_EVENT_KINDS, + KIND_CHANNEL_THREAD_SUMMARY, KIND_DELETION, KIND_NIP29_DELETE_EVENT, KIND_REACTION, @@ -17,6 +18,29 @@ import type { RelaySubscriptionFilter } from "@/shared/api/relayClientShared"; export const AUX_BACKFILL_CHUNK_SIZE = 100; export const MAX_HISTORICAL_LIMIT = 10_000; +/** + * Live window-store subscription for an open channel. + * + * Deliberately omits `since`: deriving it from the reader's clock permanently + * drops accepted events whose author clock is behind. Initial replay volume is + * bounded with `limit: 50` instead; `limit` does not constrain future fan-out. + * Keep the limit above zero because `shouldPageReconnectReplay` uses that as + * the gate for paged reconnect recovery. The authoritative HTTP window read + * owns history depth, so a larger WebSocket replay only adds serialized load. + */ +export function buildChannelLiveFilter( + channelId: string, +): RelaySubscriptionFilter { + return { + // 39005 rides only this window-store subscription — not + // CHANNEL_EVENT_KINDS, whose other consumers (unread tracking, + // timeline-cache merges) must never see summary overlays. + kinds: [...CHANNEL_EVENT_KINDS, KIND_CHANNEL_THREAD_SUMMARY], + "#h": [channelId], + limit: 50, + }; +} + /** * Live-subscription filter for an open channel: the broad * {@link CHANNEL_EVENT_KINDS} set so the tail delivers reactions/edits/ diff --git a/desktop/src/shared/api/relayClientSession.ts b/desktop/src/shared/api/relayClientSession.ts index 84ee10b68d..480098a907 100644 --- a/desktop/src/shared/api/relayClientSession.ts +++ b/desktop/src/shared/api/relayClientSession.ts @@ -9,8 +9,6 @@ import { KIND_STREAM_MESSAGE, KIND_TYPING_INDICATOR, KIND_USER_STATUS, - CHANNEL_EVENT_KINDS, - KIND_CHANNEL_THREAD_SUMMARY, } from "@/shared/constants/kinds"; import { getTextPayload, @@ -23,6 +21,7 @@ import { buildChannelAuxDeletionFilter, buildChannelFilter, buildChannelHistoryFilter, + buildChannelLiveFilter, buildChannelMentionFilter, buildGlobalStreamFilter, } from "@/shared/api/relayChannelFilters"; @@ -80,6 +79,8 @@ export const BACKOFF_RESET_STABLE_MS = 60_000; const STALL_CHECK_INTERVAL_MS = 10_000; const STALL_IDLE_TIMEOUT_MS = 60_000; +type LiveCallbacks = Partial void>>; + export class RelayClient { private wsId: number | null = null; private relayUrl: string | null = null; @@ -356,23 +357,13 @@ export class RelayClient { return this.subscribe(buildChannelFilter(channelId, 50), onEvent); } - /** Subscribe to channel rows and aux starting now, with no history replay. */ + /** Subscribe to channel rows and aux, with one callback per buffered flush. */ async subscribeToChannelLive( channelId: string, onEvent: (event: RelayEvent) => void, + options?: LiveCallbacks, ) { - return this.subscribe( - { - // 39005 rides only this window-store subscription — not - // CHANNEL_EVENT_KINDS, whose other consumers (unread tracking, - // timeline-cache merges) must never see summary overlays. - kinds: [...CHANNEL_EVENT_KINDS, KIND_CHANNEL_THREAD_SUMMARY], - "#h": [channelId], - limit: 1000, - since: Math.floor(Date.now() / 1_000), - }, - onEvent, - ); + return this.subscribe(buildChannelLiveFilter(channelId), onEvent, options); } /** @@ -600,6 +591,7 @@ export class RelayClient { private async subscribe( filter: RelaySubscriptionFilter, onEvent: (event: RelayEvent) => void, + options?: LiveCallbacks, ) { await this.ensureConnected(); @@ -622,6 +614,7 @@ export class RelayClient { filter, onEvent, resolveReady, + ...options, }); try { @@ -829,6 +822,8 @@ export class RelayClient { ["REQ", subId, filter], "Failed to restore relay subscription after CLOSED.", ), + requestHistory: (filter) => this.requestHistory(filter), + replaySubscriptionEvent: this.handleEvent.bind(this), }); return; } @@ -884,14 +879,19 @@ export class RelayClient { this.flushTimeout = null; const buffer = this.eventBuffer; this.eventBuffer = []; + const flushCallbacks = new Set<() => void>(); // Re-lookup: subscriptions removed during batch window are intentionally skipped. for (const { subId, event } of buffer) { const subscription = this.subscriptions.get(subId); if (subscription?.mode === "live") { subscription.onEvent(event); + const callback = (subscription as typeof subscription & LiveCallbacks) + .onFlush; + if (callback) flushCallbacks.add(callback); } } + for (const callback of flushCallbacks) callback(); } private handleEose(subId: string) { @@ -943,7 +943,6 @@ export class RelayClient { return false; } - private async replayLiveSubscriptions() { const generation = this.connectionGeneration; try { @@ -951,6 +950,7 @@ export class RelayClient { subscriptions: this.subscriptions, sendRaw: (payload) => this.sendRaw(payload), requestHistory: (filter) => this.requestHistory(filter), + replaySubscriptionEvent: this.handleEvent.bind(this), visibleChannelId: this.visibleChannelId, isActive: () => this.connectionGeneration === generation, }); diff --git a/desktop/src/shared/api/relayClosedRecovery.ts b/desktop/src/shared/api/relayClosedRecovery.ts index fd366671fd..a5712863b4 100644 --- a/desktop/src/shared/api/relayClosedRecovery.ts +++ b/desktop/src/shared/api/relayClosedRecovery.ts @@ -9,10 +9,16 @@ import { type RelaySubscription, type RelaySubscriptionFilter, } from "@/shared/api/relayClientShared"; +import { + buildReconnectReplayFilter, + replayReconnectHistoryPages, + shouldPageReconnectReplay, +} from "@/shared/api/relayReconnectReplay"; import type { RelayEvent } from "@/shared/api/types"; const RETRY_BASE_DELAY_MS = 1_000; const RETRY_MAX_DELAY_MS = 30_000; +const CLOSED_REPLAY_SKEW_SECS = 5; type LiveSubscription = Extract; @@ -27,11 +33,15 @@ export function handleRelayClosed({ subId, message, sendReq, + requestHistory, + replaySubscriptionEvent, }: { subscriptions: Map; subId: string; message: string; sendReq: (subId: string, filter: RelaySubscriptionFilter) => Promise; + requestHistory?: (filter: RelaySubscriptionFilter) => Promise; + replaySubscriptionEvent?: (subId: string, event: RelayEvent) => void; }) { const subscription = subscriptions.get(subId); if (!subscription) return; @@ -57,6 +67,8 @@ export function handleRelayClosed({ subscription, message, sendReq, + requestHistory, + replaySubscriptionEvent, }); } @@ -66,12 +78,16 @@ function recoverLiveSubscriptionFromClosed({ subscription, message, sendReq, + requestHistory, + replaySubscriptionEvent, }: { subscriptions: Map; subId: string; subscription: LiveSubscription; message: string; sendReq: (subId: string, filter: RelaySubscriptionFilter) => Promise; + requestHistory?: (filter: RelaySubscriptionFilter) => Promise; + replaySubscriptionEvent?: (subId: string, event: RelayEvent) => void; }) { subscription.resolveReady?.(); subscription.resolveReady = undefined; @@ -111,7 +127,43 @@ function recoverLiveSubscriptionFromClosed({ subscription.closedRetryTimeout = window.setTimeout(() => { subscription.closedRetryTimeout = undefined; if (subscriptions.get(subId) !== subscription) return; - void sendReq(subId, subscription.filter).catch((error) => { + const replaySince = + subscription.lastSeenCreatedAt === undefined + ? undefined + : Math.max(0, subscription.lastSeenCreatedAt - CLOSED_REPLAY_SKEW_SECS); + const shouldPageReplay = + replaySince !== undefined && + requestHistory !== undefined && + shouldPageReconnectReplay(subscription.filter); + + void (async () => { + await sendReq( + subId, + shouldPageReplay + ? subscription.filter + : buildReconnectReplayFilter(subscription.filter, replaySince), + ); + if (!shouldPageReplay || replaySince === undefined || !requestHistory) { + return; + } + + await replayReconnectHistoryPages({ + subscription: { + ...subscription, + onEvent: (event) => { + if (replaySubscriptionEvent) { + replaySubscriptionEvent(subId, event); + } else { + subscription.onEvent(event); + } + }, + }, + since: replaySince, + until: Math.floor(Date.now() / 1_000), + isActive: () => subscriptions.get(subId) === subscription, + requestHistory, + }); + })().catch((error) => { if (subscriptions.get(subId) !== subscription) return; console.error("Failed to restore closed relay subscription", error); recoverLiveSubscriptionFromClosed({ @@ -120,6 +172,8 @@ function recoverLiveSubscriptionFromClosed({ subscription, message, sendReq, + requestHistory, + replaySubscriptionEvent, }); }); }, delayMs); diff --git a/desktop/src/shared/api/relayReconnectReplay.ts b/desktop/src/shared/api/relayReconnectReplay.ts index 74b752f666..ce08947735 100644 --- a/desktop/src/shared/api/relayReconnectReplay.ts +++ b/desktop/src/shared/api/relayReconnectReplay.ts @@ -122,6 +122,7 @@ export async function replayLiveSubscriptions({ subscriptions, sendRaw, requestHistory, + replaySubscriptionEvent, now = Math.floor(Date.now() / 1_000), pageReplayConcurrency = RECONNECT_REPLAY_PAGE_CONCURRENCY, visibleChannelId = null, @@ -134,6 +135,7 @@ export async function replayLiveSubscriptions({ subscriptions: Map; sendRaw: (payload: unknown[]) => Promise; requestHistory: (filter: RelaySubscriptionFilter) => Promise; + replaySubscriptionEvent?: (subId: string, event: RelayEvent) => void; now?: number; pageReplayConcurrency?: number; /** Channel currently visible in the UI — its subscriptions go in the first batch. */ @@ -238,7 +240,16 @@ export async function replayLiveSubscriptions({ pageReplayConcurrency, async ({ subId, subscription, replaySince }) => { await replayReconnectHistoryPages({ - subscription, + subscription: { + ...subscription, + onEvent: (event) => { + if (replaySubscriptionEvent) { + replaySubscriptionEvent(subId, event); + } else { + subscription.onEvent(event); + } + }, + }, since: replaySince, until: now, isActive: () => subscriptions.get(subId) === subscription, diff --git a/desktop/tests/e2e/cold-switch-longtask.perf.ts b/desktop/tests/e2e/cold-switch-longtask.perf.ts index c563b12946..7ddf582303 100644 --- a/desktop/tests/e2e/cold-switch-longtask.perf.ts +++ b/desktop/tests/e2e/cold-switch-longtask.perf.ts @@ -33,9 +33,11 @@ import { installMockBridge } from "../helpers/bridge"; * SCOPE LIMIT: this measures Chromium main-thread longtasks under throttle. It * does NOT measure the WKWebView compositor feel on the shipped Tauri shell — * that is a separate real-wheel pass. + * This harness observes main-thread longtasks only. It cannot observe network + * round trips and must never be cited as evidence that a round trip was removed. * * Run it: - * pnpm build && npx playwright test --config=playwright.perf.config.ts \ + * pnpm build:e2e && npx playwright test --config=playwright.perf.config.ts \ * cold-switch-longtask.perf.ts */ diff --git a/desktop/tests/e2e/warm-switch-markdown.perf.ts b/desktop/tests/e2e/warm-switch-markdown.perf.ts index 291a052f14..47fcc1f18c 100644 --- a/desktop/tests/e2e/warm-switch-markdown.perf.ts +++ b/desktop/tests/e2e/warm-switch-markdown.perf.ts @@ -35,9 +35,11 @@ import { installMockBridge } from "../helpers/bridge"; * paths are jitted. 4x CPU throttle for the same reason as the cold spec — * absolute ms are not portable across machines, but before/after deltas on * the same machine are. + * This harness observes main-thread longtasks only. It cannot observe network + * round trips and must never be cited as evidence that a round trip was removed. * * Run it (from desktop/): - * pnpm build + * pnpm build:e2e * npx playwright test --config=playwright.perf.config.ts warm-switch-markdown.perf.ts * * NOTE: the perf web server reuses an existing server on :4173 — if one is