Start viewer lifetime after bounded computer preparation

Keep a separate starting reservation for Connect, propagate its preparation deadline, and atomically recheck the exact live owner before publishing credentials and the viewer idle deadline.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-11 09:28:17 -05:00
1 parent 7fa25f6b8c
commit 6cfdcbb659
5 files changed
+116 -24

No files matched your search

@@ -960,6 +960,28 @@ os.execve(command[0],command,env)
expect(readFileSync(f.calls, "utf8")).toBe(expected.repeat(2));
});
it("shares the remaining Connect preparation bound with input readiness and provider credentials", async () => {
let clock = 1_000_000;
const time = vi.spyOn(Date, "now").mockImplementation(() => clock);
const scoped = { ...record, id: randomUUID(), providerId: `bx_${randomUUID()}` };
const execute = vi.fn(async () => {
clock = 1_120_001;
return { exitCode: 0, stdout: "{}", stderr: "", timedOut: false, signal: null, pid: null, startedAt: "" };
});
sshFactory.mockReturnValue({ execute });
const fetcher = vi.fn(async () => {
clock += 90_000;
return json({ hostKey: "ssh-ed25519 AAAA", sshEndpoint: "fixture.invalid:2222" });
});
const backend = boatBackend(async () => "fixture", fetcher);
try {
await expect(backend.desktop(scoped, { deadlineMs: 1_120_000 })).rejects.toThrow("timed out");
expect(execute).toHaveBeenCalledWith(expect.objectContaining({ command: "timeout", timeoutMs: 30_000 }));
expect(fetcher).toHaveBeenCalledOnce(); // No late desktop credential request.
await expect(backend.renew(scoped, { deadlineMs: 1_120_000 })).rejects.toThrow("timed out");
expect(fetcher).toHaveBeenCalledOnce();
} finally { time.mockRestore(); sshFactory.mockReset(); }
});
it("requests a fresh stream on every Connect but reuses credentials during presence", async () => {
const id = randomUUID();
const scoped = { ...record, id, providerId: `bx_${id}` };
@@ -633,8 +633,8 @@ print('{}')
const value = await api(record, "GET", `/stops/${segment(id)}`);
return stopReceipt(value.stop ?? value, id);
},
async renew(record) {
await api(record, "PATCH", "", { ttlSeconds: 300 });
async renew(record, options) {
await api(record, "PATCH", "", { ttlSeconds: 300 }, options);
},
computerTool() {
return {
@@ -644,7 +644,7 @@ print('{}')
},
async desktop(record, options) {
if (options?.checkInput !== false)
await execute(record, `${desktopReadinessProgram}\nprint('{}')`, {});
await execute(record, `${desktopReadinessProgram}\nprint('{}')`, {}, true, options);
const cached = desktopCache.get(record.id);
// Explicit Connect must obtain a new stream session, including after a
// failed connection. Only presence renewals reuse existing credentials.
@@ -654,7 +654,7 @@ print('{}')
Date.parse(cached.expiresAt) > Date.now() + 60_000
)
return cached;
const value = await api(record, "POST", "/desktop", {});
const value = await api(record, "POST", "/desktop", {}, options);
const url = new URL(value.desktopUrl);
if (url.protocol !== "https:" || !url.hostname.endsWith(".on.boat.dev"))
throw new ComputerError("provider_error", "Invalid Boat desktop URL");
@@ -56,11 +56,11 @@ export interface ComputerBackend {
record: ComputerRecord,
id: string,
): Promise<{ id: string; status: string }>;
renew(record: ComputerRecord): Promise<void>;
renew(record: ComputerRecord, options?: ComputerOperationOptions): Promise<void>;
computerTool(): { command: string; args: string[] };
desktop(
record: ComputerRecord,
options?: { checkInput?: boolean },
options?: ComputerOperationOptions & { checkInput?: boolean },
): Promise<{ viewerUrl: string; expiresAt: string }>;
ingress(
record: ComputerRecord,
@@ -300,6 +300,45 @@ describe("computer ownership", () => {
expect((await f.repository.get(f.scope)).ledger).toMatchObject({ status: "detached", action: null });
expect(f.backend.stop).toHaveBeenCalledOnce();
});
it("starts viewer lifetime after bounded preparation, not when Connect was requested", async () => {
const f = fixture(); await f.attach();
vi.mocked(f.backend.ready).mockImplementationOnce(async () => {
f.advance(110_000);
await f.service.reconcile();
});
const viewer = await f.service.connect({ ...f.scope, userId: "alice", idleTimeoutMs: 10_000 });
expect(viewer.expiresAt).toBe("2026-10-10T12:02:00.000Z");
expect((await f.repository.get(f.scope)).ledger.owners).toEqual([expect.objectContaining({
id: viewer.owner.ownerId, phase: "active", deadline: viewer.expiresAt, absoluteDeadline: viewer.expiresAt,
})]);
expect(f.backend.stop).not.toHaveBeenCalled();
expect(f.backend.ready).toHaveBeenCalledWith(expect.anything(), { deadlineMs: Date.parse("2026-10-10T12:02:00Z") });
});
it.each(["ready", "desktop"] as const)("does not publish a viewer retired during %s preparation", async (step) => {
const f = fixture(); await f.attach();
const retireViewer = async () => {
const record = await f.repository.get(f.scope);
const owner = record.ledger.owners.find(owner => owner.kind === "viewer")!;
await f.service.retire({ ...f.scope, owner: { computerId: record.id, ownerId: owner.id, generation: owner.generation } });
};
if (step === "ready") vi.mocked(f.backend.ready).mockImplementationOnce(retireViewer);
else vi.mocked(f.backend.desktop).mockImplementationOnce(async () => {
await retireViewer();
return { viewerUrl: "https://test.on.boat.dev/#credential", expiresAt: "2026-10-10T12:09:00Z" };
});
await expect(f.service.connect({ ...f.scope, userId: "alice", idleTimeoutMs: 10_000 })).rejects.toMatchObject({ code: "conflict" });
if (step === "ready") expect(f.backend.desktop).not.toHaveBeenCalled();
expect((await f.repository.get(f.scope)).ledger.owners.every(owner => owner.phase === "retired")).toBe(true);
});
it("does not publish or continue viewer preparation after its absolute preparation bound", async () => {
const f = fixture(); await f.attach();
vi.mocked(f.backend.claim).mockClear();
vi.mocked(f.backend.ready).mockImplementationOnce(async () => { f.advance(120_001); });
await expect(f.service.connect({ ...f.scope, userId: "alice", idleTimeoutMs: 10_000 })).rejects.toMatchObject({ code: "provider_error" });
expect(f.backend.claim).not.toHaveBeenCalled();
expect(f.backend.desktop).not.toHaveBeenCalled();
expect((await f.repository.get(f.scope)).ledger.owners.every(owner => owner.phase === "retired")).toBe(true);
});
it("viewer renewals cannot extend the absolute warm deadline or use another user", async () => {
const f = fixture();
await f.attach();
@@ -27,6 +27,7 @@ class ComputerSessionBusyError extends ComputerError {
}
}
const VIEWER_PREPARATION_BUDGET_MS = 120_000;
const FILE_OPERATION_BUDGET_MS = 120_000;
const FILE_OPERATION_SETTLEMENT_MS = 5000;
type Scope = { companyId: string; environmentId: string };
@@ -618,36 +619,66 @@ finally:
input: Scope & { userId: string; idleTimeoutMs: number },
) {
timeout(input.idleTimeoutMs);
const deadline = new Date(
now().getTime() + input.idleTimeoutMs,
).toISOString();
const data = await admitRecord(input, (record) => {
const deadlineMs = now().getTime() + VIEWER_PREPARATION_BUDGET_MS;
const owner: Owner = {
id: randomUUID(),
generation: 1,
kind: "viewer",
phase: "active",
phase: "starting",
userId: input.userId,
port: 0,
deadline,
absoluteDeadline: deadline,
deadline: new Date(deadlineMs).toISOString(),
absoluteDeadline: null,
process: null,
};
record.ledger.owners.push(owner);
return { record: structuredClone(record), owner };
return { record: structuredClone(record), owner, deadlineMs };
});
await backend.ready(data.record);
await backend.claim(data.record);
await backend.renew(data.record);
const viewer = await backend.desktop(data.record);
return {
...viewer,
expiresAt: new Date(
Math.min(Date.parse(viewer.expiresAt), Date.parse(deadline)),
).toISOString(),
owner: ref(data.record, data.owner),
};
const ownerRef = ref(data.record, data.owner);
const operation = { deadlineMs: data.deadlineMs };
function preparing(record: ComputerRecord) {
const owner = exactOwner(record, ownerRef);
assertAdmission(record.ledger);
if (owner.kind !== "viewer" || owner.userId !== input.userId || owner.phase !== "starting")
throw new ComputerError("conflict", "Computer connection is no longer active");
if (now().getTime() >= data.deadlineMs || expired(owner, now()))
throw new ComputerError("provider_error", "Computer connection preparation timed out");
return owner;
}
const assertPreparing = async () => preparing(await repository.get(input));
try {
await assertPreparing();
await backend.ready(data.record, operation);
await assertPreparing();
await backend.claim(data.record, operation);
await assertPreparing();
await backend.renew(data.record, operation);
await assertPreparing();
const viewer = await backend.desktop(data.record, operation);
const deadline = await repository.update(input, (record) => {
const owner = preparing(record);
const expiresAt = Math.min(Date.parse(viewer.expiresAt), now().getTime() + input.idleTimeoutMs);
if (!Number.isFinite(expiresAt) || expiresAt <= now().getTime())
throw new ComputerError("provider_error", "Computer viewer credentials expired during preparation");
owner.phase = "active";
owner.deadline = new Date(expiresAt).toISOString();
owner.absoluteDeadline = owner.deadline;
return owner.deadline;
});
return { ...viewer, expiresAt: deadline, owner: ownerRef };
} catch (error) {
// Retire only this still-preparing reservation; never revive or change a
// concurrently disconnected/replaced owner when preparation completes late.
await repository.update(input, (record) => {
if (record.id !== ownerRef.computerId) return;
const owner = record.ledger.owners.find(owner => owner.id === ownerRef.ownerId && owner.generation === ownerRef.generation);
if (owner?.phase === "starting") { owner.phase = "retired"; owner.deadline = null; }
});
throw error;
}
}
async function renewViewer(
input: Scope & { owner: OwnerRef; userId: string },
) {