diff --git a/server/src/modules/computers/adapters/boat.test.ts b/server/src/modules/computers/adapters/boat.test.ts index 13e8529854..00e65b5aff 100644 --- a/server/src/modules/computers/adapters/boat.test.ts +++ b/server/src/modules/computers/adapters/boat.test.ts @@ -16,7 +16,7 @@ import { join } from "node:path"; import { spawnSync } from "node:child_process"; import { remoteProgram } from "./remote-program.js"; import { afterEach, describe, it, expect, vi } from "vitest"; -import { boatBackend, desktopReadinessProgram, processProgram } from "./boat.js"; +import { boatBackend, createBoatTransportAdmission, desktopReadinessProgram, processProgram } from "./boat.js"; import { buildGitAuthInvocation } from "../../../services/git-credentials.js"; import type { ComputerRecord } from "../domain/ledger.js"; const record: ComputerRecord = { @@ -480,3 +480,55 @@ describe("Boat desktop input readiness", () => { expect(existsSync(f.healthy)).toBe(true); }); }); + + +describe("Boat SSH admission", () => { + const result = { exitCode: 0, signal: null, timedOut: false, stdout: "", stderr: "", pid: null, startedAt: "" }; + it("reserves lifecycle and identity capacity while bulk and long legacy commands are active", async () => { + const admission = createBoatTransportAdmission(); + const completions: Array<() => void> = []; + const runner = { execute: vi.fn(async () => new Promise((resolve) => completions.push(() => resolve(result)))) }; + const input = { command: "legacy", timeoutMs: 60_000 }; + const jobs = Array.from({ length: 8 }, () => admission(runner, { ...input, stdin: "x".repeat(1024 * 1024 + 1) }, true)); + jobs.push(admission(runner, input, false), admission(runner, input, false)); + await Promise.resolve(); + expect(runner.execute).toHaveBeenCalledTimes(6); + jobs.push(admission(runner, { command: "retire", timeoutMs: 10_000 }, true)); + jobs.push(admission(runner, { command: "identity", timeoutMs: 10_000 }, true)); + await Promise.resolve(); + expect(runner.execute).toHaveBeenCalledTimes(8); + while (completions.length) { + completions.splice(0).forEach((finish) => finish()); + await new Promise((resolve) => setTimeout(resolve, 0)); + } + await Promise.all(jobs); + expect(runner.execute).toHaveBeenCalledTimes(12); + }); + it("expires queued work without starting it and preserves the original remaining timeout", async () => { + vi.useFakeTimers(); + try { + const admission = createBoatTransportAdmission(); + const completions: Array<() => void> = []; + const runner = { execute: vi.fn(async (_input: { command: string; timeoutMs?: number }) => new Promise((resolve) => completions.push(() => resolve(result)))) }; + const jobs = Array.from({ length: 6 }, () => admission(runner, { command: "long", timeoutMs: 60_000 }, false)); + await vi.advanceTimersByTimeAsync(0); + const expired = admission(runner, { command: "expired", timeoutMs: 100 }, false); + const queued = admission(runner, { command: "queued", timeoutMs: 1000 }, false); + await vi.advanceTimersByTimeAsync(200); + expect(await expired).toMatchObject({ timedOut: true, pid: null }); + expect(runner.execute).toHaveBeenCalledTimes(6); + completions.shift()!(); + await vi.advanceTimersByTimeAsync(0); + expect(runner.execute).toHaveBeenLastCalledWith(expect.objectContaining({ command: "queued", timeoutMs: 800 })); + completions.splice(0).forEach((finish) => finish()); + await Promise.all([...jobs, queued]); + } finally { vi.useRealTimers(); } + }); + it("releases all capacity after failed or cancelled transport commands without replay", async () => { + const admission = createBoatTransportAdmission(); + const runner = { execute: vi.fn(async () => { throw new Error("transport cancelled"); }) }; + const settled = await Promise.allSettled(Array.from({ length: 12 }, () => admission(runner, { command: "once", timeoutMs: 1000 }, false))); + expect(settled.every((entry) => entry.status === "rejected")).toBe(true); + expect(runner.execute).toHaveBeenCalledTimes(12); + }); +}); diff --git a/server/src/modules/computers/adapters/boat.ts b/server/src/modules/computers/adapters/boat.ts index bf3af1f9b1..bf35376117 100644 --- a/server/src/modules/computers/adapters/boat.ts +++ b/server/src/modules/computers/adapters/boat.ts @@ -15,6 +15,66 @@ import { remoteProgram } from "./remote-program.js"; const apiOrigin = "https://boat.dev/api/v1"; const transportCache = new Map>(); +type CommandInput = Parameters[0]; + +/** Reserve SSH admission for lifecycle/identity calls while bulk and legacy commands run. */ +export function createBoatTransportAdmission() { + let active = 0; + let ordinary = 0; + let bulk = 0; + const queue: Array<{ control: boolean; bulk: boolean; start: () => void }> = []; + function drain() { + while (active < 8) { + const eligible = (entry: (typeof queue)[number]) => + entry.control || (ordinary < 6 && (!entry.bulk || bulk < 4)); + let index = queue.findIndex((entry) => entry.control && eligible(entry)); + if (index < 0) index = queue.findIndex(eligible); + if (index < 0) break; + queue.splice(index, 1)[0]!.start(); + } + } + return async function execute( + runner: CommandManagedRuntimeRunner, + input: CommandInput, + control: boolean, + ) { + const startedAt = new Date().toISOString(); + const deadline = Date.now() + (input.timeoutMs ?? 15_000); + const isBulk = Buffer.byteLength(input.stdin ?? "") > 1024 * 1024; + // Even process-control callers must use the ordinary budget for large uploads. + const isControl = control && !isBulk; + if (queue.length >= 128) + throw new ComputerError("conflict", "Computer command admission is busy"); + return new Promise>>((resolve, reject) => { + const timedOut = () => resolve({ exitCode: null, signal: null, timedOut: true, + stdout: "", stderr: "Computer command timed out waiting for SSH admission", pid: null, startedAt }); + const entry = { control: isControl, bulk: isBulk, start: () => { + clearTimeout(timer); + const remaining = deadline - Date.now(); + if (remaining <= 0) { timedOut(); return; } + active++; + if (!isControl) ordinary++; + if (isBulk) bulk++; + void Promise.resolve().then(() => runner.execute({ ...input, timeoutMs: remaining })) + .then(resolve, reject).finally(() => { + active--; + if (!isControl) ordinary--; + if (isBulk) bulk--; + drain(); + }); + } }; + const timer = setTimeout(() => { + const index = queue.indexOf(entry); + if (index >= 0) queue.splice(index, 1); + timedOut(); + drain(); + }, Math.max(1, deadline - Date.now())); + queue.push(entry); + drain(); + }); + }; +} +const transportAdmission = new Map>(); const desktopCache = new Map< string, { viewerUrl: string; expiresAt: string } @@ -166,7 +226,7 @@ export function boatBackend( stop: sandbox.stop ?? null, }; } - async function runner( + async function rawRunner( record: ComputerRecord, ): Promise { const key = `${record.id}:${record.ledger.secretRef.secretId}:${record.ledger.secretRef.version ?? "latest"}`; @@ -263,13 +323,23 @@ export function boatBackend( } return pending; } + async function runner(record: ComputerRecord, options?: { control?: boolean }): Promise { + const raw = await rawRunner(record); + let admission = transportAdmission.get(record.providerId); + if (!admission) { + admission = createBoatTransportAdmission(); + transportAdmission.set(record.providerId, admission); + } + const execute = admission; + return { ...raw, execute: (input) => execute(raw, input, options?.control === true) }; + } async function execute( record: ComputerRecord, code: string, payload: unknown, ) { const result = await ( - await runner(record) + await runner(record, { control: true }) ).execute({ command: "python3", args: ["-c", code], diff --git a/server/src/modules/computers/application/ports.ts b/server/src/modules/computers/application/ports.ts index 84c4ba51c7..649acafc2b 100644 --- a/server/src/modules/computers/application/ports.ts +++ b/server/src/modules/computers/application/ports.ts @@ -31,7 +31,7 @@ export interface ComputerBackend { ready(record: ComputerRecord): Promise; claim(record: ComputerRecord): Promise; advance(record: ComputerRecord, owner: Owner): Promise; - runner(record: ComputerRecord): Promise; + runner(record: ComputerRecord, options?: { control?: boolean }): Promise; launch( record: ComputerRecord, owner: Owner, diff --git a/server/src/modules/computers/application/service.ts b/server/src/modules/computers/application/service.ts index 1ea48de01a..3fb06c0a1c 100644 --- a/server/src/modules/computers/application/service.ts +++ b/server/src/modules/computers/application/service.ts @@ -170,6 +170,7 @@ export function createComputerService( if (!placement) throw new ComputerError("conflict", "Computer placement is missing"); const raw = await backend.runner(record); + const control = await backend.runner(record, { control: true }); let pinnedProcess = owner.process ? structuredClone(owner.process) : null; async function processScope() { if (!pinnedProcess) return scoped({ ...record, owner: ownerRef }); @@ -203,7 +204,7 @@ export function createComputerService( throw new ComputerError("conflict", "Computer owner is retired"); // Start a command unit under the same owner slice while holding the remote tombstone lock. // The lock is released only after systemd knows the unit, so retirement cannot miss a delayed launch. - return raw.execute({ + return (processScoped ? control : raw).execute({ command: "python3", args: [ "-c",