From 9385e433222d54667548b33318e87298d2e7f406 Mon Sep 17 00:00:00 2001 From: Jake Gaylor Date: Mon, 7 Sep 2026 01:13:48 -0400 Subject: [PATCH] fix(paddock): reload stale tabs, honour 429s, and resume streams instead of replaying After #67 deployed (04:56Z) the account-wide list calls fell 14.5/s -> 1.1/s, but one browser tab still running the OLD bundle went into a hotter loop: 3,277 stream opens on one conversation and 4,319 receipt reads on one sandbox in five minutes, events?limit=1000 at ~7/s, and Fountain answering 429 rate_limited with Retry-After up to 35 s. The old bundle's loop was: poll -> new array -> new Tab object -> stream effect re-runs -> reopen with no Last-Event-ID -> replay the 40-turn history -> each replayed turn end re-reads the receipt and the strip. It ran at ~2/s before only because every refresh paid a 200-500 ms unfiltered list on the server; the 5 s memo made that instant, so the same loop spun 5-10x faster. And `backoff` was a local of the effect, reset to 1 s on every re-run, so the 429s slowed nothing. Four changes, each of which stands alone: 1. Stale-bundle self-heal. The server's build id is the hash of the built index.html (so a server-only change reloads no tab). It is stamped on every response as x-paddock-build, written into the served HTML as , and exposed on /api/config. The bundle reads its own from the meta tag (not from the first API answer, which a deploy between the HTML and that call would poison) and reloads once on the first mismatch (src/lib/build.ts). The strip poll refuses a request whose header does not name the current build (409 stale_client): a current tab reloads on the mismatch anyway; an old tab, which knows nothing of builds, logs a failed poll and keeps the strip it has -- and stops, because its loop only turned on a *successful* poll. That is the lever over the tab looping now. 2. Honour 429. src/lib/hold.ts is one shared pause: any 429 sets it to Retry-After (default 15 s) plus jitter, and the poll, the receipt read, the scrollback fetch and the stream reconnect all wait on it. The stream open is now made by the client itself so a non-200 reaches the caller as an ApiError with status and Retry-After (the suite's readSse reported it as "the stream closed"); the read loop is the suite's, verbatim. 3. The tail is a state machine (src/lib/tail.ts), not an effect. Last-Event-ID is kept per conversation across reconnects; the backoff resets only when a stream actually opens (onOpen fires only after a 200); a 429 waits what the server asked. App.tsx's effect depends on [client, activeId, hold] and reads its three callbacks through a ref, so no state change can re-run it. Tests pin: a turn-ended stage event does not reopen the stream and reports once; the second open carries the id the first saw; backoff 1/2/4 s and reset on open only; a 429 waits 35 s; stop means stop. 4. Bounded replay. The catch-up read on open passes after=, so a reconnect fetches the gap, not limit=1000 pages of history. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_018R1XyyWd9MDMYXqFL71owk --- apps/paddock/server/app.test.ts | 53 +++++++++++- apps/paddock/server/app.ts | 51 ++++++++++- apps/paddock/server/auth.ts | 2 +- apps/paddock/server/context.ts | 7 ++ apps/paddock/server/index.ts | 13 +++ apps/paddock/server/proxy.ts | 12 +++ apps/paddock/src/App.tsx | 121 +++++++++++++------------- apps/paddock/src/api/client.ts | 70 ++++++++++++--- apps/paddock/src/api/paddock.ts | 6 +- apps/paddock/src/lib/build.test.ts | 36 ++++++++ apps/paddock/src/lib/build.ts | 59 +++++++++++++ apps/paddock/src/lib/hold.test.ts | 30 +++++++ apps/paddock/src/lib/hold.ts | 45 ++++++++++ apps/paddock/src/lib/tail.test.ts | 129 ++++++++++++++++++++++++++++ apps/paddock/src/lib/tail.ts | 132 +++++++++++++++++++++++++++++ 15 files changed, 681 insertions(+), 85 deletions(-) create mode 100644 apps/paddock/src/lib/build.test.ts create mode 100644 apps/paddock/src/lib/build.ts create mode 100644 apps/paddock/src/lib/hold.test.ts create mode 100644 apps/paddock/src/lib/hold.ts create mode 100644 apps/paddock/src/lib/tail.test.ts create mode 100644 apps/paddock/src/lib/tail.ts diff --git a/apps/paddock/server/app.test.ts b/apps/paddock/server/app.test.ts index 6e98190..35ef6a5 100644 --- a/apps/paddock/server/app.test.ts +++ b/apps/paddock/server/app.test.ts @@ -176,7 +176,7 @@ beforeEach(async () => { }), dbPath: ":memory:", }; - ctx = { config, db: new Db(":memory:"), cipher: await Cipher.from(config.secret) }; + ctx = { config, db: new Db(":memory:"), cipher: await Cipher.from(config.secret), buildId: BUILD }; route = buildRouter(ctx); upstream = []; hub.reset(); @@ -264,9 +264,13 @@ async function guestSession(paddockId: string, conversationId = "c1"): Promise { +/** The build id this fake server is, and what a current tab sends. */ +const BUILD = "build-1"; + +function call(cookie: string | null, method: string, path: string, body?: unknown, build: string | null = BUILD): Promise { const headers: Record = {}; if (cookie) headers.cookie = cookie; + if (build) headers["x-paddock-build"] = build; if (body !== undefined) headers["content-type"] = "application/json"; return route(new Request(`http://paddock.test${path}`, { method, headers, body: body === undefined ? undefined : JSON.stringify(body) })); } @@ -305,6 +309,51 @@ async function addComputer(cookie: string, name?: string): Promise { return ((await res.json()) as { data: { id: string } }).data.id; } +describe("a tab running an older build", () => { + test("every answer names the build, refusals included", async () => { + const owner = await paddockFor(OWNER); + expect((await call(owner.cookie, "GET", `/f/${owner.id}/api/conversations`)).headers.get("x-paddock-build")).toBe(BUILD); + expect((await call(owner.cookie, "GET", "/api/config")).headers.get("x-paddock-build")).toBe(BUILD); + expect((await call(null, "GET", "/api/nope")).headers.get("x-paddock-build")).toBe(BUILD); + const cfg = (await (await call(null, "GET", "/api/config")).json()) as { buildId: string }; + expect(cfg.buildId).toBe(BUILD); + }); + + test("is turned away from the strip, and nowhere else", async () => { + const owner = await paddockFor(OWNER); + for (const stale of [null, "build-0"]) { + const res = await call(owner.cookie, "GET", `/f/${owner.id}/api/conversations`, undefined, stale); + expect(res.status).toBe(409); + expect(((await res.json()) as { error: string }).error).toBe("stale_client"); + // The rest of the machine still works for it: the tab it is on, the + // files, a prompt. Refusing those would break the tab harder than the + // loop being stopped ever did. + expect((await call(owner.cookie, "GET", `/f/${owner.id}/api/conversations/c1`, undefined, stale)).status).toBe(200); + expect((await call(owner.cookie, "GET", `/f/${owner.id}/api/sandboxes/${BOX}/files?path=/`, undefined, stale)).status).toBe(200); + expect((await call(owner.cookie, "POST", `/f/${owner.id}/api/conversations/c1/prompts`, { prompt: "hi" }, stale)).status).toBe(200); + } + }); + + test("an unstamped server turns nobody away", async () => { + ctx.buildId = undefined; + const owner = await paddockFor(OWNER); + const res = await call(owner.cookie, "GET", `/f/${owner.id}/api/conversations`, undefined, null); + expect(res.status).toBe(200); + expect(res.headers.get("x-paddock-build")).toBeNull(); + }); + + test("the HTML it serves carries the build it belongs to", async () => { + const dir = join(tmpdir(), `paddock-static-${randomToken(6)}`); + const { mkdirSync, writeFileSync } = await import("node:fs"); + mkdirSync(dir, { recursive: true }); + writeFileSync(join(dir, "index.html"), "p"); + ctx.config = { ...ctx.config, staticDir: dir }; + const html = await (await call(null, "GET", "/anything")).text(); + expect(html).toContain(``); + rmSync(dir, { recursive: true, force: true }); + }); +}); + describe("one Fountain call per burst", () => { const lists = () => upstream.filter((u) => u.method === "GET" && u.path === "/api/conversations"); diff --git a/apps/paddock/server/app.ts b/apps/paddock/server/app.ts index 51ab20a..beb53c7 100644 --- a/apps/paddock/server/app.ts +++ b/apps/paddock/server/app.ts @@ -97,7 +97,7 @@ export function buildRouter(ctx: AppContext): (req: Request) => Promise => { + const dispatch = async (req: Request): Promise => { const url = new URL(req.url); const path = url.pathname; try { @@ -115,8 +115,27 @@ export function buildRouter(ctx: AppContext): (req: Request) => Promise => { + const res = await dispatch(req); + if (ctx.buildId) { + try { + res.headers.set(BUILD_HEADER, ctx.buildId); + } catch { + // An immutable header set (a response handed straight through from + // fetch). The next answer will carry it. + } + } + return res; + }; } +/** The request header a browser sends naming the build it is running. */ +export const BUILD_HEADER = "x-paddock-build"; + /** `:name` captures a segment; a trailing `*` captures the rest as `rest`. */ function match(pattern: string[], segments: string[]): Record | null { const params: Record = {}; @@ -142,7 +161,33 @@ async function serveStatic(ctx: AppContext, path: string): Promise { if (rel.includes("..")) return json({ error: "not_found" }, 404); const file = Bun.file(`${ctx.config.staticDir}/${rel}`); if (await file.exists()) return new Response(file); - const index = Bun.file(`${ctx.config.staticDir}/index.html`); - if (await index.exists()) return new Response(index, { headers: { "content-type": "text/html; charset=utf-8" } }); + const html = await indexHtml(ctx); + if (html !== null) return new Response(html, { headers: { "content-type": "text/html; charset=utf-8" } }); return json({ error: "not_found" }, 404); } + +let indexCache: { dir: string | null; buildId: string | undefined; html: string | null } | null = null; + +/** + * The SPA's shell, stamped with the build it belongs to. + * + * The bundle learns its own build id from this tag rather than from its first + * API answer, because a deploy that lands between the HTML and that first call + * would otherwise teach an old bundle the new id — and it would never reload. + * Read once; the file does not change under a running server. + */ +async function indexHtml(ctx: AppContext): Promise { + const dir = ctx.config.staticDir; + if (indexCache && indexCache.dir === dir && indexCache.buildId === ctx.buildId) return indexCache.html; + const index = Bun.file(`${dir}/index.html`); + let html: string | null = null; + if (await index.exists()) { + html = await index.text(); + if (ctx.buildId) { + const tag = ``; + html = html.includes("") ? html.replace("", `${tag}`) : `${tag}${html}`; + } + } + indexCache = { dir, buildId: ctx.buildId, html }; + return html; +} diff --git a/apps/paddock/server/auth.ts b/apps/paddock/server/auth.ts index 1c66d8a..dbf115e 100644 --- a/apps/paddock/server/auth.ts +++ b/apps/paddock/server/auth.ts @@ -35,7 +35,7 @@ export function config(ctx: AppContext): Response { // `anonymousStart` is what tells the SPA whether to start a computer or show // the sign-in screen. It is a capability of the deployment, not of the // caller, which is why it sits on the unauthenticated config route. - return json({ fountainUrl: ctx.config.fountainUrl, anonymousStart: ctx.config.anonymousStart }); + return json({ fountainUrl: ctx.config.fountainUrl, anonymousStart: ctx.config.anonymousStart, buildId: ctx.buildId ?? null }); } /** diff --git a/apps/paddock/server/context.ts b/apps/paddock/server/context.ts index fedda0c..372b611 100644 --- a/apps/paddock/server/context.ts +++ b/apps/paddock/server/context.ts @@ -28,6 +28,13 @@ export interface AppContext { db: Db; cipher: Cipher; config: Config; + /** + * Which build this server is, stamped on every response and into the HTML + * it serves so a tab left open across a deploy notices and reloads (see + * `src/lib/build.ts`). Absent means unstamped — the dev server, and tests + * that do not care. + */ + buildId?: string; } /** diff --git a/apps/paddock/server/index.ts b/apps/paddock/server/index.ts index 28cdcbb..aef4225 100644 --- a/apps/paddock/server/index.ts +++ b/apps/paddock/server/index.ts @@ -11,10 +11,23 @@ const ctx: AppContext = { config, db: new Db(config.dbPath), cipher: await Cipher.from(config.secret), + buildId: await buildIdOf(config.staticDir), }; const fetch = buildRouter(ctx); +/** + * The build id is the built HTML's hash. Vite writes the bundle's content + * hashes into `index.html`, so any change to the client changes this, and a + * server-only change does not make every open tab reload for nothing. + */ +async function buildIdOf(staticDir: string | null): Promise { + if (!staticDir) return undefined; + const index = Bun.file(`${staticDir}/index.html`); + if (!(await index.exists())) return undefined; + return Bun.hash(await index.text()).toString(36); +} + Bun.serve({ port: config.port, fetch, idleTimeout: 0 }); /** diff --git a/apps/paddock/server/proxy.ts b/apps/paddock/server/proxy.ts index f303bb0..36eebcc 100644 --- a/apps/paddock/server/proxy.ts +++ b/apps/paddock/server/proxy.ts @@ -54,6 +54,7 @@ import { HttpError, readJson, str } from "./http"; import { withPromptLock } from "./prompt-lock"; import { hub } from "./hub"; import { agentHint, cached, forget, keyFor, rememberAgent } from "./machine-cache"; +import { BUILD_HEADER } from "./app"; /** What a role may do to one tab. Anything absent is a 404. */ function tabAllowed(method: string, sub: string, role: Role, claimed: boolean): boolean { @@ -135,6 +136,17 @@ export async function handleProxy(ctx: AppContext, req: Request, paddockId: stri // Filtered, always. The owner's raw conversation list would show a guest // every other conversation on the account, which is nobody's business here. if (method === "GET" && path === "/api/conversations") { + // The strip is what every tab polls, so it is where a tab running an + // older build is turned away. A current build reloads on the mismatch it + // sees in the response header; an older one, which knows nothing of + // builds, logs a failed poll and keeps the strip it has — and, crucially, + // stops. The bundle this replaced reopened its stream on every successful + // poll and replayed the tab's history each time, and a server fix that + // made the poll faster made that loop faster; refusing the poll is the + // one lever the server has over a tab nobody is going to find and reload. + if (ctx.buildId && req.headers.get(BUILD_HEADER) !== ctx.buildId) { + throw new HttpError(409, "stale_client", "This tab is running an older paddock. Reload the page."); + } const tabs = await visibleTabs(client, here, allowed); return jsonRes({ data: tabs.map((t) => t.conversation) }); } diff --git a/apps/paddock/src/App.tsx b/apps/paddock/src/App.tsx index acc7625..a1b20b3 100644 --- a/apps/paddock/src/App.tsx +++ b/apps/paddock/src/App.tsx @@ -35,6 +35,8 @@ import { parseReceipt, decodeFile, type Receipt } from "./lib/protocol"; import { completeLoginIfCallback } from "./lib/oauth"; import { describePaddockError, paddock, type Me, type PaddockDto, type Reachable, type Role, type TabPeopleDto } from "./api/paddock"; import { applyPrompt, bootstrapPrompt, reconcilePrompt, welcomePrompt, RECEIPT_PATH, WORK_ROOT } from "../shared/spec"; +import { Hold } from "./lib/hold"; +import { startTail } from "./lib/tail"; import { canPrompt, channelFor, findBox, holder, nextSlug, opsTab, OPS_SLUG, staleTabs, tabsOf, visibleTabs } from "../shared/tabs"; const STREAMS = ["acp", "stdout", "stderr", "stage"]; @@ -345,6 +347,12 @@ function Paddock({ /** Which paddock the boot effect has already run for. See the effect. */ const bootedRef = useRef(undefined); const client = useMemo(() => new FountainClient(`/f/${paddockId ?? "none"}`), [paddockId]); + /** + * The one pause every Fountain-bound loop here shares. A 429 from Fountain + * names how long to wait; the poll, the receipt, the scrollback and the + * stream all check this before going, so one refusal quiets all of them. + */ + const hold = useMemo(() => new Hold(), []); const role = me.role; const isOwner = role === "owner"; /** @@ -498,7 +506,7 @@ function Paddock({ // No agent filter and no identity gate: the proxy already returns exactly // this machine's tabs, to everybody in the paddock. const refreshConversations = useCallback(async () => { - if (!paddockId) return; + if (!paddockId || hold.active()) return; try { const next = await client.listConversations(); // Keep the array we have when nothing changed. Everything downstream — @@ -510,13 +518,14 @@ function Paddock({ // hour against production came from (2026-09-07). setConversations((cur) => (sameConversations(cur, next) ? cur : next)); } catch (err) { + hold.note(err); // A poll that misses is not worth interrupting anybody over — the next // one will do — but it is worth *saying*. An empty catch here hid two // bugs in a row, both of which looked like the app being stuck rather // than the app failing. console.error("paddock: could not list tabs —", describeError(err)); } - }, [client, paddockId]); + }, [client, paddockId, hold]); useEffect(() => { if (!paddockId) return; @@ -720,7 +729,7 @@ function Paddock({ /** The receipt: free to read, and it does not wake a parked box. */ const readReceipt = useCallback( async (opts: { woke?: boolean } = {}) => { - if (!boxId) return; + if (!boxId || hold.active()) return; if (opts.woke) receiptParked.current = false; else if (receiptParked.current) return; try { @@ -728,13 +737,14 @@ function Paddock({ receiptParked.current = false; setReceipt(parseReceipt(decodeFile(file))); } catch (err) { + hold.note(err); if (err instanceof ApiError && err.status === 409) receiptParked.current = true; setReceipt(null); // missing is the common case, and it is not an error } finally { setReceiptRead(true); } }, - [client, boxId], + [client, boxId, hold], ); useEffect(() => { @@ -786,19 +796,21 @@ function Paddock({ * between. Union by event id, oldest first, so calling this twice is free. */ const loadEvents = useCallback( - async (conversationId: string) => { + async (conversationId: string, after: number | null = null) => { + if (hold.active()) return; try { - const history = await client.listAllEvents(conversationId, STREAMS); + const history = await client.listAllEvents(conversationId, STREAMS, after); setEvents((m) => { const seen = new Set(history.map((e) => e.id)); const live = (m[conversationId] ?? []).filter((e) => !seen.has(e.id)); - return { ...m, [conversationId]: [...history, ...live].sort((a, b) => a.id - b.id) }; + return { ...m, [conversationId]: [...live, ...history].sort((a, b) => a.id - b.id) }; }); - } catch { + } catch (err) { + hold.note(err); /* the stream is the other half of this; a missed fetch is not fatal */ } }, - [client], + [client, hold], ); useEffect(() => { @@ -811,7 +823,7 @@ function Paddock({ /** * The newest event id seen on each tab's stream, kept across reconnects. * - * It used to be a local of the effect below, so every time the effect + * It used to be a local of the stream effect, so every time the effect * re-ran the stream came back with no `Last-Event-ID` and Fountain replayed * the tab's whole history — and every replayed turn boundary re-read the * receipt and the strip. Held here, a reconnect resumes where it left off. @@ -821,63 +833,44 @@ function Paddock({ // `tabsOf` builds fresh objects from every poll, and an effect keyed on one // of those reconnected on every poll. const activeId = active?.conversation.id ?? null; - - // The active tab's live tail. Phase 1 used Fountain's account-wide stream, - // which cannot be shared: it carries every conversation on the owner's key. + // What the tail calls back into, read at call time. The three callbacks are + // stable today, but a dependency on them is a promise that they stay so, and + // the cost of one of them changing identity is the loop this replaced — so + // the effect depends on none of them. + const latest = useRef({ refreshConversations, readReceipt, loadEvents }); + latest.current = { refreshConversations, readReceipt, loadEvents }; + + // The active tab's live tail (`lib/tail.ts`). Phase 1 used Fountain's + // account-wide stream, which cannot be shared: it carries every + // conversation on the owner's key. useEffect(() => { if (!activeId) return; const conversationId = activeId; - const ctrl = new AbortController(); - let stopped = false; - let backoff = 1000; - - const run = () => { - void client.streamConversation({ - conversationId, - lastEventId: lastEventIds.current[conversationId] ?? null, - streams: STREAMS, - signal: ctrl.signal, - onOpen: () => { - backoff = 1000; - // Whatever happened before this connection existed. The first turn - // of a brand-new machine starts the instant the conversation does, - // which is before any of this is listening — that turn was invisible - // until a reload, and so would anything missed by a dropped stream. - void loadEvents(conversationId); - }, - onMessage: (msg) => { - if (msg.id) lastEventIds.current[conversationId] = msg.id; - let ev: LogEvent; - try { - ev = JSON.parse(msg.data) as LogEvent; - } catch { - return; - } - if (msg.id) ev.id = Number(msg.id); - setEvents((m) => { - const list = m[conversationId] ?? []; - return list.some((e) => e.id === ev.id) ? m : { ...m, [conversationId]: [...list, ev] }; - }); - if (ev.kind === "stage" && ev.stage === "turn" && ev.state !== "started") { - void refreshConversations(); - // A turn just ended on this box, so it is awake: the one moment a - // parked machine's receipt is worth asking for again. - void readReceipt({ woke: true }); - } - }, - onClose: () => { - if (stopped) return; - window.setTimeout(run, backoff); - backoff = Math.min(backoff * 2, 15000); - }, - }); - }; - run(); - return () => { - stopped = true; - ctrl.abort(); - }; - }, [client, activeId, refreshConversations, readReceipt, loadEvents]); + const tail = startTail({ + conversationId, + client, + streams: STREAMS, + lastEventIds: lastEventIds.current, + hold, + // Whatever happened before this connection existed — or, on a reconnect, + // since the last event it saw. The first turn of a brand-new machine + // starts the instant the conversation does, which is before any of this + // is listening. + onOpen: (after) => void latest.current.loadEvents(conversationId, after), + onEvent: (ev) => + setEvents((m) => { + const list = m[conversationId] ?? []; + return list.some((e) => e.id === ev.id) ? m : { ...m, [conversationId]: [...list, ev] }; + }), + onTurnEnded: () => { + void latest.current.refreshConversations(); + // A turn just ended on this box, so it is awake: the one moment a + // parked machine's receipt is worth asking for again. + void latest.current.readReceipt({ woke: true }); + }, + }); + return () => tail.stop(); + }, [client, activeId, hold]); // The paddock's own channel: who is here, and when somebody else acts. useEffect(() => { diff --git a/apps/paddock/src/api/client.ts b/apps/paddock/src/api/client.ts index f160b4d..ef2ed92 100644 --- a/apps/paddock/src/api/client.ts +++ b/apps/paddock/src/api/client.ts @@ -35,7 +35,8 @@ import type { SecretKey, Vault, } from "./types"; -import { readSse, type SseMessage } from "../lib/sse"; +import { SseParser, type SseMessage } from "../lib/sse"; +import { BUILD_HEADER, buildHeaders, noteServerBuild } from "../lib/build"; export class ApiError extends Error { constructor( @@ -270,10 +271,14 @@ export class FountainClient { return this.json("POST", `/api/conversations/${conversationId}/read`); } - /** Every event of one tab on the given streams, oldest first, paging until drained. */ - async listAllEvents(conversationId: string, streams: string[]): Promise { + /** + * Every event of one tab on the given streams, oldest first, paging until + * drained — or, with `after`, only the ones newer than an id already held, + * which is what a reconnecting stream asks for so a reconnect costs the gap + * rather than the history. + */ + async listAllEvents(conversationId: string, streams: string[], after: number | null = null): Promise { const out: LogEvent[] = []; - let after: number | null = null; for (;;) { const qs = new URLSearchParams({ limit: "1000", streams: streams.join(","), blocks: "true" }); if (after !== null) qs.set("after", String(after)); @@ -320,7 +325,7 @@ export class FountainClient { * means parsing it. One connection per tab, scoped by the proxy to a tab * that is genuinely on this machine, is the honest version. */ - streamConversation(opts: { + async streamConversation(opts: { conversationId: string; lastEventId: string | null; streams: string[]; @@ -330,19 +335,56 @@ export class FountainClient { onClose: (err?: unknown) => void; }): Promise { const qs = new URLSearchParams({ streams: opts.streams.join(","), blocks: "true" }); - return readSse(`${this.baseUrl}/api/conversations/${encodeURIComponent(opts.conversationId)}/stream?${qs}`, { - lastEventId: opts.lastEventId, - signal: opts.signal, - onMessage: opts.onMessage, - onOpen: opts.onOpen, - onClose: opts.onClose, - }); + // The open is made here rather than by the suite's `readSse`, because a + // refusal has to reach the caller as an `ApiError` with its status and + // `Retry-After`: a 429 on the stream is the server saying when to come + // back, and a reader that reports it as "the stream closed" leaves the + // reconnect loop to guess. The read loop below is the suite's, verbatim. + const headers: Record = { accept: "text/event-stream", ...buildHeaders() }; + if (opts.lastEventId) headers["last-event-id"] = opts.lastEventId; + let res: Response; + try { + res = await fetch(`${this.baseUrl}/api/conversations/${encodeURIComponent(opts.conversationId)}/stream?${qs}`, { headers, signal: opts.signal }); + } catch (err) { + if (opts.signal.aborted) return; + opts.onClose(err); + return; + } + noteServerBuild(res.headers.get(BUILD_HEADER)); + if (!res.ok || !res.body) { + const ra = res.headers.get("retry-after"); + let code: string | null = null; + try { + const body = (await res.json()) as { error?: unknown }; + if (typeof body.error === "string") code = body.error; + } catch { + /* not JSON, or no body — the status is the message */ + } + opts.onClose(new ApiError(res.status, code, `stream ${res.status}`, ra ? Number(ra) : null)); + return; + } + + opts.onOpen?.(); + const reader = res.body.getReader(); + const decoder = new TextDecoder(); + const parser = new SseParser(); + try { + for (;;) { + const { value, done } = await reader.read(); + if (done) break; + const text = decoder.decode(value, { stream: true }).replace(/\r\n/g, "\n"); + for (const msg of parser.push(text)) opts.onMessage(msg); + } + if (!opts.signal.aborted) opts.onClose(); + } catch (err) { + if (!opts.signal.aborted) opts.onClose(err); + } } // ── plumbing ────────────────────────────────────────────────────────────── private async json(method: string, path: string, body?: unknown, signal?: AbortSignal): Promise { - const headers: Record = { accept: "application/json" }; + const headers: Record = { accept: "application/json", ...buildHeaders() }; if (body !== undefined) headers["content-type"] = "application/json"; const res = await fetch(`${this.baseUrl}${path}`, { method, @@ -350,6 +392,8 @@ export class FountainClient { body: body === undefined ? undefined : JSON.stringify(body), signal, }); + // A tab left open across a deploy learns about it here and reloads. + noteServerBuild(res.headers.get(BUILD_HEADER)); if (res.status === 204) return undefined as T; const text = await res.text(); let parsed: unknown = null; diff --git a/apps/paddock/src/api/paddock.ts b/apps/paddock/src/api/paddock.ts index b14fc2b..6c8f50e 100644 --- a/apps/paddock/src/api/paddock.ts +++ b/apps/paddock/src/api/paddock.ts @@ -5,6 +5,7 @@ * Same origin, cookie-authenticated, no key anywhere near the browser. */ import { readSse, type SseMessage } from "../lib/sse"; +import { BUILD_HEADER, buildHeaders, noteServerBuild } from "../lib/build"; import type { SkillHit } from "../lib/skills"; export type Role = "owner" | "member" | "guest"; @@ -108,7 +109,7 @@ export class PaddockError extends Error { } export const paddock = { - config: () => call<{ fountainUrl: string; anonymousStart: boolean }>("GET", "/api/config"), + config: () => call<{ fountainUrl: string; anonymousStart: boolean; buildId: string }>("GET", "/api/config"), me: () => call("GET", "/api/me"), /** @@ -182,9 +183,10 @@ export const paddock = { }; async function call(method: string, path: string, body?: unknown): Promise { - const headers: Record = { accept: "application/json" }; + const headers: Record = { accept: "application/json", ...buildHeaders() }; if (body !== undefined) headers["content-type"] = "application/json"; const res = await fetch(path, { method, headers, body: body === undefined ? undefined : JSON.stringify(body) }); + noteServerBuild(res.headers.get(BUILD_HEADER)); if (res.status === 204) return undefined as T; const text = await res.text(); let parsed: unknown = null; diff --git a/apps/paddock/src/lib/build.test.ts b/apps/paddock/src/lib/build.test.ts new file mode 100644 index 0000000..a45bda8 --- /dev/null +++ b/apps/paddock/src/lib/build.test.ts @@ -0,0 +1,36 @@ +import { afterEach, expect, test } from "bun:test"; +import { buildHeaders, myBuild, noteServerBuild, resetBuild } from "./build"; + +function servedWith(build: string | null) { + (globalThis as { document?: unknown }).document = { + querySelector: () => (build === null ? null : { getAttribute: () => build }), + }; + resetBuild(); +} + +afterEach(() => { + delete (globalThis as { document?: unknown }).document; + resetBuild(); +}); + +test("a tab that knows its build reloads on the first mismatch, once", () => { + servedWith("abc"); + let reloads = 0; + const reload = () => (reloads += 1); + expect(myBuild()).toBe("abc"); + expect(buildHeaders()).toEqual({ "x-paddock-build": "abc" }); + expect(noteServerBuild("abc", reload)).toBe(false); + expect(noteServerBuild(null, reload)).toBe(false); + expect(noteServerBuild("def", reload)).toBe(true); + expect(noteServerBuild("def", reload)).toBe(false); + expect(reloads).toBe(1); +}); + +test("a tab with no build stamp never reloads over one", () => { + servedWith(null); + let reloads = 0; + expect(myBuild()).toBeNull(); + expect(buildHeaders()).toEqual({}); + expect(noteServerBuild("def", () => (reloads += 1))).toBe(false); + expect(reloads).toBe(0); +}); diff --git a/apps/paddock/src/lib/build.ts b/apps/paddock/src/lib/build.ts new file mode 100644 index 0000000..fb3b898 --- /dev/null +++ b/apps/paddock/src/lib/build.ts @@ -0,0 +1,59 @@ +/** + * Which build this tab is running, and what to do when the server is not. + * + * A browser tab left open across a deploy keeps running the bundle it loaded, + * and a bug fixed in the new one goes on running in the old — which is how a + * client fix to the stream loop (paddock #67) left an already-open tab + * spinning, faster than before, against production. Nobody finds those tabs. + * + * So the server stamps every response with its build id (`x-paddock-build`) + * and writes the same id into the HTML it serves (``). + * The bundle reads its own from the meta tag; every API answer is compared; + * the first mismatch reloads the page. The meta tag rather than the first API + * answer, because a deploy between the HTML and that first call would + * otherwise teach an old bundle the new id and it would never reload. + * + * `null` means "not stamped" — the vite dev server serves no meta tag and the + * mock server sends no header — and a tab that does not know its build never + * reloads over it. + */ + +export const BUILD_HEADER = "x-paddock-build"; + +let mine: string | null | undefined; +let reloading = false; + +/** The build id the HTML was served with, or null when it carries none. */ +export function myBuild(): string | null { + if (mine === undefined) { + mine = typeof document === "undefined" ? null : (document.querySelector('meta[name="paddock-build"]')?.getAttribute("content") ?? null); + } + return mine; +} + +/** + * The server said which build it is. Reload if it is not ours. + * + * Idempotent: the first mismatch schedules exactly one reload, and everything + * after it — the answers still landing from calls already in flight — is + * ignored. + */ +export function noteServerBuild(theirs: string | null, reload: () => void = () => window.location.reload()): boolean { + const ours = myBuild(); + if (!ours || !theirs || theirs === ours || reloading) return false; + reloading = true; + reload(); + return true; +} + +/** Headers to send so the server knows which build is asking. */ +export function buildHeaders(): Record { + const ours = myBuild(); + return ours ? { [BUILD_HEADER]: ours } : {}; +} + +/** For tests. */ +export function resetBuild(): void { + mine = undefined; + reloading = false; +} diff --git a/apps/paddock/src/lib/hold.test.ts b/apps/paddock/src/lib/hold.test.ts new file mode 100644 index 0000000..cf9940b --- /dev/null +++ b/apps/paddock/src/lib/hold.test.ts @@ -0,0 +1,30 @@ +import { expect, test } from "bun:test"; +import { ApiError } from "../api/client"; +import { DEFAULT_RETRY_MS, Hold } from "./hold"; + +test("only a 429 holds anything", () => { + let now = 0; + const hold = new Hold(() => now, () => 0); + hold.note(new ApiError(500, null, "boom")); + hold.note(new Error("network")); + expect(hold.active()).toBe(false); + hold.note(new ApiError(429, "rate_limited", "slow down", 35)); + expect(hold.active()).toBe(true); + expect(hold.remainingMs()).toBe(35_000); + now = 35_000; + expect(hold.active()).toBe(false); + expect(hold.remainingMs()).toBe(0); +}); + +test("a 429 with no Retry-After still holds, for a default", () => { + const hold = new Hold(() => 0, () => 0); + hold.note(new ApiError(429, null, "slow down")); + expect(hold.remainingMs()).toBe(DEFAULT_RETRY_MS); +}); + +test("a shorter ask never cuts a longer hold short, and jitter only adds", () => { + const hold = new Hold(() => 0, () => 1); + hold.note(new ApiError(429, null, "x", 30)); + hold.note(new ApiError(429, null, "x", 5)); + expect(hold.remainingMs()).toBe(32_000); +}); diff --git a/apps/paddock/src/lib/hold.ts b/apps/paddock/src/lib/hold.ts new file mode 100644 index 0000000..b0e285b --- /dev/null +++ b/apps/paddock/src/lib/hold.ts @@ -0,0 +1,45 @@ +/** + * "Not now": the pause Fountain asked for. + * + * A 429 carries `Retry-After`, and every caller that ignores it is a caller + * that retries on its own cadence into a server that just said it was full. + * The poll, the receipt read, the scrollback fetch and the stream reconnect + * all share one of these, so one refusal quiets all of them at once — the + * server is rate-limiting the *key*, not the call, and four loops each + * honouring their own copy of the answer would still be four loops. + * + * Jitter so a room full of browsers told the same thing does not come back in + * the same instant. + */ +import { ApiError } from "../api/client"; + +/** What a 429 with no `Retry-After` is taken to mean. */ +export const DEFAULT_RETRY_MS = 15_000; +/** Added on top of what the server asked, so retries spread out. */ +export const JITTER_MS = 2_000; + +export class Hold { + private until = 0; + + constructor( + private readonly now: () => number = Date.now, + private readonly random: () => number = Math.random, + ) {} + + /** Take note of a failure; only a 429 changes anything. */ + note(err: unknown): void { + if (!(err instanceof ApiError) || err.status !== 429) return; + const asked = err.retryAfter !== null && Number.isFinite(err.retryAfter) && err.retryAfter > 0 ? err.retryAfter * 1000 : DEFAULT_RETRY_MS; + this.until = Math.max(this.until, this.now() + asked + this.random() * JITTER_MS); + } + + /** Whether a call should wait rather than go. */ + active(): boolean { + return this.until > this.now(); + } + + /** How long a caller should wait, at least; 0 when it may go now. */ + remainingMs(): number { + return Math.max(0, this.until - this.now()); + } +} diff --git a/apps/paddock/src/lib/tail.test.ts b/apps/paddock/src/lib/tail.test.ts new file mode 100644 index 0000000..ba32860 --- /dev/null +++ b/apps/paddock/src/lib/tail.test.ts @@ -0,0 +1,129 @@ +import { describe, expect, test } from "bun:test"; +import { ApiError } from "../api/client"; +import { Hold } from "./hold"; +import type { SseMessage } from "./sse"; +import { BASE_BACKOFF_MS, startTail, type TailClient } from "./tail"; + +/** + * A stream the test drives by hand: `open()` answers 200, `emit()` sends a + * frame, `close()` ends it. Each `streamConversation` call is one open attempt. + */ +function fakeClient() { + const attempts: { lastEventId: string | null; onMessage: (m: SseMessage) => void; onOpen?: () => void; onClose: (e?: unknown) => void }[] = []; + const client: TailClient = { + streamConversation(opts) { + attempts.push({ lastEventId: opts.lastEventId, onMessage: opts.onMessage, onOpen: opts.onOpen, onClose: opts.onClose }); + return new Promise(() => undefined); + }, + }; + return { client, attempts, last: () => attempts[attempts.length - 1]! }; +} + +const stage = (id: number, state: string): SseMessage => ({ id: String(id), event: "stage", data: JSON.stringify({ kind: "stage", stage: "turn", state }) }); + +function tail(opts: Partial[0]> = {}) { + const f = fakeClient(); + const scheduled: number[] = []; + const calls = { open: [] as (number | null)[], events: 0, turnsEnded: 0 }; + const lastEventIds: Record = {}; + const t = startTail({ + conversationId: "c1", + client: f.client, + streams: ["events"], + lastEventIds, + hold: new Hold(() => 0, () => 0), + onOpen: (after) => calls.open.push(after), + onEvent: () => (calls.events += 1), + onTurnEnded: () => (calls.turnsEnded += 1), + schedule: (_fn, ms) => scheduled.push(ms), + random: () => 0, + ...opts, + }); + return { ...f, t, scheduled, calls, lastEventIds }; +} + +describe("the live tail", () => { + test("a turn ending on the stream does not reopen it, and is reported once", () => { + const x = tail(); + x.last().onOpen!(); + x.last().onMessage(stage(41, "started")); + x.last().onMessage(stage(42, "completed")); + expect(x.attempts).toHaveLength(1); + expect(x.t.opens).toBe(1); + expect(x.calls.turnsEnded).toBe(1); + expect(x.calls.events).toBe(2); + expect(x.scheduled).toEqual([]); + x.t.stop(); + }); + + test("a closed stream is retried after the backoff, not at once", () => { + const x = tail(); + x.last().onOpen!(); + expect(x.calls.open).toEqual([null]); + x.last().onMessage(stage(7, "completed")); + x.last().onClose(); + expect(x.scheduled).toEqual([BASE_BACKOFF_MS]); + expect(x.lastEventIds.c1).toBe("7"); + x.t.stop(); + }); + + test("resume: the second open carries the id the first one saw", () => { + const fns: (() => void)[] = []; + const x = tail({ schedule: (fn) => fns.push(fn) }); + x.last().onOpen!(); + x.last().onMessage(stage(7, "completed")); + x.last().onClose(); + fns.shift()!(); + expect(x.attempts).toHaveLength(2); + expect(x.last().lastEventId).toBe("7"); + x.last().onOpen!(); + expect(x.calls.open).toEqual([null, 7]); + x.t.stop(); + }); + + test("the backoff grows across failed opens and resets only on a real open", () => { + const fns: (() => void)[] = []; + const delays: number[] = []; + const x = tail({ + schedule: (fn, ms) => { + fns.push(fn); + delays.push(ms); + }, + }); + x.last().onClose(new Error("stream 502")); + fns.shift()!(); + x.last().onClose(new Error("stream 502")); + fns.shift()!(); + x.last().onClose(new Error("stream 502")); + expect(delays).toEqual([1000, 2000, 4000]); + fns.shift()!(); + x.last().onOpen!(); + x.last().onClose(); + expect(delays[3]).toBe(1000); + x.t.stop(); + }); + + test("a 429 waits what the server asked, not the backoff", () => { + const fns: (() => void)[] = []; + const delays: number[] = []; + const x = tail({ + hold: new Hold(() => 0, () => 0), + schedule: (fn, ms) => { + fns.push(fn); + delays.push(ms); + }, + }); + x.last().onClose(new ApiError(429, "rate_limited", "slow down", 35)); + expect(delays).toEqual([35_000]); + x.t.stop(); + }); + + test("stopped means stopped: nothing is scheduled after stop", () => { + const x = tail(); + x.t.stop(); + x.last().onClose(); + x.last().onMessage(stage(1, "completed")); + expect(x.scheduled).toEqual([]); + expect(x.calls.turnsEnded).toBe(0); + }); +}); diff --git a/apps/paddock/src/lib/tail.ts b/apps/paddock/src/lib/tail.ts new file mode 100644 index 0000000..8ed7d43 --- /dev/null +++ b/apps/paddock/src/lib/tail.ts @@ -0,0 +1,132 @@ +/** + * One tab's live tail, as a state machine rather than an effect. + * + * This used to be the body of a `useEffect` in `App.tsx`, and two of its + * variables — the newest event id and the reconnect backoff — were locals of + * that effect. The effect was keyed on the active `Tab` *object*, which the + * strip poll rebuilt every four seconds, so every poll re-ran it: the stream + * was reopened with no `Last-Event-ID` (the whole history replayed, and every + * replayed turn boundary re-read the receipt and the strip) and the backoff + * went back to a second, so a 429 from Fountain never slowed anything down. + * At its worst one tab was opening the stream nine times a second. + * + * Kept out of React so the rules are plain and testable: + * + * - **Resume, do not replay.** `Last-Event-ID` is whatever this tail last + * saw for the conversation, and the catch-up read on open asks for events + * *after* it — a reconnect costs the gap, not the history. + * - **Back off only forward.** The backoff resets when a stream actually + * opens (`onOpen` fires only after a 200), never when one is retried. + * - **A 429 is the schedule.** `Retry-After` (plus jitter, via the shared + * `Hold`) is how long to wait, however short the backoff was. + * - **One tail per conversation id.** `start` is called once per id and + * `stop` once; nothing about state changes in between reaches here. + */ +import { ApiError } from "../api/client"; +import type { SseMessage } from "./sse"; +import type { LogEvent } from "../api/types"; +import { Hold } from "./hold"; + +export interface TailClient { + streamConversation(opts: { + conversationId: string; + lastEventId: string | null; + streams: string[]; + signal: AbortSignal; + onMessage: (msg: SseMessage) => void; + onOpen?: () => void; + onClose: (err?: unknown) => void; + }): Promise; +} + +export interface TailOptions { + conversationId: string; + client: TailClient; + streams: string[]; + /** Newest event id seen per conversation — shared across tails so a re-open resumes. */ + lastEventIds: Record; + /** The shared 429 pause. */ + hold: Hold; + /** A stream is open; `after` is the newest id already held, for the catch-up read. */ + onOpen: (after: number | null) => void; + onEvent: (ev: LogEvent) => void; + /** A turn on this tab ended — the box is awake, the strip has moved. */ + onTurnEnded: () => void; + /** Injectable for tests. */ + schedule?: (fn: () => void, ms: number) => unknown; + random?: () => number; +} + +/** The first retry, after a stream closes; doubles up to `MAX_BACKOFF_MS`. */ +export const BASE_BACKOFF_MS = 1_000; +export const MAX_BACKOFF_MS = 15_000; + +export interface Tail { + stop(): void; + /** For tests and the panel: how many times a stream was opened (attempted). */ + readonly opens: number; +} + +export function startTail(opts: TailOptions): Tail { + const ctrl = new AbortController(); + const schedule = opts.schedule ?? ((fn, ms) => window.setTimeout(fn, ms)); + const random = opts.random ?? Math.random; + const { conversationId } = opts; + let stopped = false; + let backoff = BASE_BACKOFF_MS; + let opens = 0; + + const run = () => { + if (stopped) return; + opens += 1; + const last = opts.lastEventIds[conversationId] ?? null; + void opts.client.streamConversation({ + conversationId, + lastEventId: last, + streams: opts.streams, + signal: ctrl.signal, + onOpen: () => { + if (stopped) return; + // A real open — the reader fires this only after a 200 — is the one + // thing that earns a reset. + backoff = BASE_BACKOFF_MS; + const after = last !== null ? Number(last) : null; + opts.onOpen(Number.isFinite(after) ? after : null); + }, + onMessage: (msg) => { + if (stopped) return; + if (msg.id) opts.lastEventIds[conversationId] = msg.id; + let ev: LogEvent; + try { + ev = JSON.parse(msg.data) as LogEvent; + } catch { + return; + } + if (msg.id) ev.id = Number(msg.id); + opts.onEvent(ev); + if (ev.kind === "stage" && ev.stage === "turn" && ev.state !== "started") opts.onTurnEnded(); + }, + onClose: (err) => { + if (stopped) return; + opts.hold.note(err); + // Whatever the server asked for wins over our own cadence; otherwise + // wait the backoff, and grow it for next time. Jitter either way. + const own = err instanceof ApiError && err.status === 429 ? 0 : backoff; + const wait = Math.max(opts.hold.remainingMs(), own) + random() * 500; + backoff = Math.min(backoff * 2, MAX_BACKOFF_MS); + schedule(run, wait); + }, + }); + }; + run(); + + return { + stop() { + stopped = true; + ctrl.abort(); + }, + get opens() { + return opens; + }, + }; +}