diff --git a/.changeset/lucky-jars-refactor.md b/.changeset/lucky-jars-refactor.md new file mode 100644 index 00000000000..53b30862b94 --- /dev/null +++ b/.changeset/lucky-jars-refactor.md @@ -0,0 +1,5 @@ +--- +"@hashintel/petrinaut": patch +--- + +Split the optimizations provider's reconnect policy, transport-error handling, and stored-run bookkeeping into their own modules, and turn the attach loop's failure handling into one pure decision function. No behavioural change. diff --git a/libs/@hashintel/petrinaut/src/react/optimizations/active-run-storage.ts b/libs/@hashintel/petrinaut/src/react/optimizations/active-run-storage.ts new file mode 100644 index 00000000000..17b61d5775d --- /dev/null +++ b/libs/@hashintel/petrinaut/src/react/optimizations/active-run-storage.ts @@ -0,0 +1,71 @@ +/** + * The detached runs this tab may re-attach to after a reload, recorded in + * sessionStorage as a JSON object mapping run id to its manifest and creation + * time. Session-scoped on purpose — a run belongs to the tab that started it. + * + * When storage is unavailable (e.g. Petrinaut runs in a sandboxed iframe with + * an opaque origin) every helper degrades to a no-op: reload re-attachment is + * lost, while in-page reconnection keeps working. + */ + +import type { PetrinautOptimizationInput } from "@hashintel/petrinaut-core"; + +export const ACTIVE_RUNS_STORAGE_KEY = "petrinaut:active-optimization-runs"; + +export type StoredActiveRun = { input: unknown; createdAt: number }; + +export const readStoredActiveRuns = (): Record => { + try { + const raw = sessionStorage.getItem(ACTIVE_RUNS_STORAGE_KEY); + if (!raw) { + return {}; + } + const parsed: unknown = JSON.parse(raw); + if ( + typeof parsed !== "object" || + parsed === null || + Array.isArray(parsed) + ) { + return {}; + } + const runs: Record = {}; + for (const [runId, value] of Object.entries(parsed)) { + if (typeof value === "object" && value !== null && "input" in value) { + const createdAt = (value as { createdAt?: unknown }).createdAt; + runs[runId] = { + input: (value as { input: unknown }).input, + createdAt: typeof createdAt === "number" ? createdAt : Date.now(), + }; + } + } + return runs; + } catch { + // Unavailable or corrupted storage; see the module comment. + return {}; + } +}; + +const writeStoredActiveRuns = (runs: Record): void => { + try { + sessionStorage.setItem(ACTIVE_RUNS_STORAGE_KEY, JSON.stringify(runs)); + } catch { + // Unavailable storage or exceeded quota; see the module comment. + } +}; + +export const storeActiveRun = ( + runId: string, + input: PetrinautOptimizationInput, +): void => { + const runs = readStoredActiveRuns(); + runs[runId] = { input, createdAt: Date.now() }; + writeStoredActiveRuns(runs); +}; + +export const removeStoredActiveRun = (runId: string): void => { + const runs = readStoredActiveRuns(); + if (runId in runs) { + delete runs[runId]; + writeStoredActiveRuns(runs); + } +}; diff --git a/libs/@hashintel/petrinaut/src/react/optimizations/provider.test.tsx b/libs/@hashintel/petrinaut/src/react/optimizations/provider.test.tsx index 44810f6e935..586f0ec475a 100644 --- a/libs/@hashintel/petrinaut/src/react/optimizations/provider.test.tsx +++ b/libs/@hashintel/petrinaut/src/react/optimizations/provider.test.tsx @@ -13,6 +13,7 @@ import { import { sirModel } from "@hashintel/petrinaut-core/examples"; import { PetrinautOptimizationContext } from "../optimization-context"; +import { ACTIVE_RUNS_STORAGE_KEY } from "./active-run-storage"; import { OptimizationsContext, type OptimizationsContextValue, @@ -414,9 +415,7 @@ describe("OptimizationsProvider", () => { // does not prove the server acted (some hosts fire-and-forget), so the // next reload's re-attach settles the run's true fate instead. expect(cancelledRunIds).toEqual(["run-3"]); - expect( - sessionStorage.getItem("petrinaut:active-optimization-runs"), - ).toContain("run-3"); + expect(sessionStorage.getItem(ACTIVE_RUNS_STORAGE_KEY)).toContain("run-3"); }); it("lets Remove cancel a possibly-live run after a terminal error", async () => { @@ -487,7 +486,7 @@ describe("OptimizationsProvider", () => { it("re-attaches to stored runs after a reload, rebuilding from a full replay", async () => { sessionStorage.setItem( - "petrinaut:active-optimization-runs", + ACTIVE_RUNS_STORAGE_KEY, JSON.stringify({ "run-5": { input, createdAt: 123 } }), ); const cursors: number[] = []; @@ -522,9 +521,7 @@ describe("OptimizationsProvider", () => { // The best was rebuilt locally from the replayed trial. expect(optimization.best?.trial).toBe(0); // The settled run was forgotten so the next reload doesn't re-attach. - expect(sessionStorage.getItem("petrinaut:active-optimization-runs")).toBe( - "{}", - ); + expect(sessionStorage.getItem(ACTIVE_RUNS_STORAGE_KEY)).toBe("{}"); }); it("settles a replayed cancellation as cancelled rather than failed", async () => { @@ -532,7 +529,7 @@ describe("OptimizationsProvider", () => { // stored entry, expecting the next reload to settle it. That replay must // report Cancelled — not a failed run offering Retry. sessionStorage.setItem( - "petrinaut:active-optimization-runs", + ACTIVE_RUNS_STORAGE_KEY, JSON.stringify({ "run-6": { input, createdAt: 123 } }), ); const capability: PetrinautOptimization = { @@ -559,14 +556,12 @@ describe("OptimizationsProvider", () => { expect(optimization.errorCategory).toBeNull(); // The trial applied before the cancellation is still part of the record. expect(optimization.trials).toHaveLength(1); - expect(sessionStorage.getItem("petrinaut:active-optimization-runs")).toBe( - "{}", - ); + expect(sessionStorage.getItem(ACTIVE_RUNS_STORAGE_KEY)).toBe("{}"); }); it("silently drops a stored run the service no longer knows", async () => { sessionStorage.setItem( - "petrinaut:active-optimization-runs", + ACTIVE_RUNS_STORAGE_KEY, JSON.stringify({ "run-6": { input, createdAt: 123 } }), ); const capability: PetrinautOptimization = { @@ -583,9 +578,7 @@ describe("OptimizationsProvider", () => { const getValue = renderProvider(capability); await waitFor(() => expect(getValue().optimizations).toHaveLength(0)); - expect(sessionStorage.getItem("petrinaut:active-optimization-runs")).toBe( - "{}", - ); + expect(sessionStorage.getItem(ACTIVE_RUNS_STORAGE_KEY)).toBe("{}"); }); it("treats a retryable NodeAPI error event as a dropped connection and reconnects", async () => { @@ -789,7 +782,7 @@ describe("OptimizationsProvider", () => { it("does not duplicate restored runs under StrictMode double-mounting", async () => { sessionStorage.setItem( - "petrinaut:active-optimization-runs", + ACTIVE_RUNS_STORAGE_KEY, JSON.stringify({ "run-12": { input, createdAt: 123 } }), ); const capability: PetrinautOptimization = { diff --git a/libs/@hashintel/petrinaut/src/react/optimizations/provider.tsx b/libs/@hashintel/petrinaut/src/react/optimizations/provider.tsx index b965d71790a..cf036c27d35 100644 --- a/libs/@hashintel/petrinaut/src/react/optimizations/provider.tsx +++ b/libs/@hashintel/petrinaut/src/react/optimizations/provider.tsx @@ -10,225 +10,33 @@ import { import { useBlockWindowClose } from "../hooks/use-block-window-close"; import { PetrinautOptimizationContext } from "../optimization-context"; +import { + readStoredActiveRuns, + removeStoredActiveRun, + storeActiveRun, +} from "./active-run-storage"; import { type OptimizationBest, - type OptimizationErrorCategory, - type OptimizationErrorDiagnostics, isOptimizationActive, type OptimizationRecord, OptimizationsContext, type OptimizationsContextValue, } from "./context"; +import { abortableDelay, decideAttachFailure } from "./reconnect-policy"; +import { + buildErrorMessage, + type ClassifiedError, + classifyError, + isAbortFailure, +} from "./transport-errors"; import type { PropsWithChildren } from "react"; -const ERROR_CATEGORIES = new Set([ - "network", - "http", - "protocol", - "aborted", -]); - -/** First reconnect delay after a dropped detached-run event stream. */ -const RECONNECT_BASE_DELAY_MS = 1_000; -/** Ceiling for the exponential reconnect backoff. */ -const RECONNECT_MAX_DELAY_MS = 30_000; -/** - * Consecutive failed attachments (no event received in between) after which - * reconnecting stops and the classified failure is surfaced instead. - */ -const MAX_CONSECUTIVE_RECONNECT_FAILURES = 8; - -/** - * Gateway statuses a re-attach may transiently hit while the service - * restarts or deploys; they reconnect within the same failure cap. Every - * other http status (404 unknown run, other 4xx) is definitive. - */ -const RECONNECTABLE_HTTP_STATUSES = new Set([502, 503, 504]); - -/** Exponential backoff: 1s, 2s, 4s, ... capped at 30s. */ -const reconnectDelayMs = (consecutiveFailures: number): number => - Math.min( - RECONNECT_BASE_DELAY_MS * 2 ** (consecutiveFailures - 1), - RECONNECT_MAX_DELAY_MS, - ); - -/** Resolve after `ms`, or immediately once `signal` aborts. */ -const abortableDelay = (ms: number, signal: AbortSignal): Promise => - new Promise((resolve) => { - if (signal.aborted) { - resolve(); - return; - } - const timer = setTimeout(resolve, ms); - // The listener stays attached when the delay elapses normally: at most a - // handful accumulate per run, and they die with the run's controller. - signal.addEventListener( - "abort", - () => { - clearTimeout(timer); - resolve(); - }, - { once: true }, - ); - }); - -/** - * sessionStorage key recording the detached runs this tab may re-attach to - * after a reload: a JSON object mapping run id to its manifest and creation - * time. Session-scoped on purpose — a run belongs to the tab that started it. - * - * When storage is unavailable (e.g. Petrinaut runs in a sandboxed iframe with - * an opaque origin) every helper degrades to a no-op: reload re-attachment is - * lost, while in-page reconnection keeps working. - */ -const ACTIVE_RUNS_STORAGE_KEY = "petrinaut:active-optimization-runs"; - -type StoredActiveRun = { input: unknown; createdAt: number }; - -const readStoredActiveRuns = (): Record => { - try { - const raw = sessionStorage.getItem(ACTIVE_RUNS_STORAGE_KEY); - if (!raw) { - return {}; - } - const parsed: unknown = JSON.parse(raw); - if ( - typeof parsed !== "object" || - parsed === null || - Array.isArray(parsed) - ) { - return {}; - } - const runs: Record = {}; - for (const [runId, value] of Object.entries(parsed)) { - if (typeof value === "object" && value !== null && "input" in value) { - const createdAt = (value as { createdAt?: unknown }).createdAt; - runs[runId] = { - input: (value as { input: unknown }).input, - createdAt: typeof createdAt === "number" ? createdAt : Date.now(), - }; - } - } - return runs; - } catch { - // Unavailable or corrupted storage; see ACTIVE_RUNS_STORAGE_KEY. - return {}; - } -}; - -const writeStoredActiveRuns = (runs: Record): void => { - try { - sessionStorage.setItem(ACTIVE_RUNS_STORAGE_KEY, JSON.stringify(runs)); - } catch { - // Unavailable storage or exceeded quota; see ACTIVE_RUNS_STORAGE_KEY. - } -}; - -const storeActiveRun = ( - runId: string, - input: PetrinautOptimizationInput, -): void => { - const runs = readStoredActiveRuns(); - runs[runId] = { input, createdAt: Date.now() }; - writeStoredActiveRuns(runs); -}; - -const removeStoredActiveRun = (runId: string): void => { - const runs = readStoredActiveRuns(); - if (runId in runs) { - delete runs[runId]; - writeStoredActiveRuns(runs); - } -}; - -type ClassifiedError = { - category: OptimizationErrorCategory; - /** Seconds from a `Retry-After` header, when the service sent one (429). */ - retryAfter: number | null; - diagnostics: OptimizationErrorDiagnostics; -}; - -function isAbortError(error: unknown): boolean { - return ( - (error instanceof DOMException && error.name === "AbortError") || - (error instanceof Error && error.name === "AbortError") - ); -} - -/** - * Read the structured fields off a classified transport error without - * depending on the host bridge's class: the error crosses from the app into - * this library, so it is duck-typed rather than matched with `instanceof`. - */ -function classifyError(error: unknown): ClassifiedError | null { - if (typeof error !== "object" || error === null) { - return null; - } - const candidate = error as Record; - if ( - typeof candidate.category !== "string" || - !ERROR_CATEGORIES.has(candidate.category as OptimizationErrorCategory) - ) { - return null; - } - return { - category: candidate.category as OptimizationErrorCategory, - retryAfter: - typeof candidate.retryAfter === "number" ? candidate.retryAfter : null, - diagnostics: { - hashRequestId: - typeof candidate.hashRequestId === "string" - ? candidate.hashRequestId - : null, - optimizationRunId: - typeof candidate.optimizationRunId === "string" - ? candidate.optimizationRunId - : null, - httpStatus: - typeof candidate.httpStatus === "number" ? candidate.httpStatus : null, - }, - }; -} - -/** Build a safe, actionable message from a classified failure. */ -function buildErrorMessage( - classified: ClassifiedError, - progress: { completedTrials: number; requestedTrials: number }, -): string { - const after = `after ${progress.completedTrials} of ${progress.requestedTrials} trials`; - const { httpStatus, optimizationRunId, hashRequestId } = - classified.diagnostics; - const diagnosticId = optimizationRunId ?? hashRequestId; - const diagnostic = diagnosticId ? ` (diagnostic id: ${diagnosticId})` : ""; - - switch (classified.category) { - case "http": - if (httpStatus === 429) { - return `The optimization service is busy — another optimization may already be running for your account.${ - classified.retryAfter === null - ? "" - : ` Try again in ~${classified.retryAfter}s.` - }${diagnostic}`; - } - return `The optimization service rejected the request${ - httpStatus === null ? "" : ` (status ${httpStatus})` - } ${after}. Retry the optimization.${diagnostic}`; - case "protocol": - return `The optimization stream ended unexpectedly ${after}. Retry the optimization.${diagnostic}`; - case "aborted": - return "The optimization was cancelled."; - case "network": - default: - return `Connection to the optimization service was interrupted ${after}. Retry the optimization.${diagnostic}`; - } -} - /** * Fold a completed trial into the running best. Attachments deliver - * `best: null` (the service no longer knows the objective direction after - * the creating request ends), so the provider maintains the best itself from - * every trial it applies; `event.best` is still preferred when present. + * `best: null`, so the provider maintains the best itself from every trial it + * applies; `event.best` is still preferred when present, and would be + * authoritative if the optimizer ever stamped it onto replayable frames. */ const computeRunningBest = ( current: OptimizationRecord, @@ -460,28 +268,13 @@ export const OptimizationsProvider = ({ children }: PropsWithChildren) => { ); /** - * Consume a detached run's event stream, re-attaching with exponential - * backoff when the connection drops. Every reconnect resumes from the last - * applied `seq`, and replayed events at or below that cursor are skipped so - * trials are never double-counted. Reconnecting stops after - * {@link MAX_CONSECUTIVE_RECONNECT_FAILURES} attachments in a row that - * failed before yielding an event; the classified failure is surfaced then. - * - * Four kinds of interruption reconnect, all sharing the failure cap: - * `network` failures, `protocol` failures (a proxy tearing an idle - * connection down cleanly surfaces as a `protocol` "stream ended without a - * terminal event"), NodeAPI-authored `retryable: true` error events (its - * per-attachment window died while the run continues), and gateway - * `http` statuses (502/503/504 — NodeAPI restarting or deploying). - * Resuming from the cursor is safe in every case because replayed events - * are deduplicated. Every other `http` failure (404 unknown run, other - * 4xx) is definitive and fails immediately, as do `retryable: false` - * error events. + * Consume a detached run's event stream, applying each event to the record + * and re-attaching when the connection drops. Every reconnect resumes from + * the last applied `seq`, and replayed events at or below that cursor are + * skipped so trials are never double-counted. * - * On every give-up path the run — which may still be live server-side — - * is cancelled fire-and-forget: releasing NodeAPI's per-account ownership - * slot means a follow-up run (e.g. the drawer's Retry) isn't rejected as - * busy for the rest of the ownership TTL. + * Which failures reconnect, and for how long, is + * {@link decideAttachFailure}'s decision; this loop only carries it out. */ const runAttachLoop = useCallback( async ({ @@ -571,76 +364,67 @@ export const OptimizationsProvider = ({ children }: PropsWithChildren) => { const classified = classifyError(error); const retryableInterruption = error instanceof RetryableRunInterruption ? error : null; - if ( - isCancelled() || - isAbortError(error) || - classified?.category === "aborted" - ) { - markOptimizationCancelled(optimizationId); - return; - } - if (sawTerminalEvent) { - // The run already settled; a trailing transport hiccup after the - // terminal event changes nothing. - removeStoredActiveRun(runId); - return; - } - if ( - dropRecordOnNotFound && - !receivedAnyEvent && - classified?.category === "http" && - classified.diagnostics.httpStatus === 404 - ) { - removeStoredActiveRun(runId); - dropOptimizationRecord(optimizationId); - return; - } + // Counted before deciding: every decision other than `reconnect` + // leaves the loop, and only `reconnect` reads the tally. consecutiveFailures += 1; - const reconnectable = - retryableInterruption !== null || - classified?.category === "network" || - classified?.category === "protocol" || - (classified?.category === "http" && - classified.diagnostics.httpStatus !== null && - RECONNECTABLE_HTTP_STATUSES.has( - classified.diagnostics.httpStatus, - )); - if ( - reconnectable && - consecutiveFailures < MAX_CONSECUTIVE_RECONNECT_FAILURES - ) { - patchOptimization(optimizationId, (current) => ({ - ...current, - connectionState: "reconnecting", - })); - await abortableDelay(reconnectDelayMs(consecutiveFailures), signal); - if (isCancelled()) { + const decision = decideAttachFailure({ + error, + classified, + isRetryableInterruption: retryableInterruption !== null, + aborted: isCancelled(), + sawTerminalEvent, + receivedAnyEvent, + dropRecordOnNotFound, + consecutiveFailures, + }); + + switch (decision.kind) { + case "cancelled": markOptimizationCancelled(optimizationId); return; - } - continue; - } - // Give up. The run may still be live server-side; cancelling it - // frees the account's single-flight so a fresh run (e.g. the - // drawer's Retry) isn't rejected as busy. The stored entry is - // deliberately kept — some hosts' cancel resolves before the - // server acted, so resolution proves nothing. The next reload's - // re-attach settles it: a delivered cancel replays the cancelled - // terminal, a reaped run 404s (silently dropped), and a run the - // cancel never reached is recovered live. - void cancel(runId).catch(() => undefined); - if (retryableInterruption) { - // Reconnection is exhausted: NodeAPI's own terminal error event - // (a safe, server-authored message) becomes the run's outcome. - applyOptimizationEvent( - optimizationId, - retryableInterruption.event, - { extra: { lastSeq, connectionState: null } }, - ); - } else { - markOptimizationFailed(optimizationId, error, classified); + case "settled": + // A trailing transport hiccup after the terminal event changes + // nothing about the run's outcome. + removeStoredActiveRun(runId); + return; + case "expired": + removeStoredActiveRun(runId); + dropOptimizationRecord(optimizationId); + return; + case "reconnect": + patchOptimization(optimizationId, (current) => ({ + ...current, + connectionState: "reconnecting", + })); + await abortableDelay(decision.delayMs, signal); + if (isCancelled()) { + markOptimizationCancelled(optimizationId); + return; + } + continue; + case "giveUp": + // The run may still be live server-side; cancelling it frees the + // account's single-flight so a fresh run (e.g. the drawer's + // Retry) isn't rejected as busy. The stored entry is + // deliberately kept — some hosts' cancel resolves before the + // server acted, so resolution proves nothing. The next reload's + // re-attach settles it: a delivered cancel replays the cancelled + // terminal, a reaped run 404s (silently dropped), and a run the + // cancel never reached is recovered live. + void cancel(runId).catch(() => undefined); + if (retryableInterruption) { + // Reconnection is exhausted: NodeAPI's own terminal error + // event (a safe, server-authored message) is the outcome. + applyOptimizationEvent( + optimizationId, + retryableInterruption.event, + { extra: { lastSeq, connectionState: null } }, + ); + } else { + markOptimizationFailed(optimizationId, error, classified); + } + return; } - return; } } // Aborted between attachments (e.g. while waiting to reconnect). The @@ -682,9 +466,7 @@ export const OptimizationsProvider = ({ children }: PropsWithChildren) => { } catch (error) { const classified = classifyError(error); if ( - abortController.signal.aborted || - isAbortError(error) || - classified?.category === "aborted" + isAbortFailure(error, classified, abortController.signal.aborted) ) { markOptimizationCancelled(optimizationId); } else { diff --git a/libs/@hashintel/petrinaut/src/react/optimizations/reconnect-policy.test.ts b/libs/@hashintel/petrinaut/src/react/optimizations/reconnect-policy.test.ts new file mode 100644 index 00000000000..2a5e7523ca5 --- /dev/null +++ b/libs/@hashintel/petrinaut/src/react/optimizations/reconnect-policy.test.ts @@ -0,0 +1,152 @@ +import { describe, expect, it } from "vitest"; + +import { + type AttachFailureInput, + decideAttachFailure, + MAX_CONSECUTIVE_RECONNECT_FAILURES, + reconnectDelayMs, +} from "./reconnect-policy"; + +import type { ClassifiedError } from "./transport-errors"; + +const classified = ( + category: ClassifiedError["category"], + httpStatus: number | null = null, +): ClassifiedError => ({ + category, + retryAfter: null, + diagnostics: { hashRequestId: null, optimizationRunId: null, httpStatus }, +}); + +const failure = ( + overrides: Partial = {}, +): AttachFailureInput => ({ + error: new Error("dropped"), + classified: classified("network"), + isRetryableInterruption: false, + aborted: false, + sawTerminalEvent: false, + receivedAnyEvent: true, + dropRecordOnNotFound: false, + consecutiveFailures: 1, + ...overrides, +}); + +describe("reconnectDelayMs", () => { + it("doubles from one second and caps at thirty", () => { + expect([1, 2, 3, 4, 5, 6].map(reconnectDelayMs)).toEqual([ + 1_000, 2_000, 4_000, 8_000, 16_000, 30_000, + ]); + }); +}); + +describe("decideAttachFailure", () => { + it("reconnects a network drop with backoff", () => { + expect(decideAttachFailure(failure({ consecutiveFailures: 3 }))).toEqual({ + kind: "reconnect", + delayMs: 4_000, + }); + }); + + it("reconnects a protocol failure and a gateway status", () => { + for (const candidate of [ + classified("protocol"), + classified("http", 502), + classified("http", 503), + classified("http", 504), + ]) { + expect(decideAttachFailure(failure({ classified: candidate })).kind).toBe( + "reconnect", + ); + } + }); + + it("reconnects a retryable interruption even when unclassified", () => { + expect( + decideAttachFailure( + failure({ classified: null, isRetryableInterruption: true }), + ).kind, + ).toBe("reconnect"); + }); + + it("gives up on a definitive http failure without retrying", () => { + for (const candidate of [ + classified("http", 404), + classified("http", 400), + classified("http", 500), + ]) { + expect(decideAttachFailure(failure({ classified: candidate })).kind).toBe( + "giveUp", + ); + } + }); + + it("gives up once the failure cap is reached", () => { + expect( + decideAttachFailure( + failure({ + consecutiveFailures: MAX_CONSECUTIVE_RECONNECT_FAILURES - 1, + }), + ).kind, + ).toBe("reconnect"); + expect( + decideAttachFailure( + failure({ consecutiveFailures: MAX_CONSECUTIVE_RECONNECT_FAILURES }), + ).kind, + ).toBe("giveUp"); + }); + + it("treats an abort as a cancellation however it arrives", () => { + const abortError = Object.assign(new Error("stop"), { name: "AbortError" }); + expect(decideAttachFailure(failure({ aborted: true })).kind).toBe( + "cancelled", + ); + expect(decideAttachFailure(failure({ error: abortError })).kind).toBe( + "cancelled", + ); + expect( + decideAttachFailure(failure({ classified: classified("aborted") })).kind, + ).toBe("cancelled"); + }); + + it("prefers cancellation over every other outcome", () => { + // An abort during the reconnect wait must not be reported as a failure. + expect( + decideAttachFailure( + failure({ + aborted: true, + sawTerminalEvent: true, + consecutiveFailures: MAX_CONSECUTIVE_RECONNECT_FAILURES, + }), + ).kind, + ).toBe("cancelled"); + }); + + it("treats a failure after a terminal event as already settled", () => { + expect( + decideAttachFailure( + failure({ + sawTerminalEvent: true, + classified: classified("http", 404), + }), + ).kind, + ).toBe("settled"); + }); + + it("only calls a stored run expired on a 404 before its first event", () => { + const expired = failure({ + dropRecordOnNotFound: true, + receivedAnyEvent: false, + classified: classified("http", 404), + }); + expect(decideAttachFailure(expired).kind).toBe("expired"); + // Mid-run, the same 404 is a real failure worth surfacing. + expect( + decideAttachFailure({ ...expired, receivedAnyEvent: true }).kind, + ).toBe("giveUp"); + // And a live run's 404 is never silently dropped. + expect( + decideAttachFailure({ ...expired, dropRecordOnNotFound: false }).kind, + ).toBe("giveUp"); + }); +}); diff --git a/libs/@hashintel/petrinaut/src/react/optimizations/reconnect-policy.ts b/libs/@hashintel/petrinaut/src/react/optimizations/reconnect-policy.ts new file mode 100644 index 00000000000..adbe0ebf3bd --- /dev/null +++ b/libs/@hashintel/petrinaut/src/react/optimizations/reconnect-policy.ts @@ -0,0 +1,150 @@ +/** + * When a detached run's event stream drops, whether to re-attach and how long + * to wait first. Kept free of React and of the record store so the policy can + * be read — and tested — without driving a provider through fake streams. + */ + +import { isAbortFailure } from "./transport-errors"; + +import type { ClassifiedError } from "./transport-errors"; + +/** First reconnect delay after a dropped detached-run event stream. */ +const RECONNECT_BASE_DELAY_MS = 1_000; +/** Ceiling for the exponential reconnect backoff. */ +const RECONNECT_MAX_DELAY_MS = 30_000; +/** + * Consecutive failed attachments (no event received in between) after which + * reconnecting stops and the classified failure is surfaced instead. + */ +export const MAX_CONSECUTIVE_RECONNECT_FAILURES = 8; + +/** + * Gateway statuses a re-attach may transiently hit while the service + * restarts or deploys; they reconnect within the same failure cap. Every + * other http status (404 unknown run, other 4xx) is definitive. + */ +const RECONNECTABLE_HTTP_STATUSES = new Set([502, 503, 504]); + +/** Exponential backoff: 1s, 2s, 4s, ... capped at 30s. */ +export const reconnectDelayMs = (consecutiveFailures: number): number => + Math.min( + RECONNECT_BASE_DELAY_MS * 2 ** (consecutiveFailures - 1), + RECONNECT_MAX_DELAY_MS, + ); + +/** Resolve after `ms`, or immediately once `signal` aborts. */ +export const abortableDelay = ( + ms: number, + signal: AbortSignal, +): Promise => + new Promise((resolve) => { + if (signal.aborted) { + resolve(); + return; + } + const timer = setTimeout(resolve, ms); + // The listener stays attached when the delay elapses normally: at most a + // handful accumulate per run, and they die with the run's controller. + signal.addEventListener( + "abort", + () => { + clearTimeout(timer); + resolve(); + }, + { once: true }, + ); + }); + +/** + * Whether resuming from the cursor could recover this failure. Replayed + * events are deduplicated by `seq`, so re-attaching is always safe — the + * question is only whether the run may still be live upstream. + * + * Four kinds of interruption qualify, all sharing the failure cap: `network` + * failures, `protocol` failures (a proxy tearing an idle connection down + * cleanly surfaces as a `protocol` "stream ended without a terminal event"), + * NodeAPI-authored `retryable: true` error events (its per-attachment window + * died while the run continues), and gateway `http` statuses (502/503/504 — + * NodeAPI restarting or deploying). Every other `http` failure (404 unknown + * run, other 4xx) is definitive, as are `retryable: false` error events. + */ +const isReconnectable = ( + classified: ClassifiedError | null, + isRetryableInterruption: boolean, +): boolean => + isRetryableInterruption || + classified?.category === "network" || + classified?.category === "protocol" || + (classified?.category === "http" && + classified.diagnostics.httpStatus !== null && + RECONNECTABLE_HTTP_STATUSES.has(classified.diagnostics.httpStatus)); + +/** What the attach loop should do about a failure. */ +export type AttachFailureDecision = + /** The run's own cancellation; settle the record as cancelled. */ + | { kind: "cancelled" } + /** The run already reached a terminal event; the failure changes nothing. */ + | { kind: "settled" } + /** A stored run the service no longer knows; drop it without a trace. */ + | { kind: "expired" } + /** Re-attach from the cursor after waiting this long. */ + | { kind: "reconnect"; delayMs: number } + /** Out of reconnects, or never reconnectable; surface the failure. */ + | { kind: "giveUp" }; + +export type AttachFailureInput = { + error: unknown; + classified: ClassifiedError | null; + /** A NodeAPI attachment window died while the run itself may continue. */ + isRetryableInterruption: boolean; + /** The consumer asked to stop, so any failure is really a cancellation. */ + aborted: boolean; + sawTerminalEvent: boolean; + /** False when this attachment produced nothing before failing. */ + receivedAnyEvent: boolean; + /** Set while re-attaching to a stored run that may have expired. */ + dropRecordOnNotFound: boolean; + /** Failures in a row without an event in between, including this one. */ + consecutiveFailures: number; +}; + +/** + * Classify one attach failure. Ordering matters: cancellation and a run that + * already settled both outrank the reconnect logic, and a 404 on the first + * attachment of a stored run means it expired rather than failed. + */ +export const decideAttachFailure = ({ + error, + classified, + isRetryableInterruption, + aborted, + sawTerminalEvent, + receivedAnyEvent, + dropRecordOnNotFound, + consecutiveFailures, +}: AttachFailureInput): AttachFailureDecision => { + if (isAbortFailure(error, classified, aborted)) { + return { kind: "cancelled" }; + } + if (sawTerminalEvent) { + return { kind: "settled" }; + } + if ( + dropRecordOnNotFound && + !receivedAnyEvent && + classified?.category === "http" && + classified.diagnostics.httpStatus === 404 + ) { + return { kind: "expired" }; + } + if ( + isReconnectable(classified, isRetryableInterruption) && + consecutiveFailures < MAX_CONSECUTIVE_RECONNECT_FAILURES + ) { + return { + kind: "reconnect", + delayMs: reconnectDelayMs(consecutiveFailures), + }; + } + return { kind: "giveUp" }; +}; diff --git a/libs/@hashintel/petrinaut/src/react/optimizations/transport-errors.ts b/libs/@hashintel/petrinaut/src/react/optimizations/transport-errors.ts new file mode 100644 index 00000000000..2b94e47c3e1 --- /dev/null +++ b/libs/@hashintel/petrinaut/src/react/optimizations/transport-errors.ts @@ -0,0 +1,111 @@ +/** + * Reading the structured fields off an optimization transport failure, and + * turning one into a message a user can act on. + */ + +import type { + OptimizationErrorCategory, + OptimizationErrorDiagnostics, +} from "./context"; + +const ERROR_CATEGORIES = new Set([ + "network", + "http", + "protocol", + "aborted", +]); + +export type ClassifiedError = { + category: OptimizationErrorCategory; + /** Seconds from a `Retry-After` header, when the service sent one (429). */ + retryAfter: number | null; + diagnostics: OptimizationErrorDiagnostics; +}; + +export function isAbortError(error: unknown): boolean { + return ( + (error instanceof DOMException && error.name === "AbortError") || + (error instanceof Error && error.name === "AbortError") + ); +} + +/** + * Read the structured fields off a classified transport error without + * depending on the host bridge's class: the error crosses from the app into + * this library, so it is duck-typed rather than matched with `instanceof`. + */ +export function classifyError(error: unknown): ClassifiedError | null { + if (typeof error !== "object" || error === null) { + return null; + } + const candidate = error as Record; + if ( + typeof candidate.category !== "string" || + !ERROR_CATEGORIES.has(candidate.category as OptimizationErrorCategory) + ) { + return null; + } + return { + category: candidate.category as OptimizationErrorCategory, + retryAfter: + typeof candidate.retryAfter === "number" ? candidate.retryAfter : null, + diagnostics: { + hashRequestId: + typeof candidate.hashRequestId === "string" + ? candidate.hashRequestId + : null, + optimizationRunId: + typeof candidate.optimizationRunId === "string" + ? candidate.optimizationRunId + : null, + httpStatus: + typeof candidate.httpStatus === "number" ? candidate.httpStatus : null, + }, + }; +} + +/** + * Every way a run's own cancellation reaches us as a thrown value: the local + * signal already aborted, the host threw an `AbortError`, or the bridge + * classified the failure as `aborted`. One predicate so run creation and the + * attach loop cannot disagree about what counts as a cancellation. + */ +export const isAbortFailure = ( + error: unknown, + classified: ClassifiedError | null, + aborted: boolean, +): boolean => + aborted || isAbortError(error) || classified?.category === "aborted"; + +/** Build a safe, actionable message from a classified failure. */ +export function buildErrorMessage( + classified: ClassifiedError, + progress: { completedTrials: number; requestedTrials: number }, +): string { + const after = `after ${progress.completedTrials} of ${progress.requestedTrials} trials`; + const { httpStatus, optimizationRunId, hashRequestId } = + classified.diagnostics; + const diagnosticId = optimizationRunId ?? hashRequestId; + const diagnostic = diagnosticId ? ` (diagnostic id: ${diagnosticId})` : ""; + + switch (classified.category) { + case "http": + if (httpStatus === 429) { + return `The optimization service is busy — another optimization may already be running for your account.${ + classified.retryAfter === null + ? "" + : ` Try again in ~${classified.retryAfter}s.` + }${diagnostic}`; + } + return `The optimization service rejected the request${ + httpStatus === null ? "" : ` (status ${httpStatus})` + } ${after}. Retry the optimization.${diagnostic}`; + case "protocol": + return `The optimization stream ended unexpectedly ${after}. Retry the optimization.${diagnostic}`; + case "aborted": + return "The optimization was cancelled."; + case "network": + default: + return `Connection to the optimization service was interrupted ${after}. Retry the optimization.${diagnostic}`; + } +}