fix(desktop): cancel abandoned thread history on navigation

Signed-off-by: Brain <1a02c72794dcd0f07058a353bc3a81f4028b8c77c92c87fce6d5c8b85970a20b@buzz.block.builderlab.xyz>

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",

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("<!doctype html><html><body></body></html>", {
+  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();
+  }
+});

desktop/src/features/messages/useThreadReplies.ts

@@ -47,6 +47,7 @@ export async function loadThreadReplies(
   expectedEventId?: string | null,
   exhaustedTargets?: Set<string>,
   fetcher: ThreadRepliesFetcher = getThreadReplies,
+  signal?: AbortSignal,
 ): Promise<RelayEvent[]> {
   const queryKey = threadRepliesKey(channelId, rootId);
   const cacheAtStart = queryClient.getQueryData<RelayEvent[]>(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<RelayEvent[]>(queryKey) ?? [];
@@ -130,7 +135,9 @@ export function useThreadReplies(
       activeChannel !== null &&
       activeChannel.channelType !== "forum" &&
       openThreadRootId !== null,
-    queryFn: async (): Promise<RelayEvent[]> => {
+    // 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<RelayEvent[]> => {
       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,
     })),

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);
+});

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.