From cd1bfc1aa0d9c3cef3db12e3d0f9318a7af2accc Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 10 Oct 2026 11:23:47 -0500 Subject: [PATCH] Reserve Boat SSH capacity for lifecycle and process control Bound each physical Boat to eight SSH operations with six ordinary slots and four bulk transfers. Preserve command deadlines while queued and always release admission after transport completion or failure. Co-Authored-By: Paperclip --- .../modules/computers/adapters/boat.test.ts | 54 +++++++++++++- server/src/modules/computers/adapters/boat.ts | 74 ++++++++++++++++++- .../modules/computers/application/ports.ts | 2 +- .../modules/computers/application/service.ts | 3 +- 4 files changed, 128 insertions(+), 5 deletions(-) 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",