diff --git a/docs/relay-queries.md b/docs/relay-queries.md index 91b01b697e..141779696f 100644 --- a/docs/relay-queries.md +++ b/docs/relay-queries.md @@ -365,6 +365,25 @@ roster, applied through shared discovery. That makes the channel a sidebar member and enables the composer, which then receives focus. An accepted but unconfirmed join reports that it may have taken effect and stays retryable. +### Finite read recovery + +A finite read that has no session recovery source retries transient failures +quietly through `read-retry.ts` before its existing Retry action appears. That +covers public search, directory, thread, older-history, profile, catalog, +repository, staff and media reads. The budget is one read plus three retries, +with equal jitter under session recovery's first three ceilings (1/2/4 seconds). +An admission hint is honored within that ceiling; a longer pause surfaces +immediately to the existing cooldown. Exhaustion rethrows the last failure. +Access revocation and cache clears end a session read's backoff under the same +rule that retires its in-flight requests. + +Only one owner retries a given read path. Internal roster, head, catch-up, +preference and template reads keep the session recovery coordinator. Callers +keep their cancellation, access-epoch and deadline guards. Denied, invalid and +deterministic local failures, cancellations and writes never retry. Hidden +conversations keep the last confirmed IDs while refreshing; after exhaustion +the error does not opt into indefinite session recovery. + ### Channel links A channel-only link (`buzz://channel/`) shows what the reader may see about diff --git a/src/app/shell/SearchResults.test.tsx b/src/app/shell/SearchResults.test.tsx index 9c69e3dcc5..492b67ec27 100644 --- a/src/app/shell/SearchResults.test.tsx +++ b/src/app/shell/SearchResults.test.tsx @@ -395,10 +395,14 @@ it("shows the real read failure, retains conversation choices, and retries to an }, ]); await act(async () => - request.fail(new ReadError("unavailable", "Relay read timed out")), + request.fail( + new ReadError("invalid-response", "Relay response was invalid"), + ), ); expect( - screen.getByText(/Message search couldn’t finish: Relay read timed out/), + screen.getByText( + /Message search couldn’t finish: Relay response was invalid/, + ), ).toBeVisible(); expect(screen.getByRole("option", { name: /wes-crew/ })).toBeVisible(); fireEvent.click(screen.getByRole("button", { name: "Retry messages" })); @@ -580,10 +584,14 @@ it("sorts a conversation newest first across pages and restarts paging when the { ...text, search_sort: "newest", page: 2 }, ]); await act(async () => - failed.fail(new ReadError("unavailable", "Relay read timed out")), + failed.fail( + new ReadError("invalid-response", "Relay response was invalid"), + ), ); expect( - screen.getByText("More messages couldn’t load: Relay read timed out"), + screen.getByText( + "More messages couldn’t load: Relay response was invalid", + ), ).toBeVisible(); fireEvent.click(screen.getByRole("option", { name: /Show more messages/ })); const second = await request(); diff --git a/src/bundled/builderlab/adapter-type/Create.test.tsx b/src/bundled/builderlab/adapter-type/Create.test.tsx index 84c6e909ed..06097c0c03 100644 --- a/src/bundled/builderlab/adapter-type/Create.test.tsx +++ b/src/bundled/builderlab/adapter-type/Create.test.tsx @@ -45,7 +45,7 @@ async function fixture(signedIn = true) { const pending = deferred(); const request = vi.fn(async () => { await pending.promise; - if (fail) throw new Error("offline"); + if (fail) return { status: 400, headers: {}, body: "{}" }; return { status: 200, headers: {}, @@ -133,9 +133,7 @@ it("retains instructions when catalog loading fails and is retried", async () => ); await act(async () => h.pending.resolve()); const retry = await screen.findByRole("button", { name: "Retry models" }); - expect(screen.getByRole("alert")).toHaveTextContent( - "Could not reach Builderlab", - ); + expect(screen.getByRole("alert")).toHaveTextContent("HTTP 400"); h.fail(false); await userEvent.setup().click(retry); await screen.findByRole("combobox", { name: "Model" }); diff --git a/src/bundled/builderlab/adapter-type/Instructions.test.tsx b/src/bundled/builderlab/adapter-type/Instructions.test.tsx index 1c5e5585f2..a6c7627d47 100644 --- a/src/bundled/builderlab/adapter-type/Instructions.test.tsx +++ b/src/bundled/builderlab/adapter-type/Instructions.test.tsx @@ -107,11 +107,9 @@ function fixture() { it("retries a failed load and refuses to save a draft through another account", async () => { const h = fixture(); const user = userEvent.setup(); - h.request.mockRejectedValueOnce(new Error("offline")); + h.request.mockResolvedValueOnce({ ...response(), status: 400 }); render(, { wrapper: StrictMode }); - expect(await screen.findByRole("alert")).toHaveTextContent( - "Could not reach Builderlab", - ); + expect(await screen.findByRole("alert")).toHaveTextContent("HTTP 400"); await user.click(screen.getByRole("button", { name: "Retry" })); const field = await screen.findByRole("textbox", { name: "Instructions" }); expect(field).toHaveValue("First instructions."); diff --git a/src/bundled/builderlab/adapter-type/ModelSettings.test.tsx b/src/bundled/builderlab/adapter-type/ModelSettings.test.tsx index 4686704080..796e62e24f 100644 --- a/src/bundled/builderlab/adapter-type/ModelSettings.test.tsx +++ b/src/bundled/builderlab/adapter-type/ModelSettings.test.tsx @@ -34,7 +34,7 @@ async function fixture(selection: Partial = {}) { })); await session.signIn(); const request = vi.fn(async ({ url, body }) => { - if (offline) throw new Error("offline"); + if (offline) return { status: 400, headers: {}, body: "{}" }; let result: unknown; if (url.endsWith("/get-model-catalog")) result = { @@ -192,9 +192,7 @@ it("preserves a failed draft and refuses to save it through a different account" h.offline(true); const user = userEvent.setup(); await user.click(screen.getByRole("button", { name: "Save model" })); - expect(await screen.findByRole("alert")).toHaveTextContent( - "Could not reach Builderlab", - ); + expect(await screen.findByRole("alert")).toHaveTextContent("HTTP 400"); expect( screen.queryByRole("dialog", { name: "Model preferences saved" }), ).not.toBeInTheDocument(); @@ -222,15 +220,11 @@ it("retries a failed catalog refresh without retrying a failed save", async () = await screen.findByRole("button", { name: "Refresh models" }), ); const retry = await screen.findByRole("button", { name: "Retry models" }); - expect(screen.getByRole("alert")).toHaveTextContent( - "Could not reach Builderlab", - ); + expect(screen.getByRole("alert")).toHaveTextContent("HTTP 400"); h.offline(false); await user.click(retry); await screen.findByRole("button", { name: "Refresh models" }); - expect(screen.getByRole("alert")).toHaveTextContent( - "Could not reach Builderlab", - ); + expect(screen.getByRole("alert")).toHaveTextContent("HTTP 400"); expect( screen.getByRole("combobox", { name: "Reasoning effort" }), ).toHaveTextContent("low"); diff --git a/src/bundled/builderlab/agents/RemoteAgents.test.tsx b/src/bundled/builderlab/agents/RemoteAgents.test.tsx index df86140234..ff06d5262c 100644 --- a/src/bundled/builderlab/agents/RemoteAgents.test.tsx +++ b/src/bundled/builderlab/agents/RemoteAgents.test.tsx @@ -65,8 +65,8 @@ it("shows a held loading state and retries a failed read", async () => { expect( screen.getByRole("button", { name: "Refresh agents" }), ).toHaveAttribute("aria-disabled", "true"); - await act(async () => held.resolve(response([], 503))); - expect(screen.getByRole("alert")).toHaveTextContent("HTTP 503"); + await act(async () => held.resolve(response([], 400))); + expect(screen.getByRole("alert")).toHaveTextContent("HTTP 400"); await userEvent.setup().click(screen.getByRole("button", { name: "Retry" })); expect(await screen.findByText("Helper · Active")).toBeInTheDocument(); expect(screen.queryByRole("alert")).not.toBeInTheDocument(); diff --git a/src/bundled/builderlab/agents/client.test.ts b/src/bundled/builderlab/agents/client.test.ts index 1ee27aef5b..05e00496d8 100644 --- a/src/bundled/builderlab/agents/client.test.ts +++ b/src/bundled/builderlab/agents/client.test.ts @@ -140,7 +140,15 @@ it.each([401, 403, 500])( vi.mocked(h.host.request).mockResolvedValue( response({ error: "private-secret" }, status), ); - const error = await h.client.list(h.signal).catch((error: Error) => error); + vi.useFakeTimers(); + let error: unknown; + try { + const pending = h.client.list(h.signal).catch((error: Error) => error); + await vi.advanceTimersByTimeAsync(7000); + error = await pending; + } finally { + vi.useRealTimers(); + } expect(error).toBeInstanceOf(Error); expect(String(error)).not.toContain("private-secret"); expect(h.session.snapshot().status).toBe( @@ -772,3 +780,69 @@ it("shares one binding attempt between overlapping links", async () => { .mock.calls.map(([input]) => input.url.split("/").at(-1)), ).toEqual(["current", "attest-agent"]); }); + +it.each(["network", "429"])( + "quietly recovers read-only Builderlab discovery after %s", + async (failure) => { + const h = await fixture(); + vi.useFakeTimers(); + try { + const request = vi.mocked(h.host.request); + if (failure === "network") + request.mockRejectedValueOnce(new Error("offline")); + else request.mockResolvedValueOnce(response({}, Number(failure))); + const pending = h.client.list(h.signal); + await vi.advanceTimersByTimeAsync(499); + expect(request).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(501); + expect(await pending).toHaveLength(1); + expect(request).toHaveBeenCalledTimes(2); + } finally { + vi.useRealTimers(); + } + }, +); + +it("does not replay a Builderlab mutation after a transport failure", async () => { + const h = await fixture(); + vi.useFakeTimers(); + try { + const request = vi + .mocked(h.host.request) + .mockRejectedValue(new Error("offline")); + await expect(h.client.delete({ id: "agent-1" }, h.signal)).rejects.toThrow( + "Could not reach Builderlab", + ); + await vi.advanceTimersByTimeAsync(7000); + expect(request).toHaveBeenCalledOnce(); + } finally { + vi.useRealTimers(); + } +}); + +it("exhausts Builderlab reads once and cancels the same backoff on account change", async () => { + const h = await fixture(); + vi.useFakeTimers(); + try { + const request = vi + .mocked(h.host.request) + .mockResolvedValue(response({}, 503)); + const rejected = expect(h.client.list(h.signal)).rejects.toThrow( + "HTTP 503", + ); + await vi.advanceTimersByTimeAsync(7000); + await rejected; + expect(request).toHaveBeenCalledTimes(4); + const pending = h.client.list(h.signal); + const cancelled = expect(pending).rejects.toMatchObject({ + name: "AbortError", + }); + await vi.advanceTimersByTimeAsync(0); + h.session.signOut(); + await vi.advanceTimersByTimeAsync(1000); + await cancelled; + expect(request).toHaveBeenCalledTimes(5); + } finally { + vi.useRealTimers(); + } +}); diff --git a/src/bundled/builderlab/agents/client.ts b/src/bundled/builderlab/agents/client.ts index e003076c59..81ece4633c 100644 --- a/src/bundled/builderlab/agents/client.ts +++ b/src/bundled/builderlab/agents/client.ts @@ -1,3 +1,5 @@ +import { transientStatus } from "../../../features/relay/errors"; +import { retryRead } from "../../../features/relay/read-retry"; import { invoke } from "@tauri-apps/api/core"; import { nativeIdentityEnabled } from "../../../features/identity/service"; import type { @@ -98,6 +100,8 @@ const nativeOwner: OwnerIdentity = { invoke("identity_sign_builderlab_binding", { challenge }), }; +class BuilderlabUnavailable extends Error {} + const LINK_ERRORS: Record = { identity_already_bound: "This Builderlab account is linked to a different Buzz identity.", @@ -157,7 +161,7 @@ export function createAgentClient( }); } catch { check(credential, signal); - throw new Error("Could not reach Builderlab. Try again."); + throw new BuilderlabUnavailable("Could not reach Builderlab. Try again."); } check(credential, signal); signOutIfExpired(response); @@ -180,6 +184,26 @@ export function createAgentClient( throw new Error("Builderlab returned an invalid response."); } } + function readRequest( + path: "get-model-catalog" | "list-agents", + body: unknown, + signal: AbortSignal, + ) { + const credential = remoteCredential(); + return retryRead( + (signal) => { + check(credential, signal); + return request(path, body, signal); + }, + signal, + (error) => + error instanceof BuilderlabUnavailable || + (error instanceof Error && + "status" in error && + typeof error.status === "number" && + transientStatus(error.status)), + ); + } /** Builderlab's identity routes answer failures with `{ error: { code } }`. */ async function identityRequest( path: string, @@ -335,7 +359,7 @@ export function createAgentClient( async modelCatalog( signal: AbortSignal, ): Promise { - const result = await request("get-model-catalog", {}, signal); + const result = await readRequest("get-model-catalog", {}, signal); const models = result.models ?? []; if ( !Array.isArray(models) || @@ -595,7 +619,7 @@ export function createAgentClient( }; }, async list(signal: AbortSignal): Promise { - const result = await request( + const result = await readRequest( "list-agents", { include_instructions: true }, signal, diff --git a/src/bundled/channels/useHiddenDms.test.tsx b/src/bundled/channels/useHiddenDms.test.tsx index 8cc390218f..d160d8db3a 100644 --- a/src/bundled/channels/useHiddenDms.test.tsx +++ b/src/bundled/channels/useHiddenDms.test.tsx @@ -300,7 +300,7 @@ it("restores for an offline message that sorts behind the previous head", async expect(restored.result.current.hiddenIds.has("dm")).toBe(false); }); -it("retries a transient hidden-DM history failure", async () => { +it("leaves hidden-DM history retries to the session read", async () => { vi.useFakeTimers(); const h = fixture(); h.failNextRead(); @@ -308,11 +308,10 @@ it("retries a transient hidden-DM history failure", async () => { useHiddenDms("community:alice", h.session, h.list), ); act(() => view.result.current.hide("dm")); - expect(h.read).toHaveBeenCalledTimes(1); await act(async () => { - await vi.advanceTimersByTimeAsync(500); + await vi.advanceTimersByTimeAsync(10_000); }); - expect(h.read).toHaveBeenCalledTimes(2); + expect(h.read).toHaveBeenCalledTimes(1); expect(view.result.current.hiddenIds.has("dm")).toBe(true); }); diff --git a/src/bundled/channels/useHiddenDms.ts b/src/bundled/channels/useHiddenDms.ts index b47a1785fb..195d4b4260 100644 --- a/src/bundled/channels/useHiddenDms.ts +++ b/src/bundled/channels/useHiddenDms.ts @@ -25,19 +25,6 @@ type HiddenDm = { }; const key = "hidden-dms"; -function pause(ms: number, signal: AbortSignal): Promise { - return new Promise((resolve) => { - if (signal.aborted) return resolve(); - const done = () => { - clearTimeout(timer); - signal.removeEventListener("abort", done); - resolve(); - }; - const timer = setTimeout(done, ms); - signal.addEventListener("abort", done, { once: true }); - }); -} - function restore(scope: string): HiddenDm[] { const saved = readView(scope, key, []); if (!Array.isArray(saved)) return []; @@ -209,57 +196,47 @@ export function useHiddenDms( const entry = current.current.find((item) => item.id === id); if (!entry) continue; if (!dmIds.has(id)) continue; - for ( - let attempt = 0; - attempt < 3 && !controller.signal.aborted; - attempt++ - ) { - try { - // Save a deeper initial window so deleting a recent head does not - // make older pre-hide history look like a new message. - const events = await session.read( - [ - { - kinds: [9, 40002, 40008], - "#h": [id], - limit: entry.knownIds ? 50 : 100, - }, - ], + // `session.read` owns the quiet transient retry for this finite read. + try { + // Save a deeper initial window so deleting a recent head does not + // make older pre-hide history look like a new message. + const events = await session.read( + [ { - signal: controller.signal, - priority: "background", + kinds: [9, 40002, 40008], + "#h": [id], + limit: entry.knownIds ? 50 : 100, }, + ], + { + signal: controller.signal, + priority: "background", + }, + ); + if (!current.current.includes(entry)) continue; + const latest = session.unread.snapshot({ + kind: "channel", + channelId: id, + }).latestMessage; + const ids = events.map((event) => event.id); + const changed = + entry.knownIds && + ids.some((eventId) => !entry.knownIds?.includes(eventId)); + if (changed) show([id]); + else if (entry.baseline === undefined || !entry.knownIds) + update( + current.current.map((item) => + item === entry + ? { + ...item, + baseline: latest ?? null, + knownIds: ids, + } + : item, + ), ); - if (current.current.includes(entry)) { - const latest = session.unread.snapshot({ - kind: "channel", - channelId: id, - }).latestMessage; - const ids = events.map((event) => event.id); - const changed = - entry.knownIds && - ids.some((eventId) => !entry.knownIds?.includes(eventId)); - if (changed) { - show([id]); - break; - } - if (entry.baseline === undefined || !entry.knownIds) - update( - current.current.map((item) => - item === entry - ? { - ...item, - baseline: latest ?? null, - knownIds: ids, - } - : item, - ), - ); - } - break; - } catch { - if (attempt < 2) await pause(500 * 2 ** attempt, controller.signal); - } + } catch { + // Keep the hidden entry; the next roster or hide change rechecks it. } } })(); diff --git a/src/bundled/github/GitHubBody.test.tsx b/src/bundled/github/GitHubBody.test.tsx index 0f6e14b8fb..fdc62465f2 100644 --- a/src/bundled/github/GitHubBody.test.tsx +++ b/src/bundled/github/GitHubBody.test.tsx @@ -517,9 +517,9 @@ it("retires pending PR loads, retries errors, and clears media when navigating", ).not.toBeInTheDocument(); expect(screen.getByRole("status")).toHaveTextContent("Loading"); await act(async () => { - second.resolve(new Response(null, { status: 500 })); + second.resolve(new Response(null, { status: 400 })); }); - expect(screen.getByRole("alert")).toHaveTextContent("500"); + expect(screen.getByRole("alert")).toHaveTextContent("400"); fireEvent.click(screen.getByRole("button", { name: "Retry" })); expect(screen.getByRole("status")).toHaveTextContent("Loading"); await act(async () => { diff --git a/src/bundled/github/GitHubPanel.test.tsx b/src/bundled/github/GitHubPanel.test.tsx index 89ad1b4b12..6306bfda6e 100644 --- a/src/bundled/github/GitHubPanel.test.tsx +++ b/src/bundled/github/GitHubPanel.test.tsx @@ -296,7 +296,7 @@ it.each([ }, ); -it.each([500])( +it.each([400])( "keeps the PR title and prominent external recovery action usable after API failure %s", async (status) => { const target = "https://github.com/block/buzz-app/pull/629"; diff --git a/src/bundled/github/data.test.tsx b/src/bundled/github/data.test.tsx index 617c4c6a0a..e21f044ffe 100644 --- a/src/bundled/github/data.test.tsx +++ b/src/bundled/github/data.test.tsx @@ -1,5 +1,7 @@ import { assert, afterEach, expect, it, vi } from "vitest"; import { loadGitHubDetails } from "./data"; +import { GitHubError } from "./gh"; +import { loadGitHubObject } from "./object"; import { parseGitHubReference } from "./references"; afterEach(() => vi.unstubAllGlobals()); @@ -215,3 +217,34 @@ it.each([null, "bad", undefined])( expect(data.mergedByUrl).toBeUndefined(); }, ); + +it.each(["public", "gh"])( + "recovers a transient %s GitHub read quietly", + async (source) => { + vi.useFakeTimers(); + try { + const request = vi + .fn() + .mockResolvedValueOnce(new Response(null, { status: 502 })) + .mockImplementation(async () => Response.json({ title: "Recovered" })); + vi.stubGlobal("fetch", request); + const gh = vi + .fn() + .mockRejectedValueOnce(new GitHubError("down", "SERVICE_UNAVAILABLE")) + .mockRejectedValue(new Error("stop")); + const pending = loadGitHubObject( + reference, + new AbortController().signal, + source === "gh" ? gh : undefined, + ); + const settled = pending.catch((error: unknown) => error); + await vi.advanceTimersByTimeAsync(1000); + if (source === "public") + expect((await pending).details.title).toBe("Recovered"); + else expect(await settled).toMatchObject({ message: "stop" }); + expect(source === "gh" ? gh : request).toHaveBeenCalledTimes(2); + } finally { + vi.useRealTimers(); + } + }, +); diff --git a/src/bundled/github/data.ts b/src/bundled/github/data.ts index afd2618a49..3392cd2dbe 100644 --- a/src/bundled/github/data.ts +++ b/src/bundled/github/data.ts @@ -1,3 +1,4 @@ +import { ReadError, transientStatus } from "../../features/relay/errors"; import type { GitHubReference } from "./references"; export type Branch = { label: string; url?: string | undefined }; @@ -104,7 +105,11 @@ export async function loadGitHubDetails( throw new Error( "GitHub’s public API limit was reached. Try again later or open it on GitHub.", ); - throw new Error(`GitHub couldn’t load this object (${response.status}).`); + const message = `GitHub couldn’t load this object (${response.status}).`; + // `loadGitHubObject` retries server failures quietly. + throw transientStatus(response.status) + ? new ReadError("unavailable", message, response.status) + : new Error(message); } const data: ResponseData = await response.json(); const facts: GitHubDetails["facts"] = []; diff --git a/src/bundled/github/object.ts b/src/bundled/github/object.ts index 223391b818..d8093b01f2 100644 --- a/src/bundled/github/object.ts +++ b/src/bundled/github/object.ts @@ -1,5 +1,6 @@ // What the panel shows for a GitHub link: read with the person's `gh` sign-in // when there is one, so private repositories work, else from the public API. +import { isRetryableRead, retryRead } from "../../features/relay/read-retry"; import { loadGitHubDetails, type GitHubDetails } from "./data"; import { GhUnavailable, GitHubError, type GitHubCli } from "./gh"; import { loadPull, type PullChecklist } from "./pull"; @@ -13,7 +14,21 @@ export type GitHubObject = Readonly<{ signedIn: boolean; }>; -export async function loadGitHubObject( +export function loadGitHubObject( + reference: GitHubReference, + signal: AbortSignal, + gh: GitHubCli | undefined, +): Promise { + return retryRead( + (signal) => readGitHubObject(reference, signal, gh), + signal, + (error) => + isRetryableRead(error) || + (error instanceof GitHubError && + (error.type === "INTERNAL" || error.type === "SERVICE_UNAVAILABLE")), + ); +} +async function readGitHubObject( reference: GitHubReference, signal: AbortSignal, gh: GitHubCli | undefined, diff --git a/src/bundled/pi/tabs.tsx b/src/bundled/pi/tabs.tsx index 11ec9b2700..65eb49dc81 100644 --- a/src/bundled/pi/tabs.tsx +++ b/src/bundled/pi/tabs.tsx @@ -8,6 +8,8 @@ import { useSyncExternalStore, } from "react"; import { errorText } from "../../buzz-mcp/rpc"; +import { isRetryableRead, retryRead } from "../../features/relay/read-retry"; +import { PiStopped } from "./pi"; import activityStyles from "../../features/agents/ActivityLayout.module.css"; import { ActivityTranscript } from "../../features/agents/ActivityTranscript"; import { AgentInstructions } from "../../features/agents/AgentInstructions"; @@ -113,7 +115,12 @@ function useRead( ? {} : { value: prior.value, scope }; setState((prior) => ({ status: "loading", ...kept(prior) })); - read(abort.signal).then( + retryRead( + read, + abort.signal, + // The next request starts another host. + (error) => error instanceof PiStopped || isRetryableRead(error), + ).then( (value) => { if (!abort.signal.aborted) setState({ status: "ready", value, scope }); }, diff --git a/src/bundled/relay-staff/Feedback.tsx b/src/bundled/relay-staff/Feedback.tsx index 969b4d57eb..115d859ac5 100644 --- a/src/bundled/relay-staff/Feedback.tsx +++ b/src/bundled/relay-staff/Feedback.tsx @@ -21,7 +21,13 @@ import type { StaffFailure, StaffRequest, } from "../../features/relay-staff/contract"; -import { describe, useRead, useSession, useWrite } from "./session"; +import { + describe, + retryStaffRead, + useRead, + useSession, + useWrite, +} from "./session"; import { CommunityBadge, Failure, @@ -289,8 +295,12 @@ function AttachmentView({ attachment }: { attachment: AttachmentRef }) { if (!image) return; let url: string | null = null; let current = true; - void session.attachment(attachment).then((outcome) => { - if (!current) return; + const controller = new AbortController(); + void retryStaffRead( + () => session.attachment(attachment), + controller.signal, + ).then((outcome) => { + if (!outcome || !current) return; setRetrying(false); if (!outcome.ok) return setPreview({ failure: outcome.failure }); url = URL.createObjectURL( @@ -301,6 +311,7 @@ function AttachmentView({ attachment }: { attachment: AttachmentRef }) { // The object URL is released when the attachment closes or unmounts. return () => { current = false; + controller.abort(); if (url) URL.revokeObjectURL(url); }; }, [session, attachment.feedbackId, attachment.sha256, image, attempt]); diff --git a/src/bundled/relay-staff/read-retry.test.ts b/src/bundled/relay-staff/read-retry.test.ts new file mode 100644 index 0000000000..1130b253e4 --- /dev/null +++ b/src/bundled/relay-staff/read-retry.test.ts @@ -0,0 +1,74 @@ +import { afterEach, beforeEach, expect, it, vi } from "vitest"; +import type { + StaffFailure, + StaffOutcome, +} from "../../features/relay-staff/contract"; +import { retryStaffRead } from "./session"; + +beforeEach(() => { + vi.useFakeTimers(); + vi.spyOn(Math, "random").mockReturnValue(0.5); +}); +afterEach(() => { + vi.restoreAllMocks(); + vi.useRealTimers(); +}); +function failure( + category: StaffFailure["category"], + status: number | null, +): StaffOutcome { + return { + ok: false, + failure: { + category, + status, + bodyComplete: false, + bodyEmpty: false, + code: null, + notSent: false, + authLost: status === 403, + message: "Existing failure", + }, + }; +} +it("quietly recovers intercepted staff reads using one bounded budget", async () => { + const read = vi + .fn() + .mockResolvedValueOnce(failure("intercepted", 403)) + .mockResolvedValue({ ok: true, value: "loaded" }); + const pending = retryStaffRead(read, new AbortController().signal); + await vi.advanceTimersByTimeAsync(749); + expect(read).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(1); + await expect(pending).resolves.toEqual({ ok: true, value: "loaded" }); +}); +it("returns the original staff failure after four reads", async () => { + const original = failure("ambiguous", 502); + const read = vi.fn().mockResolvedValue(original); + const pending = retryStaffRead(read, new AbortController().signal); + await vi.advanceTimersByTimeAsync(5250); + expect(await pending).toEqual(original); + expect(read).toHaveBeenCalledTimes(4); +}); +it.each(["unauthorized", "forbidden", "rejected", "notSent"] as const)( + "leaves %s failures explicit", + async (category) => { + const original = failure(category, 403); + const read = vi.fn().mockResolvedValue(original); + expect(await retryStaffRead(read, new AbortController().signal)).toBe( + original, + ); + expect(read).toHaveBeenCalledOnce(); + }, +); +it("scope retirement releases staff backoff without another request", async () => { + const read = vi.fn().mockResolvedValue(failure("ambiguous", 502)); + const controller = new AbortController(); + const pending = retryStaffRead(read, controller.signal); + await vi.advanceTimersByTimeAsync(0); + controller.abort(); + expect(await pending).toBeUndefined(); + await vi.advanceTimersByTimeAsync(7000); + expect(read).toHaveBeenCalledOnce(); + expect(vi.getTimerCount()).toBe(0); +}); diff --git a/src/bundled/relay-staff/session.tsx b/src/bundled/relay-staff/session.tsx index 1305a8aa95..c69f43d244 100644 --- a/src/bundled/relay-staff/session.tsx +++ b/src/bundled/relay-staff/session.tsx @@ -20,6 +20,57 @@ import type { } from "../../features/relay-staff/contract"; import type { Staff } from "./staff"; import type { Held, Writes } from "./writes"; +import { retryRead } from "../../features/relay/read-retry"; +import { transientStatus } from "../../features/relay/errors"; + +const READ_ROUTES = new Set([ + "probe", + "listReports", + "getReport", + "listFeedback", + "getFeedback", + "listOperators", + "listRestrictions", + "listCommunities", + "searchMembers", + "getMember", + "getEvent", +]); +class StaffReadFailure extends Error { + constructor(readonly failure: StaffFailure) { + super(failure.message); + } +} +function transientRead(failure: StaffFailure) { + return ( + failure.category === "intercepted" || + (failure.category === "ambiguous" && + (failure.status === null || transientStatus(failure.status))) + ); +} +/** Only read owners call this; frozen write intents keep their explicit retry. */ +export async function retryStaffRead( + read: () => Promise>, + signal: AbortSignal, +): Promise | undefined> { + try { + return await retryRead( + async () => { + const outcome = await read(); + if (!outcome.ok && transientRead(outcome.failure)) + throw new StaffReadFailure(outcome.failure); + return outcome; + }, + signal, + (error) => error instanceof StaffReadFailure, + ); + } catch (error) { + if (signal.aborted) return; + if (error instanceof StaffReadFailure) + return { ok: false, failure: error.failure }; + throw error; + } +} export type Session = { staff: Staff; @@ -127,13 +178,20 @@ export function useRead( setRead(({ settled }) => ({ state: "loading", settled: settled + 1 })); return; } - let current = true; - setRead(({ settled }) => ({ - state: "loading", - settled: settled + rescoped, - })); - void send(request).then((outcome) => { - if (!current) return; + const controller = new AbortController(); + setRead((previous) => + !rescoped && previous.state === "ok" + ? previous + : { + state: "loading", + settled: previous.settled + rescoped, + }, + ); + const reading = READ_ROUTES.has(request.route) + ? retryStaffRead(() => send(request), controller.signal) + : send(request); + void reading.then((outcome) => { + if (!outcome || controller.signal.aborted) return; setRead(({ settled }) => outcome.ok ? { state: "ok", value: outcome.value, settled: settled + 1 } @@ -141,7 +199,7 @@ export function useRead( ); }); return () => { - current = false; + controller.abort(); }; }, [send, attempt, ...deps]); const reload = useCallback(() => setAttempt((n) => n + 1), []); @@ -232,14 +290,23 @@ export function usePages( settled: 0, }); const generation = useRef(0); + const pending = useRef(undefined); const pageRef = useRef(page); pageRef.current = page; const load = async (cursor: string | undefined, at: number) => { const make = pageRef.current; - if (!make) return; + if (!make || pending.current) return; + const controller = new AbortController(); + pending.current = controller; setPages((current) => ({ ...current, loading: true, failure: null })); - const outcome = await send(make(cursor)); - if (at !== generation.current) return; + const outcome = await retryStaffRead( + () => send(make(cursor)), + controller.signal, + ).finally(() => { + if (pending.current === controller) pending.current = undefined; + }); + if (!outcome || controller.signal.aborted || at !== generation.current) + return; setPages((current) => outcome.ok ? { @@ -262,6 +329,8 @@ export function usePages( }; // biome-ignore lint/correctness/useExhaustiveDependencies: `deps` describe `page` useEffect(() => { + pending.current?.abort(); + pending.current = undefined; const at = ++generation.current; // A new scope (or a disabled one) ends any attempt kept from the old one. setPages(({ settled }) => ({ @@ -271,6 +340,8 @@ export function usePages( })); void load(undefined, at); return () => { + pending.current?.abort(); + pending.current = undefined; generation.current++; }; }, [send, ...deps]); @@ -278,6 +349,8 @@ export function usePages( if (pages.next && !pages.loading) void load(pages.next, generation.current); }; const restart = () => { + pending.current?.abort(); + pending.current = undefined; const at = ++generation.current; void load(undefined, at); }; diff --git a/src/bundled/voice-notes/VoiceNoteCard.tsx b/src/bundled/voice-notes/VoiceNoteCard.tsx index 2aa365995d..38b9597e7d 100644 --- a/src/bundled/voice-notes/VoiceNoteCard.tsx +++ b/src/bundled/voice-notes/VoiceNoteCard.tsx @@ -1,3 +1,4 @@ +import { useMediaReadRecovery } from "../../features/messages/use-media-read-recovery"; import { useMediaElementSource } from "../../features/messages/use-media-element-source"; import { MediaDownloadAction } from "../../features/messages/MediaDownloadAction"; import { AttachmentDownloadContext } from "../../features/messages/AttachmentDownloadContext"; @@ -80,6 +81,18 @@ function Player({ const [loading, setLoading] = useState(false); const [error, setError] = useState(false); const [rate, setRate] = useState(1); + const { recover: recoverMedia, cancel: cancelMedia } = useMediaReadRecovery( + active ? source : undefined, + () => { + setError(true); + setLoading(false); + setPlaying(false); + }, + () => { + setError(false); + setLoading(false); + }, + ); const peaks = waveform?.length ? waveformPeaks(Float32Array.from(waveform.slice(0, 100)), count) : Array.from({ length: count }, () => 0); @@ -139,11 +152,13 @@ function Player({ if (!element || !source) return; const operation = ++attempt.current; if (loading || !element.paused) { + cancelMedia(); element.pause(); setLoading(false); return; } claimAudio(element, () => { + cancelMedia(); attempt.current++; element.pause(); setLoading(false); @@ -155,7 +170,7 @@ function Player({ await element.play(); if (attempt.current === operation) setLoading(false); } catch { - if (attempt.current === operation) { + if (attempt.current === operation && !recoverMedia(element)) { setLoading(false); releaseAudio(element); setError(true); @@ -319,6 +334,7 @@ function Player({ paintProgress(0, duration); }} onError={(event) => { + if (recoverMedia(event.currentTarget)) return; setMetadataReady(false); releaseAudio(event.currentTarget); attempt.current++; diff --git a/src/features/agents/agent-config-history.ts b/src/features/agents/agent-config-history.ts index 7e3ad05365..b438b3d059 100644 --- a/src/features/agents/agent-config-history.ts +++ b/src/features/agents/agent-config-history.ts @@ -1,6 +1,7 @@ // NIP-AP config history (plan §7): offset pages of one artifact's revisions, // every chain segment they form, the cause of each gap, diffs and revert. // Display only: history never decides artifact state (§3.2). +import { retryRead } from "../relay/read-retry"; import type { EventData, RelayEvent } from "../relay/events"; import { type SettingsSchema, syncedSettings } from "./agent-config"; import { @@ -49,18 +50,23 @@ export async function loadOlder( if (history.done || history.next > HISTORY_MAX_OFFSET) return { ...history, failed: false }; try { - const page = await read( - [ - { - kinds: [ARTIFACT_KIND], - "#h": [channel], - "#d": [d], - artifact: "history", - limit: HISTORY_PAGE, - offset: history.next, - }, - ], - { signal }, + const active = signal ?? new AbortController().signal; + const page = await retryRead( + (signal) => + read( + [ + { + kinds: [ARTIFACT_KIND], + "#h": [channel], + "#d": [d], + artifact: "history", + limit: HISTORY_PAGE, + offset: history.next, + }, + ], + { signal }, + ), + active, ); const seen = new Set(history.revisions.map((event) => event.id)); return { diff --git a/src/features/agents/inventory.test.ts b/src/features/agents/inventory.test.ts index 78a41e0ded..df2838d019 100644 --- a/src/features/agents/inventory.test.ts +++ b/src/features/agents/inventory.test.ts @@ -92,6 +92,30 @@ it("keeps either available source visible and makes partial failure explicit", a ); }); +it("quietly retries a transient relay inventory failure before merging", async () => { + vi.useFakeTimers(); + vi.spyOn(Math, "random").mockReturnValue(0.5); + try { + const identity = { pubkey: "b".repeat(64), name: "Relay" }; + const relay = vi + .fn() + .mockRejectedValueOnce(new TypeError("offline")) + .mockResolvedValue({ definitions: [], identities: [identity] }); + const pending = inventoryReader( + async () => ({ definitions: [], identities: [] }), + relay, + )(new AbortController().signal); + await vi.advanceTimersByTimeAsync(750); + const result = await pending; + expect(relay).toHaveBeenCalledTimes(2); + expect(result.identities).toEqual([identity]); + expect(result).not.toHaveProperty("error"); + } finally { + vi.restoreAllMocks(); + vi.useRealTimers(); + } +}); + it("actual session reads owner-signed relay inventory and clears/disposes it", async () => { const viewer = keypair(); const query = vi.fn().mockResolvedValue([ diff --git a/src/features/agents/inventory.ts b/src/features/agents/inventory.ts index 2d08cd7f53..cddf524645 100644 --- a/src/features/agents/inventory.ts +++ b/src/features/agents/inventory.ts @@ -1,5 +1,6 @@ import type { AgentLibrary, AgentLibraryReader } from "./library"; import { definitionSlug } from "./relay-library"; +import { retryRead } from "../relay/read-retry"; /** Preserve local names/artwork; add relay-only keys. Names never join identities. */ export function combineInventory( @@ -66,7 +67,8 @@ export function inventoryReader( const empty = { definitions: [], identities: [] }; const [saved, remote] = await Promise.allSettled([ local ? local(signal) : Promise.resolve(empty), - relay(signal), + // Only the relay source is a finite network read; it owns its retries. + retryRead(relay, signal), ]); signal.throwIfAborted(); if (remote.status === "rejected" && (!local || saved.status === "rejected")) diff --git a/src/features/agents/memory.ts b/src/features/agents/memory.ts index 63046b8847..2daa876341 100644 --- a/src/features/agents/memory.ts +++ b/src/features/agents/memory.ts @@ -1,4 +1,5 @@ /** Owner-view NIP-AE snapshot from a purpose-bound key-owning host. */ +import { retryRead } from "../relay/read-retry.ts"; /** Budgets of the development broker's single memory query. */ export const MEMORY_EVENT_LIMIT = 256; export const MEMORY_WIRE_BYTES = 2 * 1024 * 1024; @@ -171,7 +172,9 @@ export function createAgentMemories( owned.signal, AbortSignal.timeout(10000), ]); - const listing = memoryListing(await source(agent, signal)); + const listing = memoryListing( + await retryRead((signal) => source(agent, signal), signal), + ); signal.throwIfAborted(); if (!owned.signal.aborted) publish( diff --git a/src/features/channel-members/administration.ts b/src/features/channel-members/administration.ts index b27c9d9261..2920aed872 100644 --- a/src/features/channel-members/administration.ts +++ b/src/features/channel-members/administration.ts @@ -1,5 +1,6 @@ import { getEventHash } from "nostr-tools"; import type { RelayReader } from "../relay/reader"; +import { retryRead } from "../relay/read-retry"; import type { RelayWriter } from "../relay/transport"; import { newer, type RelayEvent } from "../relay/events"; import { attestedOwner } from "../agents/owner-attestation"; @@ -128,7 +129,7 @@ export function createMemberAdministration({ const previous = snapshot(id); emit(id, { ...previous, status: "loading" }); try { - const result = await read(id, signal); + const result = await retryRead(() => read(id, signal), signal); const operation = previous.operation; const unresolved = operation?.status === "uncertain"; const confirmed = diff --git a/src/features/channel-navigation/ChannelInventory.test.tsx b/src/features/channel-navigation/ChannelInventory.test.tsx index ede20ade0e..8782ee1181 100644 --- a/src/features/channel-navigation/ChannelInventory.test.tsx +++ b/src/features/channel-navigation/ChannelInventory.test.tsx @@ -15,6 +15,8 @@ import { DiscoveryState } from "../relay/discovery"; import { keypair, metadata, roster, signed } from "../relay/testing"; import { huddleDescription } from "../huddle/lifecycle"; import { createRelaySession } from "../relay/session"; +import { createChannelLifecycle } from "../relay/channel-lifecycle"; +import { ReadError } from "../relay/errors"; import { createSidebarPreferencesStore } from "../relay/sidebar-preferences-store"; import type { SidebarGroups, @@ -146,6 +148,12 @@ function fixture( }; for (const listener of listeners) listener(); }); + // A discovery commit, as live roster events produce. + const commit = () => + act(() => { + list = { ...list, asOf: (list.asOf ?? 0) + 1 }; + for (const listener of listeners) listener(); + }); const drop = (id: string) => act(() => { list = { @@ -216,6 +224,7 @@ function fixture( list, session, publish, + commit, drop, connection(next: RelaySnapshot) { act(() => { @@ -226,6 +235,86 @@ function fixture( }; } +function visibilityFixture() { + const f = fixture(); + const read = vi + .fn() + .mockRejectedValue(new ReadError("unavailable", "offline", 503)); + const lifecycle = createChannelLifecycle({ + reader: { read }, + viewer: f.session.viewer, + relayAuthor: "b".repeat(64), + canAccess: () => true, + acceptDiscovery() {}, + removed() {}, + }); + f.session.channelLifecycle = lifecycle.capability; + return { f, read, lifecycle }; +} + +it("keeps visibility recovery quiet, then offers the existing retry after exhaustion", async () => { + vi.useFakeTimers(); + const random = vi.spyOn(Math, "random").mockReturnValue(0.5); + const { f, read, lifecycle } = visibilityFixture(); + try { + render(f.view("alpha")); + await act(() => vi.advanceTimersByTimeAsync(0)); + expect(read).toHaveBeenCalledOnce(); + expect( + screen.queryByText("Hidden conversations could not be refreshed."), + ).not.toBeInTheDocument(); + expect( + screen.queryByRole("button", { name: "Retry hidden conversations" }), + ).not.toBeInTheDocument(); + f.commit(); + await act(() => vi.advanceTimersByTimeAsync(1000)); + f.commit(); + await act(() => vi.advanceTimersByTimeAsync(4250)); + expect(read).toHaveBeenCalledTimes(4); + expect( + screen.getByText("Hidden conversations could not be refreshed."), + ).toBeVisible(); + read.mockResolvedValue([]); + fireEvent.click( + screen.getByRole("button", { name: "Retry hidden conversations" }), + ); + await act(() => vi.advanceTimersByTimeAsync(0)); + expect(read).toHaveBeenCalledTimes(5); + expect( + screen.queryByRole("button", { name: "Retry hidden conversations" }), + ).not.toBeInTheDocument(); + } finally { + cleanup(); + lifecycle.dispose(); + random.mockRestore(); + vi.useRealTimers(); + } +}); + +it("cancels visibility retries on sidebar unmount and session changes", async () => { + vi.useFakeTimers(); + const { f, read, lifecycle } = visibilityFixture(); + try { + const mounted = render(f.view("alpha")); + await act(() => vi.advanceTimersByTimeAsync(0)); + expect(read).toHaveBeenCalledOnce(); + const next = fixture(); + mounted.rerender(next.view("beta")); + await act(() => vi.advanceTimersByTimeAsync(120_000)); + expect(read).toHaveBeenCalledOnce(); + mounted.rerender(f.view("alpha")); + await act(() => vi.advanceTimersByTimeAsync(0)); + expect(read).toHaveBeenCalledTimes(2); + mounted.unmount(); + await act(() => vi.advanceTimersByTimeAsync(120_000)); + expect(read).toHaveBeenCalledTimes(2); + } finally { + cleanup(); + lifecycle.dispose(); + vi.useRealTimers(); + } +}); + it.each(["Retry", "Dismiss"] as const)( "%s restores mute recovery focus after cached-to-live session handoff without remounting", async (action) => { diff --git a/src/features/channel-navigation/ChannelInventory.tsx b/src/features/channel-navigation/ChannelInventory.tsx index 2077bf2f54..28dfeec8b4 100644 --- a/src/features/channel-navigation/ChannelInventory.tsx +++ b/src/features/channel-navigation/ChannelInventory.tsx @@ -145,9 +145,19 @@ export function ChannelInventory({ lifecycle.snapshot, lifecycle.snapshot, ); + const visibilityRequest = useRef(undefined); + // Roster commits join an in-flight refresh; only unmount or a new lifecycle + // cancels it, so live discovery cannot restart its retry budget. + // biome-ignore lint/correctness/useExhaustiveDependencies: a new lifecycle retires the previous refresh signal. + useEffect(() => { + const request = new AbortController(); + visibilityRequest.current = request; + return () => request.abort(); + }, [lifecycle]); // biome-ignore lint/correctness/useExhaustiveDependencies: a completed roster refresh also refreshes per-viewer visibility. useEffect(() => { - if (list.status === "ready") void lifecycle.refreshVisibility(); + if (list.status === "ready") + void lifecycle.refreshVisibility(visibilityRequest.current?.signal); }, [lifecycle, list.asOf, list.status]); const lifecycleFocus = useRef(undefined); const lifecycleTrigger = useRef(undefined); @@ -1030,16 +1040,20 @@ export function ChannelInventory({ ))} - {!cached && - dmVisibility.status === "error" && - !dmVisibility.transient && ( -
- Hidden conversations could not be refreshed.{" "} - -
- )} + {!cached && dmVisibility.status === "error" && ( +
+ Hidden conversations could not be refreshed.{" "} + +
+ )} ), content: ( diff --git a/src/features/channels/usePublicChannelSearch.ts b/src/features/channels/usePublicChannelSearch.ts index ecb51ea953..bff6b1fb4c 100644 --- a/src/features/channels/usePublicChannelSearch.ts +++ b/src/features/channels/usePublicChannelSearch.ts @@ -1,3 +1,4 @@ +import { retryRead } from "../relay/read-retry"; import { useCallback, useEffect, useMemo, useState } from "react"; import type { ChannelSummary } from "../relay/contracts"; import type { ComposerSession } from "../messages/composer-session"; @@ -38,21 +39,25 @@ export function usePublicChannelSearch( let kept = none; // Typeahead waits for a brief typing pause; cancellation also owns the delay. const timer = setTimeout(() => { - search(query, { - signal: controller.signal, - priority: "foreground", - exact, - // An exact lookup names one channel, so it waits for every page. - progress: exact - ? undefined - : ({ channels }) => { - const ids = channels.map(({ id }) => id).join(); - if (controller.signal.aborted || ids === shown) return; - shown = ids; - kept = channels; - setResult({ owner, channels, partial: false, pending: true }); - }, - }).then( + retryRead( + (signal) => + search(query, { + signal, + priority: "foreground", + exact, + // An exact lookup names one channel, so it waits for every page. + progress: exact + ? undefined + : ({ channels }) => { + const ids = channels.map(({ id }) => id).join(); + if (controller.signal.aborted || ids === shown) return; + shown = ids; + kept = channels; + setResult({ owner, channels, partial: false, pending: true }); + }, + }), + controller.signal, + ).then( ({ channels, partial }) => { if (!controller.signal.aborted) setResult({ owner, channels, partial }); diff --git a/src/features/communities/api.ts b/src/features/communities/api.ts index 17a45753ec..89584b5194 100644 --- a/src/features/communities/api.ts +++ b/src/features/communities/api.ts @@ -1,4 +1,5 @@ import { nativeIdentityEnabled } from "../identity/service"; +import { retryRead } from "../relay/read-retry"; import { nativeCommunityRequest } from "./native-api"; import { connectCommunityTransport, registerCommunity } from "./connection"; import { settledRefusal, type SettledRefusal } from "./leave-protocol"; @@ -55,7 +56,10 @@ export async function inspectProfile(id: string, session?: RelaySession) { // update its shared profile directory. Joining uses the selected host adapter. const transport = session ?? - (await connectCommunityTransport(id, AbortSignal.timeout(12000))); + (await retryRead( + (signal) => connectCommunityTransport(id, signal), + AbortSignal.timeout(12000), + )); if (!transport.viewer) throw new Error("Profile identity is unavailable"); const filters = [ { @@ -71,7 +75,10 @@ export async function inspectProfile(id: string, session?: RelaySession) { signal: AbortSignal.timeout(12000), fresh: true, }) - : await transport.query(filters, AbortSignal.timeout(12000)); + : await retryRead( + (signal) => transport.query(filters, signal), + AbortSignal.timeout(12000), + ); const event = events .filter((e) => e.kind === 0 && e.pubkey === transport.viewer) .sort((a, b) => b.created_at - a.created_at || a.id.localeCompare(b.id))[0]; diff --git a/src/features/messages/MediaAttachment.tsx b/src/features/messages/MediaAttachment.tsx index fcfc85c542..d6f9de3e4b 100644 --- a/src/features/messages/MediaAttachment.tsx +++ b/src/features/messages/MediaAttachment.tsx @@ -1,3 +1,4 @@ +import { useMediaReadRecovery } from "./use-media-read-recovery"; import { useConversationPresentation } from "../conversation/ConversationPresentation"; import { prepareReviewEntrance } from "./use-review-entrance"; import { useMediaCorners } from "./use-media-corners"; @@ -129,6 +130,10 @@ export function MediaAttachment({ attachment.kind === "video" ? source : undefined, ); const videoSource = element.src; + const { recover: recoverMedia } = useMediaReadRecovery( + active ? videoSource : undefined, + () => setFailed(true), + ); if (!source && !compact) return ( @@ -312,7 +317,9 @@ export function MediaAttachment({ } } }} - onError={() => setFailed(true)} + onError={(event) => { + if (!recoverMedia(event.currentTarget)) setFailed(true); + }} onPause={() => setPlaying(false)} onPlay={() => { setStarted(true); diff --git a/src/features/messages/VideoPlayer.tsx b/src/features/messages/VideoPlayer.tsx index 3e1954f7cc..9e97209526 100644 --- a/src/features/messages/VideoPlayer.tsx +++ b/src/features/messages/VideoPlayer.tsx @@ -1,3 +1,4 @@ +import { useMediaReadRecovery } from "./use-media-read-recovery"; import { useVideoGestures } from "./use-video-gestures"; import { useMediaControls } from "./use-media-controls"; import { useMediaCorners } from "./use-media-corners"; @@ -446,6 +447,10 @@ export function VideoPlayer({ // Callers render their unavailable state instead of a player when // `useMediaElementSource(source).unavailable`. const { src: elementSource } = useMediaElementSource(source); + const { recover: recoverMedia } = useMediaReadRecovery( + elementSource, + onError, + ); return (
onTime(event.currentTarget.currentTime)} - onError={onError} + onError={(event) => { + if (!recoverMedia(event.currentTarget)) onError(); + }} /> {feedback && (
{ + vi.useFakeTimers(); + vi.spyOn(Math, "random").mockReturnValue(0.5); + vi.spyOn(HTMLMediaElement.prototype, "load").mockImplementation(() => {}); +}); +afterEach(() => { + cleanup(); + vi.restoreAllMocks(); + vi.useRealTimers(); +}); +function setup() { + const failed = vi.fn(), + recovered = vi.fn(); + function Probe({ source }: { source: string }) { + const { recover } = useMediaReadRecovery(source, failed, recovered); + return ( + + ); + } + const mounted = render(); + const element = mounted.container.querySelector("audio"); + if (!element) throw new Error("Audio fixture missing"); + const error = (code = 2) => { + Object.defineProperty(element, "error", { + configurable: true, + value: { code }, + }); + fireEvent.error(element); + }; + return { + ...mounted, + element, + error, + failed, + recovered, + change: () => mounted.rerender(), + }; +} +async function advance(ms: number) { + await act(() => vi.advanceTimersByTimeAsync(ms)); +} + +it("reloads a network failure quietly after backoff and preserves explicit playback", async () => { + const h = setup(); + const play = vi.spyOn(h.element, "play"); + h.error(); + await advance(749); + expect(h.element.load).not.toHaveBeenCalled(); + expect(h.failed).not.toHaveBeenCalled(); + await advance(1); + expect(h.element.load).toHaveBeenCalledOnce(); + expect(h.element.preload).toBe("metadata"); + fireEvent.loadedMetadata(h.element); + await advance(0); + expect(h.recovered).toHaveBeenCalledOnce(); + expect(h.element.preload).toBe("none"); + expect(h.failed).not.toHaveBeenCalled(); + expect(play).not.toHaveBeenCalled(); +}); + +it("surfaces a later failure of a recovered source instead of reloading again", async () => { + const h = setup(); + h.error(); + await advance(750); + fireEvent.loadedMetadata(h.element); + await advance(0); + expect(h.recovered).toHaveBeenCalledOnce(); + h.error(); + await advance(7000); + expect(h.element.load).toHaveBeenCalledOnce(); + expect(h.failed).toHaveBeenCalledOnce(); +}); + +it("counts the initial browser failure and exposes failure after three reloads", async () => { + const h = setup(); + h.error(); + for (const [index, delay] of [750, 1500, 3000].entries()) { + await advance(delay); + expect(h.element.load).toHaveBeenCalledTimes(index + 1); + h.error(); + } + await advance(0); + expect(h.failed).toHaveBeenCalledOnce(); + expect(vi.getTimerCount()).toBe(0); +}); + +it.each(["unmount", "source"])("%s cancels future reloads", async (action) => { + const h = setup(); + h.error(); + await advance(0); + if (action === "unmount") h.unmount(); + else h.change(); + await advance(7000); + expect(h.element.load).not.toHaveBeenCalled(); + expect(h.failed).not.toHaveBeenCalled(); + expect(vi.getTimerCount()).toBe(0); +}); + +it.each([1, 3, 4])("leaves media error %s explicit", async (code) => { + const h = setup(); + h.error(code); + await advance(7000); + expect(h.failed).toHaveBeenCalledOnce(); + expect(h.element.load).not.toHaveBeenCalled(); +}); diff --git a/src/features/messages/use-media-read-recovery.ts b/src/features/messages/use-media-read-recovery.ts new file mode 100644 index 0000000000..8acf27997f --- /dev/null +++ b/src/features/messages/use-media-read-recovery.ts @@ -0,0 +1,96 @@ +import { useEffect, useRef } from "react"; +import { retryRead } from "../relay/read-retry"; + +function mediaFailure(element: HTMLMediaElement): Error { + return element.error?.code === 2 + ? new TypeError("Media network read failed") + : new Error("Media could not be decoded"); +} + +/** Reload bytes after a network failure. Playback permission and decoding + * failures remain explicit; recovery never starts playback itself. */ +export function useMediaReadRecovery( + source: string | undefined, + failed: () => void, + recovered: () => void = () => {}, +) { + const pending = useRef(undefined); + const spent = useRef(undefined); + const callbacks = useRef({ failed, recovered }); + callbacks.current = { failed, recovered }; + useEffect(() => { + void source; + return () => { + pending.current?.abort(); + pending.current = undefined; + }; + }, [source]); + const recover = (element: HTMLMediaElement): boolean => { + if (pending.current) return true; + if (!source || element.error?.code !== 2) return false; + // One quiet budget per source: loaded metadata does not prove the body + // downloads, so a later failure surfaces instead of reloading forever. + if (spent.current === source) return false; + spent.current = source; + const controller = new AbortController(); + pending.current = controller; + const initial = mediaFailure(element); + const preload = element.preload; + let first = true; + void retryRead((signal) => { + // The browser already issued the initial request. Count its failure + // before scheduling the same three quiet retries used by other reads. + if (first) { + first = false; + return Promise.reject(initial); + } + return new Promise((resolve, reject) => { + const finish = (error?: unknown) => { + clearTimeout(timer); + element.removeEventListener("loadedmetadata", ready); + element.removeEventListener("error", errorEvent); + signal.removeEventListener("abort", abort); + if (error) reject(error); + else resolve(); + }; + const ready = () => finish(); + const errorEvent = () => finish(mediaFailure(element)); + const abort = () => finish(signal.reason); + const timer = setTimeout( + () => + finish(new DOMException("Media read timed out", "TimeoutError")), + 10_000, + ); + element.addEventListener("loadedmetadata", ready, { once: true }); + element.addEventListener("error", errorEvent, { once: true }); + signal.addEventListener("abort", abort, { once: true }); + try { + if (preload === "none") element.preload = "metadata"; + element.load(); + } catch (error) { + finish(error); + } + }); + }, controller.signal) + .then( + () => { + if (!controller.signal.aborted) callbacks.current.recovered(); + }, + () => { + if (!controller.signal.aborted) callbacks.current.failed(); + }, + ) + .finally(() => { + element.preload = preload; + if (pending.current === controller) pending.current = undefined; + }); + return true; + }; + return { + recover, + cancel() { + pending.current?.abort(); + pending.current = undefined; + }, + }; +} diff --git a/src/features/relay/channel-details.ts b/src/features/relay/channel-details.ts index 8c45f1d720..8857de2095 100644 --- a/src/features/relay/channel-details.ts +++ b/src/features/relay/channel-details.ts @@ -1,5 +1,6 @@ import { getEventHash, verifyEvent } from "nostr-tools"; import type { RelayReader } from "./reader"; +import { retryRead } from "./read-retry"; import type { RelayWriter } from "./transport"; import type { RelayEvent } from "./events"; import { PublishRejected } from "./outbox"; @@ -140,7 +141,10 @@ export function createChannelDetails({ return owned( id, async (signal) => { - const { details, metadata } = await read(id, signal); + const { details, metadata } = await retryRead( + () => read(id, signal), + signal, + ); acceptDiscovery([metadata]); signal.throwIfAborted(); return details; diff --git a/src/features/relay/channel-lifecycle.test.ts b/src/features/relay/channel-lifecycle.test.ts index 76ab40c4f3..827d3509d4 100644 --- a/src/features/relay/channel-lifecycle.test.ts +++ b/src/features/relay/channel-lifecycle.test.ts @@ -3,7 +3,16 @@ import type { RelaySession } from "./session"; import { createHash } from "node:crypto"; import { schnorr } from "@noble/curves/secp256k1.js"; import { bytesToHex } from "nostr-tools/utils"; -import { assert, describe, expect, it, vi } from "vitest"; +import { + afterEach, + beforeEach, + assert, + describe, + expect, + it, + vi, +} from "vitest"; +import { ReadError } from "./errors"; import { finalizeEvent, getPublicKey, type EventTemplate } from "nostr-tools"; import { PublishRejected } from "./outbox"; import { @@ -154,6 +163,174 @@ function harness(role = "owner", type = "stream", owners = 1) { }; } +describe("bounded visibility recovery", () => { + beforeEach(() => { + vi.useFakeTimers(); + vi.spyOn(Math, "random").mockReturnValue(0.5); + }); + afterEach(() => { + vi.restoreAllMocks(); + vi.useRealTimers(); + }); + + it("keeps loaded visibility quiet through a transient failure and success", async () => { + const h = harness(); + h.setVisible([ + h.record(30622, [ + ["d", viewer], + ["p", viewer], + ["h", id], + ]), + ]); + await h.owner.capability.refreshVisibility(); + h.read.mockClear(); + h.read.mockRejectedValueOnce(new TypeError("fetch failed")); + const pending = h.owner.capability.refreshVisibility(); + await vi.advanceTimersByTimeAsync(0); + expect(h.owner.capability.snapshot()).toEqual({ + status: "loading", + hidden: [id], + }); + await vi.advanceTimersByTimeAsync(749); + expect(h.read).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(1); + await pending; + expect(h.read).toHaveBeenCalledTimes(2); + expect(h.owner.capability.snapshot()).toEqual({ + status: "ready", + hidden: [id], + }); + h.owner.dispose(); + }); + + it("exhausts four reads, then surfaces an error a manual retry can clear", async () => { + const h = harness(); + const lifecycle = h.owner.capability; + h.read.mockRejectedValue(new ReadError("unavailable", "offline", 503)); + const pending = lifecycle.refreshVisibility(); + await vi.advanceTimersByTimeAsync(0); + for (const [index, delay] of [750, 1500, 3000].entries()) { + await vi.advanceTimersByTimeAsync(delay - 1); + expect(h.read).toHaveBeenCalledTimes(index + 1); + expect(lifecycle.snapshot().status).toBe("loading"); + await vi.advanceTimersByTimeAsync(1); + } + await pending; + expect(lifecycle.snapshot()).toEqual({ + status: "error", + hidden: [], + error: "offline", + }); + await vi.advanceTimersByTimeAsync(120_000); + expect(h.read).toHaveBeenCalledTimes(4); + h.read.mockResolvedValue([]); + await lifecycle.refreshVisibility(); + expect(lifecycle.snapshot().status).toBe("ready"); + h.owner.dispose(); + }); + + it.each([ + new ReadError("denied", "denied", 403), + new Error("Invalid DM visibility snapshot"), + ])("surfaces permanent failure immediately: %s", async (error) => { + const h = harness(); + h.read.mockRejectedValue(error); + await h.owner.capability.refreshVisibility(); + expect(h.owner.capability.snapshot()).toMatchObject({ + status: "error", + error: error.message, + }); + await vi.advanceTimersByTimeAsync(120_000); + expect(h.read).toHaveBeenCalledOnce(); + h.owner.dispose(); + }); + + it("coalesces concurrent triggers during reads and backoff", async () => { + const h = harness(); + const gate = deferred(); + h.read.mockImplementationOnce(() => gate.promise); + const first = h.owner.capability.refreshVisibility(); + const concurrent = h.owner.capability.refreshVisibility(); + expect(concurrent).toBe(first); + await vi.advanceTimersByTimeAsync(0); + expect(h.read).toHaveBeenCalledOnce(); + gate.reject(new TypeError("offline")); + await vi.advanceTimersByTimeAsync(0); + expect(h.owner.capability.refreshVisibility()).toBe(first); + await vi.advanceTimersByTimeAsync(750); + await first; + expect(h.read).toHaveBeenCalledTimes(2); + h.owner.dispose(); + }); + + it.each(["caller", "cancel", "clear", "dispose"] as const)( + "%s cancels backoff without a terminal error or another read", + async (action) => { + const h = harness(); + const caller = new AbortController(); + h.read.mockRejectedValue(new TypeError("offline")); + const pending = h.owner.capability.refreshVisibility(caller.signal); + await vi.advanceTimersByTimeAsync(0); + if (action === "caller") caller.abort(); + else h.owner[action](); + await pending; + await vi.advanceTimersByTimeAsync(120_000); + expect(h.read).toHaveBeenCalledOnce(); + expect(h.owner.capability.snapshot()).not.toHaveProperty("error"); + h.owner.dispose(); + }, + ); + + it("aborts an obsolete read and rejects its late response after a superseding request", async () => { + const h = harness(); + const gate = deferred(); + const caller = new AbortController(); + h.read.mockImplementationOnce(() => gate.promise); + const old = h.owner.capability.refreshVisibility(caller.signal); + await vi.advanceTimersByTimeAsync(0); + const signal = h.read.mock.calls[0]?.[1]?.signal; + caller.abort(); + expect(signal?.aborted).toBe(true); + await h.owner.capability.refreshVisibility(); + gate.resolve([ + h.record(30622, [ + ["d", viewer], + ["p", viewer], + ["h", id], + ]), + ]); + await old; + expect(h.owner.capability.snapshot()).toEqual({ + status: "ready", + hidden: [], + }); + h.owner.dispose(); + }); + + it("retries a visibility read invalidated by an unrelated channel denial", async () => { + const h = harness(); + h.read.mockImplementationOnce( + (_filters, options) => + new Promise((_resolve, reject) => { + options?.signal?.addEventListener( + "abort", + () => reject(cancelled()), + { once: true }, + ); + }), + ); + const pending = h.owner.capability.refreshVisibility(); + await vi.advanceTimersByTimeAsync(0); + expect(h.read).toHaveBeenCalledOnce(); + h.owner.cancel([id]); + await vi.advanceTimersByTimeAsync(750); + await pending; + expect(h.read).toHaveBeenCalledTimes(2); + expect(h.owner.capability.snapshot().status).toBe("ready"); + h.owner.dispose(); + }); +}); + it.each(["archive", "delete", "leave", "hide"] as const)( "confirms %s while ordinary reads still see pre-write state", async (action) => { diff --git a/src/features/relay/channel-lifecycle.ts b/src/features/relay/channel-lifecycle.ts index 3dde527bdd..c97bbfab35 100644 --- a/src/features/relay/channel-lifecycle.ts +++ b/src/features/relay/channel-lifecycle.ts @@ -4,6 +4,7 @@ import type { RelayReader } from "./reader"; import type { RelayWriter } from "./transport"; import { PublishRejected } from "./outbox"; import { isTransientFailure, readErrorKind } from "./errors"; +import { retryRead } from "./read-retry"; import { hasTag, newer, type RelayEvent } from "./events"; import { openMetadata } from "./discovery"; import { @@ -31,8 +32,6 @@ export type DmVisibility = Readonly<{ status: "idle" | "loading" | "ready" | "error"; hidden: readonly string[]; error?: string; - /** Set with an error that a later retry can fix; see isTransientFailure. */ - transient?: true; }>; export interface ChannelLifecycleCapability { readonly available: boolean; @@ -47,7 +46,7 @@ export interface ChannelLifecycleCapability { ): Promise; snapshot(): DmVisibility; subscribe(listener: () => void): () => void; - refreshVisibility(): Promise; + refreshVisibility(signal?: AbortSignal): Promise; observeRemoval(listener: (channelId: string) => void): () => void; } @@ -81,6 +80,8 @@ export function createChannelLifecycle({ }); let visibility: RelayEvent | undefined; let refresh: Promise | undefined; + let refreshSignal: AbortSignal | undefined; + let refreshController: AbortController | undefined; const available = !!reader && !!writer; const emit = (next: DmVisibility) => { snapshot = Object.freeze(next); @@ -182,9 +183,17 @@ export function createChannelLifecycle({ try { for (let start = 0; start < owners.length; start += 4) { const batch = owners.slice(start, start + 4); - const profiles = await reader.read( - batch.map((pubkey) => ({ kinds: [0], authors: [pubkey], limit: 1 })), - { signal: profileSignal, fresh: true, priority: "foreground" }, + const profiles = await retryRead( + (signal) => + reader.read( + batch.map((pubkey) => ({ + kinds: [0], + authors: [pubkey], + limit: 1, + })), + { signal, fresh: true, priority: "foreground" }, + ), + profileSignal, ); profileSignal.throwIfAborted(); assertAccess(id); @@ -276,47 +285,70 @@ export function createChannelLifecycle({ listeners.delete(listener); }; }, - refreshVisibility() { - if (refresh) return refresh; - if (!reader || closed) return Promise.resolve(); + refreshVisibility(caller?: AbortSignal) { + if (!reader || closed || caller?.aborted) return Promise.resolve(); + if (refresh && !refreshSignal?.aborted) return refresh; const started = epoch; + const controller = new AbortController(); + refreshController = controller; + const signal = AbortSignal.any([ + controller.signal, + ...(caller ? [caller] : []), + ]); + refreshSignal = signal; emit({ status: "loading", hidden: snapshot.hidden }); - const operation = owned({}, async (signal) => { - await readVisibility(signal); - }) + // Exhausted errors omit `transient`, so session recovery cannot restart + // them. A read retired by another channel's access change (`cancelled`) + // retries within the same budget. + const operation = retryRead( + async () => { + await owned({ caller: signal }, readVisibility); + }, + signal, + isTransientFailure, + ) .catch((error: unknown) => { - if (started === epoch && !closed) + if (started === epoch && !closed && refresh === operation) emit({ - status: "error", + status: signal.aborted + ? visibility + ? "ready" + : "idle" + : "error", hidden: snapshot.hidden, - error: error instanceof Error ? error.message : String(error), - ...(isTransientFailure(error) - ? { transient: true as const } + ...(!signal.aborted + ? { + error: + error instanceof Error ? error.message : String(error), + } : {}), }); }) .finally(() => { - if (refresh === operation) refresh = undefined; + if (refresh === operation) { + refresh = undefined; + refreshController = undefined; + } }); refresh = operation; return operation; }, async load(value: string, caller?: AbortSignal) { const id = lifecycleChannelId(value); - return owned({ id, caller }, async (signal) => { - for (;;) { - try { - return await load(id, signal); - } catch (error) { - // Owner cancel/clear/dispose, the caller and the deadline all - // arrive through `signal`. Any other cancellation is the session - // retiring shared reads for another channel's access change, - // which is no evidence about this channel: read again. - if (signal.aborted || readErrorKind(error) !== "cancelled") - throw error; + return owned({ id, caller }, (signal) => + retryRead(async () => { + for (;;) { + try { + return await load(id, signal); + } catch (error) { + // Another channel's revocation retires shared reads. It is not + // a failed attempt for this channel; reread under the same deadline. + if (signal.aborted || readErrorKind(error) !== "cancelled") + throw error; + } } - } - }); + }, signal), + ); }, async run( action: ChannelLifecycleCommand, @@ -472,11 +504,15 @@ export function createChannelLifecycle({ capability, /** Abort work on `ids` and id-less visibility reads; without `ids`, everything. */ cancel(ids?: readonly string[]) { + // A channel-scoped invalidation retires a read, not the viewer's whole + // refresh. Its cancellation can retry within the existing budget. + if (!ids) refreshController?.abort(); for (const [controller, id] of controllers) if (!ids || id === undefined || ids.includes(id)) controller.abort(); }, clear() { epoch++; + refreshController?.abort(); for (const controller of controllers.keys()) controller.abort(); refresh = undefined; visibility = undefined; @@ -485,6 +521,7 @@ export function createChannelLifecycle({ dispose() { closed = true; epoch++; + refreshController?.abort(); for (const controller of controllers.keys()) controller.abort(); listeners.clear(); removalListeners.clear(); diff --git a/src/features/relay/emoji-directory.ts b/src/features/relay/emoji-directory.ts index e8e761d00a..cf479a112a 100644 --- a/src/features/relay/emoji-directory.ts +++ b/src/features/relay/emoji-directory.ts @@ -1,3 +1,4 @@ +import { retryRead } from "./read-retry"; import type { UploadedAttachment } from "./attachments"; import { byteSize } from "./budget"; import { @@ -124,9 +125,13 @@ export function createEmojiDirectory( publish({ status: "loading", error: undefined }); pending = (async () => { try { - const events = await reader.read( - [{ kinds: [30030], "#d": [EMOJI_SET], limit: LIMIT }], - { signal: owned.signal, priority: "background" }, + const events = await retryRead( + (signal) => + reader.read([{ kinds: [30030], "#d": [EMOJI_SET], limit: LIMIT }], { + signal, + priority: "background", + }), + owned.signal, ); if (closed || owned.signal.aborted) return; accept(events); diff --git a/src/features/relay/errors.ts b/src/features/relay/errors.ts index 166d31466f..f843e3dd45 100644 --- a/src/features/relay/errors.ts +++ b/src/features/relay/errors.ts @@ -15,6 +15,10 @@ export class ReadError extends Error { this.name = "ReadError"; } } +/** HTTP statuses where the same request can succeed later unchanged. */ +export function transientStatus(status: number) { + return status === 408 || status === 429 || status >= 500; +} export function httpReadError(status: number) { return new ReadError( status === 401 || status === 403 ? "denied" : "unavailable", diff --git a/src/features/relay/gifs.ts b/src/features/relay/gifs.ts index d622ea0b24..4f7371a367 100644 --- a/src/features/relay/gifs.ts +++ b/src/features/relay/gifs.ts @@ -1,3 +1,5 @@ +import { retryRead } from "./read-retry"; +import { ReadError, transientStatus } from "./errors"; import { nativeIdentityEnabled } from "../identity/service"; import { relayKlipySearchPath, @@ -148,8 +150,12 @@ async function brokerRequest( ...(signal ? { signal } : {}), }); const result = (await response.json()) as T & { error?: string }; - if (!response.ok) - throw new Error(result.error ?? `GIF request failed (${response.status})`); + if (!response.ok) { + const message = result.error ?? `GIF request failed (${response.status})`; + throw transientStatus(response.status) + ? new ReadError("unavailable", message, response.status) + : new Error(message); + } return result; } @@ -204,15 +210,19 @@ export async function fetchKlipyGifs( if (nativeIdentityEnabled() && route !== "/gifs/search") throw new Error("GIF search is unavailable"); try { - const response = await brokerRequest( - community, - route, - { - customer_id: customerId(), - locale: navigator.language || "en-US", - query: query.trim(), - }, - signal, + const response = await retryRead( + (signal) => + brokerRequest( + community, + route, + { + customer_id: customerId(), + locale: navigator.language || "en-US", + query: query.trim(), + }, + signal, + ), + signal ?? new AbortController().signal, ); if (response.result === false) throw new Error("GIF search failed"); return normalizeKlipyGifs(response.data?.data ?? []); diff --git a/src/features/relay/global-search.test.ts b/src/features/relay/global-search.test.ts index dace5535f3..d373f48da3 100644 --- a/src/features/relay/global-search.test.ts +++ b/src/features/relay/global-search.test.ts @@ -297,6 +297,8 @@ it("rejects missing/forged/private evidence and purges prior public content on a }); it("surfaces bounded metadata failures instead of silently returning an empty page", async () => { + vi.useFakeTimers(); + vi.spyOn(Math, "random").mockReturnValue(0.5); const viewer = keypair(), relay = keypair(); const owner = createRelaySession({ @@ -310,11 +312,15 @@ it("surfaces bounded metadata failures instead of silently returning an empty pa }, }); try { - await expect(owner.session.read(search)).rejects.toThrow( + const rejected = expect(owner.session.read(search)).rejects.toThrow( "metadata unavailable", ); + await vi.advanceTimersByTimeAsync(5250); + await rejected; } finally { owner.dispose(); + vi.restoreAllMocks(); + vi.useRealTimers(); } }); diff --git a/src/features/relay/identity-archives.ts b/src/features/relay/identity-archives.ts index 4d6ba3a19c..93b5df0de6 100644 --- a/src/features/relay/identity-archives.ts +++ b/src/features/relay/identity-archives.ts @@ -1,3 +1,4 @@ +import { retryRead } from "./read-retry"; import { getEventHash } from "nostr-tools"; import { authTagOwner, withinTimeBounds } from "../agents/owner-attestation"; import { byteSize } from "./budget"; @@ -79,9 +80,14 @@ export function createIdentityArchives( void (async () => { try { if (closed || owned.signal.aborted) return; - const events = await reader.read( - [{ kinds: [13535], authors: [authority], limit: 1 }], - { signal: owned.signal, priority: "background", fresh: true }, + const events = await retryRead( + (signal) => + reader.read([{ kinds: [13535], authors: [authority], limit: 1 }], { + signal, + priority: "background", + fresh: true, + }), + owned.signal, ); if (closed || owned.signal.aborted) return; // The relay publishes its first snapshot with its first archive, so diff --git a/src/features/relay/profile-directory.ts b/src/features/relay/profile-directory.ts index 3f3b136a7e..c06ad26abc 100644 --- a/src/features/relay/profile-directory.ts +++ b/src/features/relay/profile-directory.ts @@ -1,3 +1,4 @@ +import { retryRead } from "./read-retry"; import { attestedOwner } from "../agents/owner-attestation"; import { hasEventProof } from "./events"; import type { Outbox } from "./outbox"; @@ -30,7 +31,12 @@ export function createProfileDirectory( const owners = new WeakMap(); const pending = new Map< string, - { work: Promise; authors: string[]; priority: Priority } + { + work: Promise; + authors: string[]; + priority: Priority; + reading: boolean; + } >(); const listeners = new Set<() => void>(); let snapshot: ReadonlyMap = new Map(); @@ -116,12 +122,13 @@ export function createProfileDirectory( // waiting on the old promise alone would leave it behind background work. if (priority === "foreground" && batch.priority === "background") { batch.priority = priority; - void reader - .read([{ kinds: [0], authors: batch.authors, limit: 500 }], { - signal, - priority, - }) - .catch(() => {}); + if (batch.reading) + void reader + .read([{ kinds: [0], authors: batch.authors, limit: 500 }], { + signal, + priority, + }) + .catch(() => {}); } waiting.add(batch.work); } @@ -129,20 +136,39 @@ export function createProfileDirectory( }); for (let offset = 0; offset < missing.length; offset += 500) { const authors = missing.slice(offset, offset + 500); + let answered = false; const batch = { authors, priority, - work: reader - .read([{ kinds: [0], authors, limit: 500 }], { signal, priority }) - .then((result) => { - if (!signal.aborted) - accept(result.filter((event) => authors.includes(event.pubkey))); - }) - .finally(() => { - for (const id of authors) - if (pending.get(id) === batch) pending.delete(id); - }), + reading: false, + work: Promise.resolve(), }; + batch.work = retryRead(async (signal) => { + // A promotion during backoff applies to the next attempt instead. + batch.reading = true; + try { + const result = await reader.read( + [{ kinds: [0], authors, limit: 500 }], + { signal, priority: batch.priority }, + ); + answered = true; + return result; + } finally { + batch.reading = false; + } + }, signal) + .then((result) => { + if (!signal.aborted) + accept(result.filter((event) => authors.includes(event.pubkey))); + }) + .catch((error: unknown) => { + // The old directory contract silently discards a retired success. + if (!signal.aborted || !answered) throw error; + }) + .finally(() => { + for (const id of authors) + if (pending.get(id) === batch) pending.delete(id); + }); for (const id of authors) pending.set(id, batch); waiting.add(batch.work); } diff --git a/src/features/relay/read-retry.test.ts b/src/features/relay/read-retry.test.ts new file mode 100644 index 0000000000..c45a87fe3c --- /dev/null +++ b/src/features/relay/read-retry.test.ts @@ -0,0 +1,155 @@ +import { afterEach, beforeEach, expect, it, vi } from "vitest"; +import { ReadError } from "./errors"; +import { retryRead } from "./read-retry"; + +beforeEach(() => { + vi.useFakeTimers(); + vi.spyOn(Math, "random").mockReturnValue(0.5); +}); +afterEach(() => { + vi.restoreAllMocks(); + vi.useRealTimers(); +}); + +it("retries transient reads at the backoff boundaries and returns success", async () => { + const read = vi + .fn() + .mockRejectedValueOnce(new TypeError("offline")) + .mockResolvedValue("loaded"); + const pending = retryRead(read, new AbortController().signal); + await vi.advanceTimersByTimeAsync(749); + expect(read).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(1); + expect(await pending).toBe("loaded"); + expect(read).toHaveBeenCalledTimes(2); +}); + +it("rethrows the last failure after one read and three retries", async () => { + const error = new ReadError("unavailable", "offline", 503); + const read = vi.fn().mockRejectedValue(error); + const pending = retryRead(read, new AbortController().signal); + const rejected = expect(pending).rejects.toBe(error); + for (const [index, delay] of [750, 1500, 3000].entries()) { + await vi.advanceTimersByTimeAsync(delay - 1); + expect(read).toHaveBeenCalledTimes(index + 1); + await vi.advanceTimersByTimeAsync(1); + } + await rejected; + await vi.advanceTimersByTimeAsync(120_000); + expect(read).toHaveBeenCalledTimes(4); +}); + +it.each([400, 403, 404])( + "retries transport-classified edge HTTP %s pages", + async (status) => { + const read = vi + .fn() + .mockRejectedValueOnce(new ReadError("unavailable", "edge", status)) + .mockResolvedValue("loaded"); + const pending = retryRead(read, new AbortController().signal); + await vi.advanceTimersByTimeAsync(750); + expect(await pending).toBe("loaded"); + expect(read).toHaveBeenCalledTimes(2); + }, +); + +it("does not start a retry after the 10-second quiet window", async () => { + const error = new ReadError("unavailable", "Relay read timed out"); + const read = vi.fn( + () => + new Promise((_, reject) => setTimeout(() => reject(error), 4000)), + ); + const pending = retryRead(read, new AbortController().signal); + const rejected = expect(pending).rejects.toBe(error); + await vi.advanceTimersByTimeAsync(9999); + expect(read).toHaveBeenCalledTimes(2); + await vi.advanceTimersByTimeAsync(120_000); + await rejected; + expect(read).toHaveBeenCalledTimes(2); +}); + +it.each([ + new ReadError("denied", "denied", 403), + new ReadError("invalid-response", "invalid", 400), + new Error("Invalid input"), +])("surfaces permanent failures immediately: %s", async (error) => { + const read = vi.fn().mockRejectedValue(error); + await expect(retryRead(read, new AbortController().signal)).rejects.toBe( + error, + ); + await vi.advanceTimersByTimeAsync(120_000); + expect(read).toHaveBeenCalledOnce(); +}); + +it.each([5000, 60000, 86401000])( + "surfaces long admission delays immediately: %s", + async (delay) => { + const error = new ReadError("unavailable", "quota", 429, delay); + const read = vi.fn().mockRejectedValue(error); + await expect(retryRead(read, new AbortController().signal)).rejects.toBe( + error, + ); + expect(read).toHaveBeenCalledOnce(); + expect(vi.getTimerCount()).toBe(0); + }, +); + +it("honors admission hints within the bounded backoff", async () => { + const read = vi + .fn() + .mockRejectedValueOnce(new ReadError("unavailable", "quota", 429, 900)) + .mockResolvedValue([]); + const pending = retryRead(read, new AbortController().signal); + await vi.advanceTimersByTimeAsync(899); + expect(read).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(1); + await pending; + expect(read).toHaveBeenCalledTimes(2); +}); + +it.each([ + new DOMException("retired", "AbortError"), + new ReadError("cancelled", "cancelled"), + new Error("context", { cause: new DOMException("retired", "AbortError") }), +])("does not retry owner cancellations: %s", async (error) => { + const read = vi.fn().mockRejectedValue(error); + await expect(retryRead(read, new AbortController().signal)).rejects.toBe( + error, + ); + expect(read).toHaveBeenCalledOnce(); + expect(vi.getTimerCount()).toBe(0); +}); + +it("aborts backoff and releases the timer", async () => { + const controller = new AbortController(); + const read = vi.fn().mockRejectedValue(new TypeError("offline")); + const pending = retryRead(read, controller.signal); + const rejected = expect(pending).rejects.toMatchObject({ + name: "AbortError", + }); + await vi.advanceTimersByTimeAsync(0); + controller.abort(); + await rejected; + await vi.advanceTimersByTimeAsync(120_000); + expect(read).toHaveBeenCalledOnce(); + expect(vi.getTimerCount()).toBe(0); +}); + +it("rejects late success from a read that ignores cancellation", async () => { + const controller = new AbortController(); + let resolve!: (value: string) => void; + const read = vi.fn( + () => + new Promise((done) => { + resolve = done; + }), + ); + const pending = retryRead(read, controller.signal); + const rejected = expect(pending).rejects.toMatchObject({ + name: "AbortError", + }); + controller.abort(); + resolve("obsolete"); + await rejected; + expect(read).toHaveBeenCalledOnce(); +}); diff --git a/src/features/relay/read-retry.ts b/src/features/relay/read-retry.ts new file mode 100644 index 0000000000..debc1ef8aa --- /dev/null +++ b/src/features/relay/read-retry.ts @@ -0,0 +1,57 @@ +import { isTransientFailure, readErrorKind, ReadError } from "./errors.ts"; +import { RECOVERY_DELAYS_MS } from "./recovery.ts"; + +const QUIET_MS = 10_000; + +/** Transient, but not the owner's own cancellation. */ +export function isRetryableRead(error: unknown): boolean { + if (readErrorKind(error) === "cancelled") return false; + if (error instanceof Error && error.cause !== undefined) + return isRetryableRead(error.cause); + return isTransientFailure(error); +} + +/** One initial read and up to three quiet retries under session recovery's + * first three ceilings (1/2/4 s) with equal jitter, none starting after 10 s. + * Exhaustion rethrows the last failure. Use one owner per read path; the + * caller owns its lifetime. */ +export async function retryRead( + read: (signal: AbortSignal) => Promise, + signal: AbortSignal, + retryable: (error: unknown) => boolean = isRetryableRead, +): Promise { + const began = performance.now(); + for (let attempt = 0; ; attempt++) { + signal.throwIfAborted(); + try { + const value = await read(signal); + signal.throwIfAborted(); + return value; + } catch (error) { + signal.throwIfAborted(); + const ceiling = RECOVERY_DELAYS_MS[attempt] ?? 0; + const hint = error instanceof ReadError ? (error.retryAfterMs ?? 0) : 0; + const delay = Math.max(ceiling * (0.5 + Math.random() / 2), hint); + // A longer admission pause keeps its existing cooldown and action; a + // read that already waited out its own deadline surfaces it. + if ( + attempt === 3 || + !retryable(error) || + hint > ceiling || + performance.now() - began + delay > QUIET_MS + ) + throw error; + await new Promise((resolve, reject) => { + const abort = () => { + clearTimeout(timer); + reject(signal.reason); + }; + const timer = setTimeout(() => { + signal.removeEventListener("abort", abort); + resolve(); + }, delay); + signal.addEventListener("abort", abort, { once: true }); + }); + } + } +} diff --git a/src/features/relay/reminders.ts b/src/features/relay/reminders.ts index ecd2688100..82b101bcb9 100644 --- a/src/features/relay/reminders.ts +++ b/src/features/relay/reminders.ts @@ -1,5 +1,6 @@ // NIP-ER reminders (kind 30300): the viewer's self-encrypted, replaceable reminder list. // Encryption and signing stay in the host; this owner sees only decoded intent. +import { retryRead } from "./read-retry"; import type { ReadFilter, RelayEvent } from "./events"; export const REMINDER_KIND = 30300; @@ -150,6 +151,7 @@ export function createReminders(options: { // started after (re-)establishment can. Offline reads still load the list. let offline = true; let generation = 0; + let refreshController: AbortController | undefined; let loadingGeneration = -1; const publishState = ( next: Omit, @@ -241,14 +243,21 @@ export function createReminders(options: { refresh() { if (loading && loadingGeneration === generation) return loading; const current = generation; + const controller = new AbortController(); + refreshController = controller; + const readSignal = AbortSignal.any([signal, controller.signal]); loadingGeneration = current; if (state.status !== "loading") publishState({ status: "loading" }); let read: Promise | undefined; read = (async () => { try { - const events = await options.query( - [{ kinds: [REMINDER_KIND], authors: [viewer], limit: 200 }], - signal, + const events = await retryRead( + (signal) => + options.query( + [{ kinds: [REMINDER_KIND], authors: [viewer], limit: 200 }], + signal, + ), + readSignal, ); // A disposed session's late reply, even an empty one, proves nothing. if (signal.aborted) return; @@ -265,6 +274,7 @@ export function createReminders(options: { }); } finally { if (loading === read) loading = undefined; + if (refreshController === controller) refreshController = undefined; } })(); loading = read; @@ -308,6 +318,7 @@ export function createReminders(options: { recover() { offline = false; generation++; + refreshController?.abort(); return capability.refresh(); }, /** Live delivery from another device; failures leave the last good list. */ diff --git a/src/features/relay/session-read-retry.test.ts b/src/features/relay/session-read-retry.test.ts new file mode 100644 index 0000000000..e5ea134f1f --- /dev/null +++ b/src/features/relay/session-read-retry.test.ts @@ -0,0 +1,122 @@ +import { afterEach, beforeEach, expect, it, vi } from "vitest"; +import { createRelaySession } from "./session"; +import { ReadError } from "./errors"; +import { keypair, profile } from "./testing"; +import type { ReadFilter, RelayEvent } from "./events"; + +const owners: ReturnType[] = []; +beforeEach(() => { + vi.useFakeTimers(); + vi.spyOn(Math, "random").mockReturnValue(0.5); +}); +afterEach(() => { + for (const owner of owners.splice(0)) owner.dispose(); + vi.restoreAllMocks(); + vi.useRealTimers(); +}); +function setup(options?: Parameters[1]) { + const viewer = keypair(), + relay = keypair(); + const query = vi.fn( + async (_filters: readonly ReadFilter[], _signal?: AbortSignal) => + [] as RelayEvent[], + ); + const owner = createRelaySession( + { + viewer: viewer.pubkey, + relayAuthor: relay.pubkey, + query, + media: () => undefined, + subscribe: () => ({ update() {}, retry() {}, dispose() {} }), + }, + options, + ); + owners.push(owner); + const filter = [{ kinds: [0], authors: [viewer.pubkey], limit: 1 }]; + return { owner, query, viewer, filter }; +} + +it("retries a finite public read without changing transport edge classification", async () => { + const h = setup(); + h.query.mockRejectedValueOnce(new ReadError("unavailable", "edge", 403)); + const pending = h.owner.session.read(h.filter); + await vi.advanceTimersByTimeAsync(749); + expect(h.query).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(1); + await expect(pending).resolves.toEqual([]); + expect(h.query).toHaveBeenCalledTimes(2); +}); + +it("retains observed data and coalesces refreshes during finite recovery", async () => { + const h = setup(); + const row = profile(h.viewer, { name: "Kept" }); + h.query.mockResolvedValueOnce([row]); + const view = h.owner.session.observe(h.filter); + await view.refresh(); + h.query.mockRejectedValueOnce(new TypeError("offline")); + const pending = view.refresh(); + await vi.advanceTimersByTimeAsync(0); + await view.refresh(); + expect(h.query).toHaveBeenCalledTimes(2); + expect(view.snapshot().status).toBe("loading"); + expect(view.snapshot().events.map((event) => event.id)).toContain(row.id); + await vi.advanceTimersByTimeAsync(750); + await pending; + expect(view.snapshot().status).toBe("ready"); + expect(h.query).toHaveBeenCalledTimes(3); + view.dispose(); +}); + +it("leaves roster errors transient for their existing recovery owner", async () => { + const h = setup(); + h.query.mockRejectedValue(new ReadError("unavailable", "offline", 503)); + h.owner.session.channels.refreshList?.(); + await vi.advanceTimersByTimeAsync(0); + expect(h.owner.session.live.snapshot().roster).toMatchObject({ + state: "error", + transient: true, + }); + expect(h.query).toHaveBeenCalledOnce(); +}); + +it.each(["caller", "clearCache", "dispose"] as const)( + "%s cancels a finite public read's backoff", + async (action) => { + const h = setup(); + const caller = new AbortController(); + h.query.mockRejectedValue(new TypeError("offline")); + const pending = h.owner.session.read(h.filter, { signal: caller.signal }); + const rejected = expect(pending).rejects.toMatchObject({ + name: "AbortError", + }); + await vi.advanceTimersByTimeAsync(0); + const before = h.query.mock.calls.filter(([filters]) => + filters[0]?.kinds?.includes(0), + ).length; + if (action === "caller") caller.abort(); + else if (action === "clearCache") await h.owner.clearCache(); + else h.owner.dispose(); + await rejected; + await vi.advanceTimersByTimeAsync(7000); + expect( + h.query.mock.calls.filter(([filters]) => filters[0]?.kinds?.includes(0)), + ).toHaveLength(before); + }, +); + +it("surfaces a cached-only inbox failure without retrying it", async () => { + const h = setup({ cachedOnly: true }); + await h.owner.session.inboxFeed.ensure(); + expect(h.owner.session.inboxFeed.snapshot().status).toBe("error"); + expect(vi.getTimerCount()).toBe(0); + expect(h.query).not.toHaveBeenCalled(); +}); + +it("retries a transient inbox read through the session", async () => { + const h = setup(); + h.query.mockRejectedValueOnce(new TypeError("offline")); + const pending = h.owner.session.inboxFeed.ensure(); + await vi.advanceTimersByTimeAsync(750); + await pending; + expect(h.owner.session.inboxFeed.snapshot().status).toBe("ready"); +}); diff --git a/src/features/relay/session.ts b/src/features/relay/session.ts index 167b1c9ea8..4d79019372 100644 --- a/src/features/relay/session.ts +++ b/src/features/relay/session.ts @@ -34,7 +34,10 @@ import { import { createAgentActivity } from "../agents/activity"; import { OBSERVER_KIND } from "../agents/observer"; import { createDirectMessages } from "./direct-messages"; -import { createProfileSearch } from "./profile-search"; +import { + type ProfileSearchOptions, + createProfileSearch, +} from "./profile-search"; import { createChannelWrites } from "./channel-writes"; import { inventoryReader } from "../agents/inventory"; import { @@ -101,6 +104,7 @@ import { import { relayPartition, transportPartition } from "./partition"; import { browserThreadFollows } from "./thread-follows"; import { createMessages } from "./messages"; +import { retryRead } from "./read-retry"; import { createThreadView } from "./threads"; import { ByteLru } from "./budget"; import { createRelayProfiler } from "./profiling"; @@ -155,6 +159,51 @@ export function createRelaySession( const profiling = options.profiling ?? transport?.profiling ?? createRelayProfiler(); const requests = createRelayReader(transport, { profiling }); + // Finite public consumers have no recovery source. Internal roster, history, + // catch-up and preference reads keep their existing coordinator and cooldowns. + // The request scheduler retires in-flight reads on access and cache changes; + // `retireBackoffs` applies the same rule to reads waiting out a backoff. + const backoffs = new Map< + AbortController, + readonly ReadFilter[] | undefined + >(); + function finiteRead( + filters: readonly ReadFilter[] | undefined, + settings: ReadOptions | undefined, + work: (signal: AbortSignal) => Promise, + ): Promise { + const controller = new AbortController(); + const signal = AbortSignal.any([ + lifetime.signal, + controller.signal, + ...(settings?.signal ? [settings.signal] : []), + ]); + // Offline and cached-membership reads cannot recover by retrying. + if ( + !transport || + options.cachedOnly || + filters?.some((filter) => + filter["#h"]?.some((id) => channels.queries.get?.(id)?.cached), + ) + ) + return work(signal); + return retryRead(async (signal) => { + backoffs.delete(controller); + try { + return await work(signal); + } catch (error) { + backoffs.set(controller, filters); + throw error; + } + }, signal).finally(() => backoffs.delete(controller)); + } + /** Without `retired`, every backoff ends; filterless reads ignore `retired`. */ + function retireBackoffs( + retired?: (filters: readonly ReadFilter[]) => boolean, + ) { + for (const [controller, filters] of backoffs) + if (!retired || (filters && retired(filters))) controller.abort(); + } let revision = 0; let accessEpoch = 0; let cacheClearEpoch = 0; @@ -313,13 +362,13 @@ export function createRelaySession( typing.clear(); // Saved owner inventory does not depend on channel access. Preserve only // that narrow read; broad, mixed, ID and channel reads must still retire. - requests.invalidate( - (filters) => - !filters.length || - filters.some( - (filter) => !isOwnerInventoryFilter(filter, transport?.viewer), - ), - ); + const retired = (filters: readonly ReadFilter[]) => + !filters.length || + filters.some( + (filter) => !isOwnerInventoryFilter(filter, transport?.viewer), + ); + requests.invalidate(retired); + retireBackoffs(retired); const visible = visibility(); const revoked = recent .entries() @@ -385,6 +434,8 @@ export function createRelaySession( ) => void, page?: { withRelayCount: true }, ) { + // A null session can never acquire a transport; retrying it cannot recover. + if (!transport) throw new Error("Relay is disconnected"); if ( options.cachedOnly || filters.some((filter) => @@ -553,7 +604,18 @@ export function createRelaySession( const searchProfiles = createProfileSearch(verified); /** Directory reads bypass shared profile admission. Retry one access * invalidation here, where caller/cache/connection lifetimes are known. */ - async function directoryRead( + function directoryRead( + settings: ReadOptions | undefined, + run: () => Promise, + ): Promise { + const generation = liveGeneration; + return finiteRead(undefined, settings, (signal) => { + if (liveGeneration !== generation) + throw new DOMException("Stale directory read", "AbortError"); + return directoryReadOnce({ ...settings, signal }, run); + }); + } + async function directoryReadOnce( settings: ReadOptions | undefined, run: () => Promise, ): Promise { @@ -804,7 +866,9 @@ export function createRelaySession( const repositories = repoDestinations(async (filters, signal) => { const bound = AbortSignal.any([signal, lifetime.signal]); bound.throwIfAborted(); - const events = await requests.reader.read(filters, { signal: bound }); + const events = await finiteRead(filters, { signal: bound }, (signal) => + requests.reader.read(filters, { signal }), + ); bound.throwIfAborted(); // NIP-34/NIP-MP metadata is global; channel tags are associations, not ACLs. return events; @@ -937,24 +1001,33 @@ export function createRelaySession( // Fail closed at raw/admitted admission; do not relax reference visibility. reader: { read: (filters, settings) => - readVerified(filters, settings, true, (visible, raw) => { - const admitted = new Set(visible.map((event) => event.id)); - if ( - raw.some( - (event) => - [40003, 5, 9005].includes(event.kind) && - !admitted.has(event.id), - ) - ) - throw new Error( - "Inbox message updates could not be verified for current access. Retry inbox.", - ); - }), + finiteRead(filters, settings, (signal) => + readVerified( + filters, + { ...settings, signal }, + true, + (visible, raw) => { + const admitted = new Set(visible.map((event) => event.id)); + if ( + raw.some( + (event) => + [40003, 5, 9005].includes(event.kind) && + !admitted.has(event.id), + ) + ) + throw new Error( + "Inbox message updates could not be verified for current access. Retry inbox.", + ); + }, + ), + ), }, retainedEvent: unread.event, retainedEditIds: unread.retainedEditIds, addressedRead: (filter, signal, prepare) => - readVerified([filter], { signal }, true, prepare), + finiteRead([filter], { signal }, (signal) => + readVerified([filter], { signal }, true, prepare), + ), channels: channels.queries, viewer: transport?.viewer ?? "", notify, @@ -1787,6 +1860,8 @@ export function createRelaySession( () => loadState(sidebarPreferences.queries.snapshot()), () => void sidebarPreferences.queries.refresh(), ), + // Visibility owns its quiet retries and exhausts without `transient`, + // so this source only reports `busy`; its retry is not reached. recoverable( lifecycle.capability.subscribe, () => loadState(lifecycle.capability.snapshot()), @@ -2146,6 +2221,7 @@ export function createRelaySession( local: localViews, clock, canAccess: () => !closed && canAccess(channelId), + canRefresh: () => !closed && canReadRemote(channelId), visible: (events) => events.filter(visibility(events)), notify, }); @@ -2198,7 +2274,10 @@ export function createRelaySession( /** A plugin may request writes from this same interface when the host supports them. */ outbox: writes?.outbox, /** People by profile name only; see `createProfileSearch`. */ - searchProfiles, + searchProfiles: (query: string, settings: ProfileSearchOptions) => + finiteRead(undefined, settings, (signal) => + searchProfiles(query, { ...settings, signal }), + ), /** A fresh verified remote page under the normal access guards, without * merging live observations or local outbox intent into paging evidence. */ async readRemote(filters: readonly ReadFilter[], settings?: ReadOptions) { @@ -2207,11 +2286,11 @@ export function createRelaySession( ? AbortSignal.any([settings.signal, lifetime.signal]) : lifetime.signal; signal.throwIfAborted(); - const result = await readVerified(filters, { - ...settings, - signal, - fresh: true, - }); + const result = await finiteRead( + filters, + { ...settings, signal }, + (signal) => readVerified(filters, { ...settings, signal, fresh: true }), + ); signal.throwIfAborted(); // Applying discovery can revoke access itself. As with read(), let // authority-only reads finish; mixed/content reads must remain fenced. @@ -2236,12 +2315,13 @@ export function createRelaySession( ): Promise<{ events: readonly VisibleEvent[]; relayCount: number }> { const began = revision; const epoch = accessEpoch; - const { events: result, relayCount } = await readVerified( + const { events: result, relayCount } = await finiteRead( filters, settings, - true, - undefined, - { withRelayCount: true }, + (signal) => + readVerified(filters, { ...settings, signal }, true, undefined, { + withRelayCount: true, + }), ); // Applying discovery can itself revoke access. Let that authority-only // read finish under current visibility; content/search completions must @@ -2458,9 +2538,11 @@ export function createRelaySession( controller = owned; update({ status: "loading", error: undefined }); try { - const events = await verified.read(filters, { - signal: owned.signal, - }); + const events = await finiteRead( + filters, + { signal: owned.signal }, + (signal) => verified.read(filters, { signal }), + ); if (disposed || owned.signal.aborted) return; remote = retainEvents([...events, ...remote]); update({ status: "ready" }); @@ -3097,6 +3179,7 @@ export function createRelaySession( unread.clear(); inboxFeed.clear(); requests.invalidate(); + retireBackoffs(); repositories.clearHomes(); profiles.clear(); emoji.clear(); diff --git a/src/features/relay/store.ts b/src/features/relay/store.ts index 8b7e7c8fba..8f3b831721 100644 --- a/src/features/relay/store.ts +++ b/src/features/relay/store.ts @@ -1,3 +1,4 @@ +import { retryRead } from "./read-retry"; import { channelRowKind, rowProfileIds, messagePreview } from "./membership"; import type { Outbox } from "./outbox"; import { MessageProjection } from "./message-projection"; @@ -576,10 +577,17 @@ export function createChannelStore( settings: ReadOptions, ) { if (!transport) throw new Error("Relay is unavailable"); - const events = await transport.read( - [windowFilter(channelId, cursor)], - settings, - ); + const source = transport; + const read = (signal?: AbortSignal) => + source.read([windowFilter(channelId, cursor)], { + ...settings, + ...(signal ? { signal } : {}), + }); + // Head reads belong to session recovery; only manual pagination needs a + // finite retry owner here. + const events = cursor + ? await retryRead(read, settings.signal ?? new AbortController().signal) + : await read(); return { events, page: parseWindow(channelId, cursor, transport.relayAuthor, events), diff --git a/src/features/relay/thread-window.test.ts b/src/features/relay/thread-window.test.ts index 2b12e3b911..5afcaeb0c0 100644 --- a/src/features/relay/thread-window.test.ts +++ b/src/features/relay/thread-window.test.ts @@ -377,7 +377,10 @@ it.each([401, 403, 429, 500])( ? Response.json({ error: "fixture failure" }, { status }) : original(url, options); }); - await h.view.refresh(); + vi.useFakeTimers(); + const pending = h.view.refresh(); + await vi.advanceTimersByTimeAsync(7000); + await pending; if (status === 401 || status === 403) { expect(h.view.snapshot()).toMatchObject({ status: "idle", @@ -392,6 +395,7 @@ it.each([401, 403, 429, 500])( await h.view.refresh(); expect(h.view.snapshot().status).toBe("ready"); } + vi.useRealTimers(); }, ); diff --git a/src/features/relay/threads.ts b/src/features/relay/threads.ts index 05b0c7d139..be1193a2fe 100644 --- a/src/features/relay/threads.ts +++ b/src/features/relay/threads.ts @@ -1,3 +1,4 @@ +import { isRetryableRead, retryRead } from "./read-retry"; import { canonicalChannel, MissingThreadBounds, @@ -70,12 +71,13 @@ export function createThreadView({ relayAuthor, signingAuthority, agentOwner, - reader, - targetReader = reader, + reader: sourceReader, + targetReader: sourceTargetReader = sourceReader, admitTarget, seed, local, canAccess, + canRefresh = canAccess, visible, notify, exact = false, @@ -94,6 +96,8 @@ export function createThreadView({ seed: RelayEvent | undefined; local: LocalEvents | undefined; canAccess(): boolean; + /** Cached membership can display saved rows but cannot authorize a network retry. */ + canRefresh?(): boolean; visible(events: readonly RelayEvent[]): readonly RelayEvent[]; notify(listener: () => void): void; exact?: boolean; @@ -104,6 +108,18 @@ export function createThreadView({ | ((events: readonly RelayEvent[]) => readonly RelayEvent[]) | undefined; }) { + // These finite thread reads have no session recovery source. Their existing + // selection/page controllers still own cancellation and stale-result fencing. + const retrying = (source: RelayReader): RelayReader => ({ + read: (filters, settings) => + retryRead( + (signal) => source.read(filters, { ...settings, signal }), + settings?.signal ?? new AbortController().signal, + (error) => canRefresh() && isRetryableRead(error), + ), + }); + const reader = retrying(sourceReader); + const targetReader = retrying(sourceTargetReader); const acceptTarget = admitTarget ?? admit; let disposed = false; let rootId: string | undefined; diff --git a/src/features/repos/git-read-retry.test.ts b/src/features/repos/git-read-retry.test.ts new file mode 100644 index 0000000000..70eec0cec2 --- /dev/null +++ b/src/features/repos/git-read-retry.test.ts @@ -0,0 +1,61 @@ +import { afterEach, beforeEach, expect, it, vi } from "vitest"; +import { entityFailure } from "./destinations"; +import { repoGitHost } from "./git"; + +beforeEach(() => { + vi.useFakeTimers(); + vi.spyOn(Math, "random").mockReturnValue(0.5); +}); +afterEach(() => { + vi.restoreAllMocks(); + vi.useRealTimers(); +}); +const input = { owner: "a".repeat(64), dtag: "repo" }; +const empty = { + head: null, + commits: [], + files: [], + readme: null, + file: null, + diff: null, +}; +it("retries immutable Git reads and still validates their returned snapshot", async () => { + const request = vi + .fn() + .mockResolvedValueOnce(new Response(null, { status: 502 })) + .mockImplementation(async () => Response.json(empty)); + const pending = repoGitHost(request).read( + input, + new AbortController().signal, + ); + await vi.advanceTimersByTimeAsync(749); + expect(request).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(1); + await expect(pending).resolves.toEqual(empty); + expect(request).toHaveBeenCalledTimes(2); + for (const [body] of request.mock.calls) expect(body).toEqual(input); +}); +it("rejects a successful but invalid Git snapshot without retrying", async () => { + const request = vi + .fn() + .mockImplementation(async () => + Response.json({ ...empty, files: [{ path: "x", hash: "bad", size: 1 }] }), + ); + await expect( + repoGitHost(request).read(input, new AbortController().signal), + ).rejects.toThrow("Invalid repository response"); + expect(request).toHaveBeenCalledOnce(); +}); + +it("preserves Git not-found without consuming a retry budget", async () => { + const request = vi + .fn() + .mockImplementation(async () => new Response(null, { status: 404 })); + const error = await repoGitHost(request) + .read(input, new AbortController().signal) + .catch((error: unknown) => error); + expect(entityFailure(error)).toBe("not-found"); + expect(error).toMatchObject({ status: 404 }); + await vi.advanceTimersByTimeAsync(7000); + expect(request).toHaveBeenCalledOnce(); +}); diff --git a/src/features/repos/git.ts b/src/features/repos/git.ts index fa231732fb..d1a18298b3 100644 --- a/src/features/repos/git.ts +++ b/src/features/repos/git.ts @@ -1,4 +1,5 @@ -import { ReadError } from "../relay/errors.ts"; +import { ReadError, transientStatus } from "../relay/errors.ts"; +import { isRetryableRead, retryRead } from "../relay/read-retry.ts"; import { entityDtag, entityHex, gitHash } from "./routes.ts"; export type GitRead = { @@ -114,16 +115,27 @@ export function repoGitHost( return { async read(input, signal) { const body = parseGitRead(input); - const response = await request(body, signal); - if (!response.ok) - throw new ReadError( - response.status === 403 || response.status === 401 - ? "denied" - : "unavailable", - "Repository content could not be read", - response.status, - ); - const value = JSON.parse(await gitReadText(response)) as GitSnapshot; + const text = await retryRead( + async (signal) => { + const response = await request(body, signal); + if (!response.ok) + throw new ReadError( + response.status === 403 || response.status === 401 + ? "denied" + : "unavailable", + "Repository content could not be read", + response.status, + ); + return gitReadText(response); + }, + signal, + // A missing repository or invalid request stays actionable. + (error) => + error instanceof ReadError && error.status !== undefined + ? transientStatus(error.status) + : isRetryableRead(error), + ); + const value = JSON.parse(text) as GitSnapshot; signal.throwIfAborted(); if ( !value || diff --git a/tests/browser/hidden-conversation-retry.spec.mjs b/tests/browser/hidden-conversation-retry.spec.mjs new file mode 100644 index 0000000000..fbea78de35 --- /dev/null +++ b/tests/browser/hidden-conversation-retry.spec.mjs @@ -0,0 +1,86 @@ +import { openPage } from "./navigation.mjs"; +import { test, expect } from "./fixture.mjs"; + +test.use({ + productionBroker: true, + channelLifecycle: true, + pluginFixtures: true, + historyCounts: { alpha: 1, beta: 1 }, +}); + +// App/session/broker wiring must expose exhausted visibility reads on the +// sidebar and let the existing action recover. Timing/error matrices stay in +// service and mounted component tests. +test("hidden-conversation exhaustion exposes the existing action and manual retry recovers", async ({ + page, + app, +}) => { + let failing = true; + let reads = 0; + await page.goto(app.origin); + await openPage(page, "Messages"); + // Startup can supersede visibility reads as roster evidence arrives. Count + // one explicit visibility refresh after the live roster and visibility settle. + await expect + .poll(() => + page.evaluate(() => { + const session = window.fixtureRelay.snapshot().session; + return [ + session.live.snapshot().status, + session.channels.list().status, + session.channelLifecycle.snapshot().status, + ]; + }), + ) + .toEqual(["connected", "ready", "ready"]); + await page.route("**/api/relay/**/query", async (route) => { + if ( + !route + .request() + .postDataJSON() + .some((filter) => filter.kinds?.includes(30622)) + ) + return route.continue(); + reads++; + if (!failing) return route.continue(); + app.report.startupFailures ??= []; + app.report.startupFailures.push(route.request().url()); + return route.fulfill({ + status: 502, + json: { error: "Fixture visibility unavailable" }, + }); + }); + await page.evaluate(() => + window.fixtureRelay.snapshot().session.channelLifecycle.refreshVisibility(), + ); + const retry = page.getByRole("button", { + name: "Retry hidden conversations", + exact: true, + }); + await expect(retry).toBeVisible(); + expect(reads).toBe(4); + await expect( + page.getByRole("alert").filter({ + hasText: "Hidden conversations could not be refreshed.", + }), + ).toBeVisible(); + expect( + await page.evaluate( + () => + window.fixtureRelay.snapshot().session.recovery.snapshot().recovering, + ), + ).toBe(false); + failing = false; + await retry.click(); + await expect + .poll(() => + page.evaluate( + () => + window.fixtureRelay.snapshot().session.channelLifecycle.snapshot() + .status, + ), + ) + .toBe("ready"); + await expect(retry).toHaveCount(0); + expect(reads).toBe(5); +}); diff --git a/tests/browser/history-loading.spec.mjs b/tests/browser/history-loading.spec.mjs index 16d2002b77..5d7bb8afa7 100644 --- a/tests/browser/history-loading.spec.mjs +++ b/tests/browser/history-loading.spec.mjs @@ -150,7 +150,8 @@ test.describe("quota retry without threshold-driven continuation", () => { if (filters.some(olderAlpha)) browserCursors.push(filters); } }); - app.relay.quotaNextOlder("alpha", 0); + // A hint past the first backoff ceiling surfaces instead of retrying quietly. + app.relay.quotaNextOlder("alpha"); const cursors = () => app.report.queries.filter(({ filter }) => olderAlpha(filter)); await history(page).hover(); @@ -594,7 +595,7 @@ test.describe("pages inside one second", () => { app, }) => { await open(page, app); - app.relay.quotaNextOlder("alpha", 0); + app.relay.quotaNextOlder("alpha"); await history(page).hover(); await page.mouse.wheel(0, -100000); await expect(history(page).getByRole("alert")).toContainText(