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