fix(boat): coalesce concurrent readiness requests

server/src/modules/computers/adapters/boat.test.ts

@@ -46,6 +46,146 @@ const json = (value: unknown, status = 200, headers?: Record<string, string>) =>
     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<void>(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<void>(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<void>(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<void>(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<void>(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);

server/src/modules/computers/adapters/boat.ts

@@ -238,6 +238,10 @@ export function boatBackend(
   resolveKey: (record: ComputerRecord) => Promise<string>,
   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<void> };
+  const readiness = new Map<string, Readiness>();
   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<any> {
+    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<string, string>)[`${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<void>((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(