diff --git a/desktop/playwright.config.ts b/desktop/playwright.config.ts index 41a74c4228..58e90761c8 100644 --- a/desktop/playwright.config.ts +++ b/desktop/playwright.config.ts @@ -22,6 +22,7 @@ export default defineConfig({ "**/smoke.spec.ts", "**/owned-agent-discovery.spec.ts", "**/thread-head-stale-edit.spec.ts", + "**/thread-navigation-freshness.spec.ts", "**/sidebar-offcanvas-rail.spec.ts", "**/tooltip-semantics.spec.ts", "**/search-scope-screenshots.spec.ts", diff --git a/desktop/src/features/messages/threadNavigationFreshness.test.mjs b/desktop/src/features/messages/threadNavigationFreshness.test.mjs new file mode 100644 index 0000000000..df990871ee --- /dev/null +++ b/desktop/src/features/messages/threadNavigationFreshness.test.mjs @@ -0,0 +1,308 @@ +import assert from "node:assert/strict"; +import { after, afterEach, before, mock, test } from "node:test"; +import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; +import { JSDOM } from "jsdom"; +import React from "react"; +import { + useThreadReplies, + useThreadRepliesForRoots, +} from "./useThreadReplies.ts"; +import { useChannelSubscription } from "./hooks.ts"; +import { relayClient } from "../../shared/api/relayClient.ts"; +import { threadRepliesKey } from "./lib/messageQueryKeys.ts"; + +const dom = new JSDOM("", { + url: "http://localhost", +}); +before(() => { + Object.assign(globalThis, { + document: dom.window.document, + HTMLElement: dom.window.HTMLElement, + IS_REACT_ACT_ENVIRONMENT: true, + window: dom.window, + }); +}); +afterEach(() => mock.restoreAll()); +after(() => dom.window.close()); +const channel = { id: "channel-a", channelType: "stream" }; +const rootId = "a".repeat(64); +const oldReply = { + id: "b".repeat(64), + pubkey: "c".repeat(64), + created_at: 100, + kind: 9, + tags: [ + ["h", channel.id], + ["e", rootId, "", "root"], + ], + content: "old reply", + sig: "", +}; +const latestReply = { + ...oldReply, + id: "d".repeat(64), + created_at: 201, + content: "latest reply", +}; +function deferred() { + let resolve; + const promise = new Promise((r) => { + resolve = r; + }); + return { promise, resolve }; +} + +async function mount(fetchReplies, { multiRoot = false } = {}) { + const { act, renderHook, waitFor } = await import("@testing-library/react"); + const client = new QueryClient({ + defaultOptions: { queries: { retry: false, gcTime: Infinity } }, + }); + const frames = []; + const priorWsId = relayClient.wsId; + relayClient.wsId = 7; + window.__TAURI_INTERNALS__ = { + invoke: async (command, args) => { + if (command === "get_thread_replies") return fetchReplies(args); + if (command === "plugin:websocket|send") { + const frame = JSON.parse(args.message.data); + frames.push(frame); + if (frame[0] === "REQ") { + await relayClient.handleWsMessage( + { type: "Text", data: JSON.stringify(["EOSE", frame[1]]) }, + relayClient.connectionGeneration, + ); + } + return; + } + assert.fail(`Unexpected command: ${command}`); + }, + }; + const hook = renderHook( + ({ active, root }) => { + const single = useThreadReplies(active, multiRoot ? null : root); + const multiple = useThreadRepliesForRoots( + active, + multiRoot && root ? [root] : [], + ); + useChannelSubscription(active); + return multiRoot ? multiple : single; + }, + { + initialProps: { active: channel, root: rootId }, + wrapper: ({ children }) => + React.createElement(QueryClientProvider, { client }, children), + }, + ); + return { + hook, + client, + frames, + act, + waitFor, + async dispose() { + hook.unmount(); + client.clear(); + await act(async () => { + await new Promise((r) => setImmediate(r)); + }); + relayClient.wsId = priorWsId; + }, + }; +} + +for (const multiRoot of [false, true]) { + test(`returning ${multiRoot ? "multi-root" : "single-root"} reader does not reuse an abandoned pre-subscription snapshot`, async () => { + let now = 200; + mock.method(Date, "now", () => now * 1000); + const staleSnapshot = deferred(); + let reads = 0; + const h = await mount( + async () => { + reads += 1; + if (reads === 1) return staleSnapshot.promise; + return { events: [oldReply, latestReply], next_cursor: null }; + }, + { multiRoot }, + ); + try { + await h.waitFor(() => assert.equal(reads, 1)); + await h.waitFor(() => + assert.equal(h.frames.filter((f) => f[0] === "REQ").length, 1), + ); + h.hook.rerender({ + active: { id: "channel-b", channelType: "stream" }, + root: null, + }); + // A reply is committed while this channel has no foreground subscription. + // The background unread listener does not populate the thread cache. + now = 202; + h.hook.rerender({ active: channel, root: rootId }); + await h.waitFor(() => + assert.equal(h.frames.filter((f) => f[0] === "REQ").length, 3), + ); + const returningFilter = h.frames.filter((f) => f[0] === "REQ").at(-1)[2]; + assert.equal(returningFilter.since, 202); + assert.ok(latestReply.created_at < returningFilter.since); + // A returned reader must finish without waiting for the abandoned IPC. + await h.waitFor(() => + assert.equal(reads, 2, "return must issue a fresh thread snapshot"), + ); + await h.waitFor(() => + assert.equal( + h.client.getQueryState(threadRepliesKey(channel.id, rootId)).status, + "success", + ), + ); + await h.act(async () => { + staleSnapshot.resolve({ events: [oldReply], next_cursor: null }); + await new Promise((r) => setImmediate(r)); + }); + assert.deepEqual( + h.client + .getQueryData(threadRepliesKey(channel.id, rootId)) + .map((e) => e.id), + [oldReply.id, latestReply.id], + ); + } finally { + await h.act(async () => { + staleSnapshot.resolve({ events: [oldReply], next_cursor: null }); + await new Promise((r) => setImmediate(r)); + }); + await h.dispose(); + } + }); +} + +test("leaving a thread stops its remaining serial pages", async () => { + const firstPage = deferred(); + let reads = 0; + const h = await mount(async () => { + reads += 1; + if (reads === 1) return firstPage.promise; + return { + events: [latestReply], + next_cursor: + reads < 5 + ? { created_at: 201 + reads, event_id: latestReply.id } + : null, + }; + }); + try { + await h.waitFor(() => assert.equal(reads, 1)); + h.hook.rerender({ active: null, root: null }); + await h.act(async () => { + firstPage.resolve({ + events: [oldReply], + next_cursor: { created_at: 100, event_id: oldReply.id }, + }); + await new Promise((r) => setImmediate(r)); + }); + assert.equal( + reads, + 1, + "obsolete navigation must not keep issuing history pages", + ); + } finally { + firstPage.resolve({ events: [], next_cursor: null }); + await h.dispose(); + } +}); + +test("a surviving multi-root consumer keeps shared pagination alive", async () => { + const page = deferred(); + let reads = 0; + const h = await mount(async () => { + reads += 1; + if (reads === 1) return page.promise; + return { events: [latestReply], next_cursor: null }; + }); + const { renderHook } = await import("@testing-library/react"); + const other = renderHook(() => useThreadRepliesForRoots(channel, [rootId]), { + wrapper: ({ children }) => + React.createElement(QueryClientProvider, { client: h.client }, children), + }); + try { + await h.waitFor(() => assert.equal(reads, 1)); + h.hook.rerender({ active: null, root: null }); + await h.act(async () => { + page.resolve({ + events: [oldReply], + next_cursor: { created_at: 100, event_id: oldReply.id }, + }); + }); + await h.waitFor(() => assert.equal(other.result.current.isPending, false)); + assert.equal(reads, 2); + assert.deepEqual( + other.result.current.events.map((e) => e.id), + [oldReply.id, latestReply.id], + ); + } finally { + page.resolve({ events: [], next_cursor: null }); + other.unmount(); + await h.dispose(); + } +}); + +test("multi-root pagination stops when its last consumer leaves", async () => { + const page = deferred(); + let reads = 0; + const h = await mount( + async () => { + reads += 1; + if (reads === 1) return page.promise; + return { events: [latestReply], next_cursor: null }; + }, + { multiRoot: true }, + ); + try { + await h.waitFor(() => assert.equal(reads, 1)); + h.hook.rerender({ active: null, root: null }); + await h.act(async () => { + page.resolve({ + events: [oldReply], + next_cursor: { created_at: 100, event_id: oldReply.id }, + }); + await new Promise((r) => setImmediate(r)); + }); + assert.equal(reads, 1); + } finally { + page.resolve({ events: [], next_cursor: null }); + await h.dispose(); + } +}); + +test("canceling a refresh retains cached replies and live arrivals", async () => { + const page = deferred(); + let reads = 0; + const h = await mount(async () => { + reads += 1; + if (reads === 1) return { events: [oldReply], next_cursor: null }; + return page.promise; + }); + try { + await h.waitFor(() => assert.equal(h.hook.result.current.isSuccess, true)); + await h.act(async () => { + void h.hook.result.current.refetch(); + }); + await h.waitFor(() => assert.equal(reads, 2)); + const key = threadRepliesKey(channel.id, rootId); + // Same cache seam used by live channel delivery while history is pending. + await h.act(async () => { + h.client.setQueryData(key, [oldReply, latestReply]); + }); + h.hook.rerender({ active: null, root: null }); + await h.act(async () => { + page.resolve({ events: [oldReply], next_cursor: null }); + await new Promise((r) => setImmediate(r)); + }); + assert.deepEqual( + h.client.getQueryData(key).map((e) => e.id), + [oldReply.id, latestReply.id], + ); + assert.equal(h.client.getQueryState(key).fetchStatus, "idle"); + assert.equal(h.client.getQueryState(key).status, "success"); + } finally { + page.resolve({ events: [], next_cursor: null }); + await h.dispose(); + } +}); diff --git a/desktop/src/features/messages/useThreadReplies.ts b/desktop/src/features/messages/useThreadReplies.ts index 9ad86e46a8..b9b211b0ee 100644 --- a/desktop/src/features/messages/useThreadReplies.ts +++ b/desktop/src/features/messages/useThreadReplies.ts @@ -47,6 +47,7 @@ export async function loadThreadReplies( expectedEventId?: string | null, exhaustedTargets?: Set, fetcher: ThreadRepliesFetcher = getThreadReplies, + signal?: AbortSignal, ): Promise { const queryKey = threadRepliesKey(channelId, rootId); const cacheAtStart = queryClient.getQueryData(queryKey) ?? []; @@ -54,10 +55,14 @@ export async function loadThreadReplies( const replies: RelayEvent[] = []; let cursor: ThreadCursor | null = null; for (let page = 0; page < MAX_THREAD_PAGES; page += 1) { + signal?.throwIfAborted(); const response = await fetcher(rootId, channelId, { limit: THREAD_PAGE_LIMIT, cursor, }); + // Tauri invokes cannot be cancelled after dispatch. Discard their late + // result and stop pagination when the query no longer has a consumer. + signal?.throwIfAborted(); replies.push(...response.events); if (!response.nextCursor) { const current = queryClient.getQueryData(queryKey) ?? []; @@ -130,7 +135,9 @@ export function useThreadReplies( activeChannel !== null && activeChannel.channelType !== "forum" && openThreadRootId !== null, - queryFn: async (): Promise => { + // Reading signal lets QueryClient cancel an abandoned fetch instead of + // reusing its old snapshot if the user returns after live delivery stopped. + queryFn: async ({ signal }): Promise => { if (!activeChannel || !openThreadRootId) return []; if (expectedEventId) { // Reset the counter when the target changes. @@ -154,6 +161,8 @@ export function useThreadReplies( openThreadRootId, expectedEventId, exhaustedTargetsRef.current, + getThreadReplies, + signal, ); }, staleTime: 0, @@ -259,7 +268,16 @@ export function useThreadRepliesForRoots( queries: rootIds.map((rootId) => ({ queryKey: threadRepliesKey(channelId, rootId), enabled: activeChannel !== null && activeChannel.channelType !== "forum", - queryFn: () => loadThreadReplies(queryClient, channelId, rootId), + queryFn: ({ signal }) => + loadThreadReplies( + queryClient, + channelId, + rootId, + undefined, + undefined, + getThreadReplies, + signal, + ), staleTime: 0, gcTime: 60 * 60 * 1_000, })), diff --git a/desktop/tests/e2e/thread-navigation-freshness.spec.ts b/desktop/tests/e2e/thread-navigation-freshness.spec.ts new file mode 100644 index 0000000000..6beb33e7b4 --- /dev/null +++ b/desktop/tests/e2e/thread-navigation-freshness.spec.ts @@ -0,0 +1,59 @@ +import { expect, test } from "@playwright/test"; + +import { installMockBridge, TEST_IDENTITIES } from "../helpers/bridge"; + +// Hold the actual IPC snapshot across navigation. A reply committed while the +// channel is closed predates the returning live subscription: only a fresh +// history read can recover it. No timeout can release the obsolete snapshot. +test("returning to a thread displays replies missed while its old fetch was pending", async ({ + page, +}) => { + const start = Math.floor(Date.now() / 1000); + await page.clock.setFixedTime(start * 1000); + await installMockBridge(page, { deferThreadReplies: true }); + await page.goto("/"); + await page.getByTestId("channel-general").click(); + await expect(page.getByTestId("chat-title")).toHaveText("general"); + const openThread = async () => { + await page.getByTestId("reply-message-mock-general-welcome").click({ + force: true, + }); + await expect(page.getByTestId("message-thread-panel")).toBeVisible(); + }; + await openThread(); + await expect + .poll(() => + page.evaluate(() => window.__BUZZ_E2E_THREAD_REPLIES_PENDING__?.()), + ) + .toBe(1); + + await page.getByTestId("channel-random").click(); + await expect(page.getByTestId("chat-title")).toHaveText("random"); + await expect(page.getByTestId("message-thread-panel")).toHaveCount(0); + const reply = await page.evaluate( + ({ start, pubkey }) => + window.__BUZZ_E2E_EMIT_MOCK_MESSAGE__?.({ + channelName: "general", + parentEventId: "mock-general-welcome", + content: "Reply committed while away from this thread", + pubkey, + createdAt: start + 1, + }), + { start, pubkey: TEST_IDENTITIES.alice.pubkey }, + ); + expect(reply?.id).toBeTruthy(); + await page.clock.setFixedTime((start + 2) * 1000); + await page.getByTestId("channel-general").click(); + await expect(page.getByTestId("chat-title")).toHaveText("general"); + await openThread(); + // Release old and replacement snapshots together. Buggy code has only the + // old one and renders no latest reply; fixed code discards the old response. + await page.evaluate(() => window.__BUZZ_E2E_RELEASE_THREAD_REPLIES__?.()); + await expect( + page + .getByTestId("message-thread-panel") + .locator(`[data-message-id="${reply?.id}"]`) + .getByTestId("message-body"), + ).toHaveText("Reply committed while away from this thread"); + await expect(page.getByTestId("message-thread-replies-error")).toHaveCount(0); +}); diff --git a/perf/THREAD_NAVIGATION.md b/perf/THREAD_NAVIGATION.md new file mode 100644 index 0000000000..63d3167a43 --- /dev/null +++ b/perf/THREAD_NAVIGATION.md @@ -0,0 +1,87 @@ +# Thread navigation: abandon obsolete history, fetch fresh on return + +## Behavior and scope + +Leaving a thread previously left its history request running. A return before +that request finished reused its old snapshot. Replies committed while the +channel was closed could be newer than that snapshot but older than the returning +live subscription's `since`, so the thread never displayed them. Abandoned +queries also kept dispatching serial history pages after navigation. + +Consume TanStack Query's existing AbortSignal in both single-root and multi-root +thread hooks. Check cancellation before and after each page. The last observer +leaving cancels the query; a remaining observer keeps its pagination alive. +An already-dispatched Tauri invoke cannot be canceled: its result is discarded, +and further pages stop. Returning starts fresh without waiting for the old IPC. +No new timers, caches, schedulers, transport contracts, page sizes, or loading UI. +Production diff: `useThreadReplies.ts`, 20 additions / 2 deletions. + +## Before/after evidence (2026-09-20) + +Control thread implementation: `9fff1acca9b357b1209532de3c502e5d402a3484` +(and identical at `9e5a9518e7f3fb62cdf14dc735d1092ed5844a85`). +Control source SHA-256: `353bcf814d462b50ba9459d43dd082b7bb92e1e7e5c58cbb727a5a05478becfe`. +Candidate source SHA-256: `97b46bc7ba48b237585615554d6d73555dbbd6e82577ef428f62d4b00b8b0cea`. +Dependency inspected: actual installed `@tanstack/query-core` 5.100.14, +`src/query.ts:362-375` (last-observer handling). + +| Controlled workflow | Before | After | +| --- | --- | --- | +| Return while obsolete snapshot remains pending | Reuses old request; latest reply missing | Fresh request completes independently; latest reply present | +| Leave after dispatching page 1 of a five-page thread | 5 history reads | 1 history read | +| Leave multi-root reader after page 1 of two | 2 history reads | 1 history read | +| Another reader remains | All pages delivered | All pages delivered | +| Cancel refresh after live cache update | Replies retained | Replies retained | + +The six new real-hook tests use QueryClient, RelayClient, channel subscription, +and Tauri API adapter with controlled IPC responses. Both return tests explicitly +assert that the returning live filter starts at 202 and the missing reply is at +201. Four fail on the control for the expected fresh-fetch / page-count +assertions; all six pass on the candidate. Nine existing thread tests pass too, +including missing-target retries and cold-fetch target changes. Existing tests +emit React act warnings. A first negative-control cleanup leaked a timer after +an assertion failure; cleanup now flushes the held response before clearing the +client. The final negative control terminates with four failures and two passes. + +The new Playwright regression drives the full rendered app: open thread, hold +its IPC snapshot, switch channels, commit reply, return, release snapshots, +assert reply text in the panel. It fails on the isolated old build because the +reply element is absent, and passes on the candidate. It uses the existing mock +bridge gate, not new production test hooks. The old source/build directory has +separate dependencies and outputs and no shared Git metadata. + +Candidate browser run: 129 passed, 1 skipped across navigation, messaging, +thread unread/error/edits, project conversation failures, Huddle thread failures, +and the new regression. The existing skipped case is direct forum thread links +in navigation.spec.ts. Desktop lint/format checks and both E2E and production +frontend builds passed, with existing style/bundle-size warnings. + +## Reproduce + +Activate repo Hermit, then from `desktop/`: + +```sh +node --import ./test-loader.mjs --experimental-strip-types --test \ + src/features/messages/threadNavigationFreshness.test.mjs \ + src/features/messages/useThreadReplies.test.mjs +pnpm test:e2e:smoke thread-navigation-freshness.spec.ts \ + thread-head-stale-edit.spec.ts thread-unread.spec.ts thread-load-failure.spec.ts \ + project-conversation-load-failure.spec.ts huddle-thread-load-failure.spec.ts \ + navigation.spec.ts messaging.spec.ts +``` + +For a negative control, copy only the new tests / Playwright registration into +an isolated old-code checkout; rebuild its E2E bundle. Verify port 4173 isn't +serving another checkout. Do not change a live application or its caches. + +## Boundaries + +This demonstrates missing-message recovery and less obsolete work, **not** a +measured startup/channel-switch latency improvement or a diagnosis of a specific +live session. Active long threads still load serially oldest-first. The initial +startup subscription workload is unchanged. Transport errors retain existing +retry/error UI. The abort primitive matches the existing channel-history path; +this work does not establish compatibility on every supported native WebKit. +No native launch, live-relay navigation run, or repository-wide `just ci` has been +claimed. Package-wide tests run through the normal push hook, separately from +the focused evidence above. Nothing is merged or deployed by these checks.