From 72042f165cbc5bfd10d3b4d3d8bcf9cced2dfcdc Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 11 Oct 2026 04:03:52 -0500 Subject: [PATCH] fix(boat): coalesce concurrent readiness requests --- .../modules/computers/adapters/boat.test.ts | 140 ++++++++++++++++++ server/src/modules/computers/adapters/boat.ts | 98 ++++++++---- 2 files changed, 211 insertions(+), 27 deletions(-) diff --git a/server/src/modules/computers/adapters/boat.test.ts b/server/src/modules/computers/adapters/boat.test.ts index 365a6d813b..f2cd70e153 100644 --- a/server/src/modules/computers/adapters/boat.test.ts +++ b/server/src/modules/computers/adapters/boat.test.ts @@ -46,6 +46,146 @@ const json = (value: unknown, status = 200, headers?: Record) => headers: { "content-type": "application/json", ...headers }, }); describe("Boat transport validation", () => { + it("shares concurrent readiness without issuing a conflicting second resume", async () => { + let resumes = 0; + let resumed = false; + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const fetcher = vi.fn(async (_url: any, options?: any) => { + if (options?.method === "POST") { + if (++resumes > 1) return json({ error: { code: "private-conflict-canary" } }, 409); + await gate; + resumed = true; + return json({ ok: true }); + } + return json({ sandbox: { id: record.providerId, state: resumed ? "idle" : "archived", snapshots: true, stop: null } }); + }); + const backend = boatBackend(async () => "secret", fetcher); + const jobs = Promise.allSettled([backend.ready(record), backend.ready(structuredClone(record))]); + try { + await vi.waitFor(() => expect(resumes).toBeGreaterThan(0)); + } finally { release(); } + expect((await jobs).map(result => result.status)).toEqual(["fulfilled", "fulfilled"]); + expect(resumes).toBe(1); + }); + + it("expires a short readiness waiter without cancelling its sibling", async () => { + vi.useFakeTimers(); + let resumed = false; + let release!: () => void; + let signal: AbortSignal | undefined; + const gate = new Promise(resolve => { release = resolve; }); + const fetcher = vi.fn(async (_url: any, options?: any) => { + if (options?.method === "POST") { signal = options.signal; await gate; resumed = true; return json({ ok: true }); } + return json({ sandbox: { id: record.providerId, state: resumed ? "idle" : "archived", snapshots: true, stop: null } }); + }); + const backend = boatBackend(async () => "secret", fetcher); + try { + const short = backend.ready(record, { deadlineMs: Date.now() + 20 }); + const shortResult = expect(short).rejects.toThrow("readiness timed out"); + const long = backend.ready(record, { deadlineMs: Date.now() + 1000 }); + await vi.advanceTimersByTimeAsync(21); + await shortResult; + expect(signal).toBeDefined(); + expect(signal!.aborted).toBe(false); + release(); + await long; + expect(fetcher.mock.calls.filter(([, options]) => options?.method === "POST")).toHaveLength(1); + } finally { release(); vi.useRealTimers(); } + }); + it("does not resume when every waiter expired during inspection", async () => { + vi.useFakeTimers(); + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const fetcher = vi.fn(async () => { + await gate; + return json({ sandbox: { id: record.providerId, state: "archived", snapshots: true, stop: null } }); + }); + const backend = boatBackend(async () => "secret", fetcher); + try { + const failure = expect(backend.ready(record, { deadlineMs: Date.now() + 20 })).rejects.toThrow("timed out"); + await vi.advanceTimersByTimeAsync(21); + await failure; + release(); + await vi.advanceTimersByTimeAsync(0); + expect(fetcher).toHaveBeenCalledOnce(); + fetcher.mockImplementation(async () => json({ sandbox: { id: record.providerId, state: "idle", snapshots: true } })); + await backend.ready(record); + expect(fetcher).toHaveBeenCalledTimes(2); + } finally { release(); vi.useRealTimers(); } + }); + it("does not dispatch resume after its last caller expires during secret resolution", async () => { + vi.useFakeTimers(); + let release!: () => void; + let keyCalls = 0; + const gate = new Promise(resolve => { release = resolve; }); + const fetcher = vi.fn(async () => json({ sandbox: { id: record.providerId, state: "archived", snapshots: true } })); + const backend = boatBackend(async () => { if (++keyCalls === 2) await gate; return "secret"; }, fetcher); + try { + const failure = expect(backend.ready(record, { deadlineMs: Date.now() + 20 })).rejects.toThrow("timed out"); + await vi.advanceTimersByTimeAsync(0); + expect(keyCalls).toBe(2); + await vi.advanceTimersByTimeAsync(21); + await failure; + release(); + await vi.advanceTimersByTimeAsync(0); + expect(fetcher).toHaveBeenCalledOnce(); // Initial inspection only, no late resume. + } finally { release(); vi.useRealTimers(); } + }); + it.each(["company", "environment", "computer", "provider", "controller", "secret", "version"])("does not share readiness across a different %s identity", async field => { + const other = structuredClone(record); + if (field === "company") other.companyId = "other"; + if (field === "environment") other.environmentId = "other"; + if (field === "computer") other.id = "other"; + if (field === "provider") other.providerId = "bx_other"; + if (field === "controller") other.ledger.controllerId = "other"; + if (field === "secret") other.ledger.secretRef.secretId = "other"; + if (field === "version") other.ledger.secretRef.version = 2; + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const fetcher = vi.fn(async (url: any) => { + await gate; + return json({ sandbox: { id: String(url).split("/").at(-1), state: "idle", snapshots: true } }); + }); + const backend = boatBackend(async () => "secret", fetcher); + const jobs = Promise.all([backend.ready(record), backend.ready(other)]); + release(); + await jobs; + expect(fetcher).toHaveBeenCalledTimes(2); + }); + it("evicts completed readiness and propagates409 without retry or private response data", async () => { + let ready = true; + const fetcher = vi.fn(async (_url: any, options?: any) => options?.method === "POST" + ? json({ message: "private-message-canary", error: { code: "private-code-canary" }, url: "private-url-canary" }, 409) + : json({ sandbox: { id: record.providerId, state: ready ? "idle" : "archived", snapshots: true } })); + const backend = boatBackend(async () => "private-key-canary", fetcher); + await backend.ready(record); + ready = false; + await expect(backend.ready(record)).rejects.toThrow("Boat request failed (409) [operation=resume]"); + expect(fetcher.mock.calls.filter(([, options]) => options?.method === "POST")).toHaveLength(1); + ready = true; + await backend.ready(record); // A failed attempt is evicted too. + expect(fetcher).toHaveBeenCalledTimes(4); + const unavailable = boatBackend(async () => "private-key-canary", vi.fn(async () => json({ message: "private-message-canary" }, 409))); + await expect(unavailable.inspect(record)).rejects.toThrow("Boat request failed (409) [operation=inspect]"); + }); + it.each(["pending", "failing", "snapshots", "identity"])("rejects invalid %s state during resume polling", async invalid => { + let resumed = false; + const fetcher = vi.fn(async (_url: any, options?: any) => { + if (options?.method === "POST") { resumed = true; return json({ ok: true }); } + return json({ sandbox: { id: resumed && invalid === "identity" ? "bx_other" : record.providerId, + state: resumed ? "idle" : "archived", snapshots: !(resumed && invalid === "snapshots"), + stop: resumed && ["pending", "failing"].includes(invalid) ? { id: "stop", status: invalid } : null } }); + }); + await expect(boatBackend(async () => "secret", fetcher).ready(record)).rejects.toThrow( + invalid === "identity" ? "different computer" : invalid === "snapshots" ? "snapshots must be enabled" : "saving a previous stop"); + expect(fetcher.mock.calls.filter(([, options]) => options?.method === "POST")).toHaveLength(1); + }); + it.each(["pending", "failing"])("never resumes a %s previous stop", async status => { + const fetcher = vi.fn(async () => json({ sandbox: { id: record.providerId, state: "archived", snapshots: true, stop: { id: "stop", status } } })); + await expect(boatBackend(async () => "secret", fetcher).ready(record)).rejects.toThrow("saving a previous stop"); + expect(fetcher).toHaveBeenCalledOnce(); + }); it("uses only the file budget remaining after transport preparation", async () => { let clock = 1_000_000; const time = vi.spyOn(Date, "now").mockImplementation(() => clock); diff --git a/server/src/modules/computers/adapters/boat.ts b/server/src/modules/computers/adapters/boat.ts index 6e74fff226..eddda3aa20 100644 --- a/server/src/modules/computers/adapters/boat.ts +++ b/server/src/modules/computers/adapters/boat.ts @@ -238,6 +238,10 @@ export function boatBackend( resolveKey: (record: ComputerRecord) => Promise, fetcher: typeof fetch = fetch, ): ComputerBackend { + // Coalesce provider readiness only; each admitted owner still claims and + // validates its own execution authority in the application service. + type Readiness = { deadlineMs: number; waiters: Set<{ deadlineMs: number }>; promise: Promise }; + const readiness = new Map(); function remaining(options: ComputerOperationOptions | undefined, limit: number) { const ms = options?.deadlineMs === undefined ? limit : Math.min(limit, options.deadlineMs - Date.now()); if (ms <= 0) throw new ComputerError("provider_error", "Computer file operation timed out"); @@ -249,31 +253,40 @@ export function boatBackend( suffix = "", body?: unknown, options?: ComputerOperationOptions, + checkDispatch?: () => void, ): Promise { + const key = await resolveKey(record); + checkDispatch?.(); const response = await fetcher( `${apiOrigin}/sandboxes/${segment(record.providerId)}${suffix}`, { method, headers: { - authorization: `Bearer ${await resolveKey(record)}`, + authorization: `Bearer ${key}`, "content-type": "application/json", }, ...(body === undefined ? {} : { body: JSON.stringify(body) }), signal: AbortSignal.timeout(remaining(options, 90_000)), }, ); - if (!response.ok) - throw new ComputerError( - "provider_error", - `Boat request failed (${response.status})`, - ); + if (!response.ok) { + // Closed operation labels identify the failing endpoint without exposing + // the provider response, sandbox URL, stop ID, or credential material. + const operation = ({ + "GET ": "inspect", "PATCH ": "renew", "POST /resume": "resume", + "POST /sshkey": "ssh_key", "POST /host": "host", "POST /stop": "stop", + "POST /desktop": "desktop", + } as Record)[`${method} ${suffix}`] + ?? (method === "GET" && suffix.startsWith("/stops/") ? "stop_status" : "unknown"); + throw new ComputerError("provider_error", `Boat request failed (${response.status}) [operation=${operation}]`); + } const data = await response.json(); if (!data || typeof data !== "object") throw new ComputerError("provider_error", "Invalid Boat response"); return data; } - async function inspect(record: ComputerRecord, options?: ComputerOperationOptions) { - const data = await api(record, "GET", "", undefined, options); + async function inspect(record: ComputerRecord, options?: ComputerOperationOptions, checkDispatch?: () => void) { + const data = await api(record, "GET", "", undefined, options, checkDispatch); const sandbox = data.sandbox ?? data; if (sandbox.id !== record.providerId || typeof sandbox.state !== "string") throw new ComputerError( @@ -501,26 +514,57 @@ export function boatBackend( inspect, runner, async ready(record, options) { - const state = await inspect(record, options); - if (!state.snapshots) - throw new ComputerError("invalid", "Boat snapshots must be enabled"); - if (state.stop && ["pending", "failing"].includes(state.stop.status)) - throw new ComputerError( - "conflict", - "Boat is still saving a previous stop", - ); - if (!["ready", "idle", "running"].includes(state.state)) { - await api(record, "POST", "/resume", { ttlSeconds: 300 }, options); - transportCache.delete( - `${record.id}:${record.ledger.secretRef.secretId}:${record.ledger.secretRef.version ?? "latest"}`, - ); - for (let i = 0; i < 90; i++) { - const status = await inspect(record, options); - if (["ready", "idle", "running"].includes(status.state)) return; - await new Promise((r) => setTimeout(r, remaining(options, 1000))); - } - throw new ComputerError("provider_error", "Boat did not become ready"); + const waiter = { deadlineMs: Date.now() + remaining(options, 120_000) }; + const key = JSON.stringify([record.companyId, record.environmentId, record.id, record.providerId, + record.ledger.controllerId, record.ledger.secretRef.secretId, record.ledger.secretRef.version ?? "latest"]); + let pending = readiness.get(key); + if (!pending) { + const entry: Readiness = { deadlineMs: Date.now() + 120_000, waiters: new Set(), promise: Promise.resolve() }; + readiness.set(key, entry); + pending = entry; + const budget = () => { + // An in-flight request may finish after its last waiter times out, but + // cannot initiate another request without a still-live caller. + if (![...entry.waiters].some(value => value.deadlineMs > Date.now())) + throw new ComputerError("provider_error", "Computer readiness timed out"); + remaining({ deadlineMs: entry.deadlineMs }, 120_000); + return { deadlineMs: entry.deadlineMs }; + }; + entry.promise = Promise.resolve().then(async () => { + const checked = async () => { + const state = await inspect(record, budget(), budget); + if (!state.snapshots) throw new ComputerError("invalid", "Boat snapshots must be enabled"); + if (state.stop && ["pending", "failing"].includes(state.stop.status)) + throw new ComputerError("conflict", "Boat is still saving a previous stop"); + return ["ready", "idle", "running"].includes(state.state); + }; + if (await checked()) return; + await api(record, "POST", "/resume", { ttlSeconds: 300 }, budget(), budget); + transportCache.delete( + `${record.id}:${record.ledger.secretRef.secretId}:${record.ledger.secretRef.version ?? "latest"}`, + ); + for (let i = 0; i < 90; i++) { + if (await checked()) return; + await new Promise(resolve => setTimeout(resolve, remaining(budget(), 1000))); + } + throw new ComputerError("provider_error", "Boat did not become ready"); + }); + const clear = () => { if (readiness.get(key) === entry) readiness.delete(key); }; + void entry.promise.then(clear, clear); } + waiter.deadlineMs = Math.min(waiter.deadlineMs, pending.deadlineMs); + pending.waiters.add(waiter); + const entry = pending; + await new Promise((resolve, reject) => { + const cleanup = () => { clearTimeout(timer); entry.waiters.delete(waiter); }; + const expired = () => new ComputerError("provider_error", "Computer readiness timed out"); + const timer = setTimeout(() => { cleanup(); reject(expired()); }, Math.max(1, waiter.deadlineMs - Date.now())); + entry.promise.then(() => { + cleanup(); + if (Date.now() >= waiter.deadlineMs) reject(expired()); + else resolve(); + }, error => { cleanup(); reject(error); }); + }); }, async claim(record, options) { await execute(