diff --git a/.changeset/js-exec-interrupts.md b/.changeset/js-exec-interrupts.md new file mode 100644 index 000000000..7a2a049b3 --- /dev/null +++ b/.changeset/js-exec-interrupts.md @@ -0,0 +1,5 @@ +--- +"@cloudflare/computer": minor +--- + +Add experimental `ctx.interrupt()` for host module functions, with `runtime.pending()` and `runtime.respond()`, so a replayable run can wait for an answer such as an approval. diff --git a/docs/17_isolate_javascript.md b/docs/17_isolate_javascript.md index 5de7f3561..91019f825 100644 --- a/docs/17_isolate_javascript.md +++ b/docs/17_isolate_javascript.md @@ -195,6 +195,40 @@ modules: { } ``` +### Waiting for an answer + +A host module function can pause a replayable run until someone answers, for example to approve an action. Like replay, this is experimental. The code being run just calls the function and gets its result, however long the answer takes: + +```ts +modules: { + "ws:github": { + postComment: async ([issue, body], ctx) => { + const { approved } = await ctx.interrupt({ kind: "approval", issue, body }); + if (!approved) throw new Error("The comment was rejected."); + return github.comment(issue, body, { idempotencyKey: ctx.callId }); + }, + }, +} +``` + +`ctx.interrupt(payload)` releases the run's isolate once the run's other calls in flight finish. The run stays running, and time spent waiting doesn't count toward its timeout. `workspace.runtime.pending()` lists the calls waiting, with their payloads, and `respond()` answers one: + +```ts +for (const call of await workspace.runtime.pending()) { + await workspace.runtime.respond(call.id, call.callId, { approved: true }); +} +``` + +To hear about a wait when it starts, for example to message whoever answers, give the backend an `onWait` function. It gets the same fields, at most once per wait, so `pending()` stays the source of truth. A Worker that calls the object can reach `pending()` and `respond()` through `workspace.stub().runtime`. + +Once none of its calls are waiting, the run replays to the call and the function runs again with the same `callId`. This time `interrupt()` returns the answer straight away, and the function carries on. Anything it does before `interrupt()` runs twice, so keep that part safe to repeat. A waiting run survives a deploy like any interrupted run, and `killExec()` ends one. + +- A call can interrupt once. A second `interrupt()` in the same call throws. +- Only a run with `onInterrupt: "replay"` can wait. Anywhere else, `interrupt()` throws, so a Workspace with modules that interrupt should make `"replay"` its backend's default. +- A run is released once each call in flight has finished or is waiting too, and those waits are answered together. A call the run makes after the first interrupt is held until it resumes, so a call that interrupts straight away can leave the next one to wait for a later round. +- `respond()` throws `ENOTWAITING` for a call that isn't waiting, such as one already answered, so answering twice changes nothing. +- Every answer replays the run from the top. + ### Limits - A run gets three attempts, counted again each time it journals a call, and has to finish within its original timeout, counted from when it first started. diff --git a/examples/rlm/worker/capability.test.ts b/examples/rlm/worker/capability.test.ts index b17848ebb..6aaf5f084 100644 --- a/examples/rlm/worker/capability.test.ts +++ b/examples/rlm/worker/capability.test.ts @@ -40,6 +40,7 @@ function context(signal = new AbortController().signal): WorkspaceModuleCallCont deadline: Date.now() + 1_000, access: "read", callId: "test-call", + interrupt: () => Promise.reject(new Error("not used")), resolvePath: async (path) => path, }; } diff --git a/packages/computer/README.md b/packages/computer/README.md index fb518a2e8..3721cb503 100644 --- a/packages/computer/README.md +++ b/packages/computer/README.md @@ -31,7 +31,7 @@ removed in a future release; each one names its replacement. | Observability hook and its Cloudflare adapter | `@cloudflare/computer`, `@cloudflare/computer/observe/cloudflare` | Stable | | `ContainerBackend` and `withWorkspaceContainer` | `@cloudflare/computer/backends/container` | Experimental | | `LegacyContainerBackend` and `withLegacyWorkspaceContainer` | `@cloudflare/computer/backends/container-legacy` | Deprecated: use `ContainerBackend` | -| Resuming interrupted isolate JavaScript: `onInterrupt: "restart"` and `"replay"`, `workspace.runtime.recover()`, and `unsafeToRepeat()` | `@cloudflare/computer`, `@cloudflare/computer/backends/worker-javascript` | Experimental | +| Resuming interrupted isolate JavaScript: `onInterrupt: "restart"` and `"replay"`, `workspace.runtime.recover()`, `unsafeToRepeat()`, and waiting for answers with `ctx.interrupt()`, `pending()` and `respond()` | `@cloudflare/computer`, `@cloudflare/computer/backends/worker-javascript` | Experimental | | `WorkerShellBackend` and the shell command groups | `@cloudflare/computer/backends/worker-shell`, `@cloudflare/computer/shell/*` | Experimental | | The `ws:container` host module | `@cloudflare/computer/modules/container` | Experimental | | `workspace.push()`, `workspace.pull()`, and the sync protocol between the durable object and `computerd` | `@cloudflare/computer` | Experimental | diff --git a/packages/computer/src/backends/worker-javascript/journal.test.ts b/packages/computer/src/backends/worker-javascript/journal.test.ts index 333453376..3728394db 100644 --- a/packages/computer/src/backends/worker-javascript/journal.test.ts +++ b/packages/computer/src/backends/worker-javascript/journal.test.ts @@ -3,12 +3,14 @@ import { SQLiteTestStorage } from "@cloudflare/dofs/testing"; import { describe, expect, it, vi } from "vitest"; import { + answerCall, callKey, createResumeTables, decodeValue, describeCall, encodeValue, ExecutionJournal, + waitingCalls, } from "./journal.js"; function database() { @@ -97,6 +99,66 @@ describe("ExecutionJournal", () => { expect(replay.expectedCall()).toBeUndefined(); }); + it("holds an interrupted call until it's answered, then gives it the answer", () => { + const db = database(); + db.run( + `INSERT INTO workspace_runtime_resume (backend, execution_id, on_interrupt, attempt) + VALUES ('js', 'run', 'replay', 1)`, + ); + const first = new ExecutionJournal(db, "js", "run"); + const step = first.begin("host/ws:test.approve", ["deploy"]); + if (step.kind !== "live") throw new Error("expected a live call"); + first.interrupt(step.seq, { action: "deploy" }); + expect(waitingCalls(db, "js")).toEqual([ + { id: "run", seq: 1, payload: { action: "deploy" }, since: expect.any(Number) }, + ]); + expect(answerCall(db, "js", "run", 2, true)).toBeUndefined(); + expect(answerCall(db, "js", "run", 1, { ok: true })).toBe(0); + expect(answerCall(db, "js", "run", 1, { ok: true })).toBeUndefined(); + expect(waitingCalls(db, "js")).toEqual([]); + const answer = { value: { ok: true } }; + const resumed = new ExecutionJournal(db, "js", "run"); + expect(resumed.begin("host/ws:test.approve", ["deploy"])).toEqual({ + kind: "live", + seq: 1, + rerun: false, + answer, + }); + // Interrupted again while it runs with its answer: a rerun that keeps it. + const again = new ExecutionJournal(db, "js", "run"); + expect(again.begin("host/ws:test.approve", ["deploy"])).toEqual({ + kind: "live", + seq: 1, + rerun: true, + answer, + }); + }); + + it("keeps when each call started waiting, and the run's first", () => { + const db = database(); + db.run( + `INSERT INTO workspace_runtime_resume (backend, execution_id, on_interrupt, attempt) + VALUES ('js', 'run', 'replay', 1)`, + ); + const journal = new ExecutionJournal(db, "js", "run"); + const now = vi.spyOn(Date, "now"); + for (const [action, at] of [ + ["one", 1000], + ["two", 5000], + ] as const) { + const step = journal.begin("host/ws:test.approve", [action]); + if (step.kind !== "live") throw new Error("expected a live call"); + now.mockReturnValue(at); + expect(journal.interrupt(step.seq, { action })).toBe(at); + } + now.mockRestore(); + expect(waitingCalls(db, "js").map((call) => call.since)).toEqual([1000, 5000]); + expect( + db.one<{ waiting_since: number }>("SELECT waiting_since FROM workspace_runtime_resume") + ?.waiting_since, + ).toBe(1000); + }); + it("numbers responses in the order they were committed", () => { const db = database(); const first = new ExecutionJournal(db, "js", "run"); diff --git a/packages/computer/src/backends/worker-javascript/journal.ts b/packages/computer/src/backends/worker-javascript/journal.ts index f5c6dae4f..3d6211b5a 100644 --- a/packages/computer/src/backends/worker-javascript/journal.ts +++ b/packages/computer/src/backends/worker-javascript/journal.ts @@ -80,14 +80,29 @@ export interface ResumeState { /** Calls journaled when the attempt started. */ readonly callsAtAttempt: number; readonly start: ResumeStart; + /** When the run's first unanswered interrupt was made, if one is waiting. */ + readonly waitingSince: number | null; + /** Time the run has spent waiting for answers, which its timeout leaves out. */ + readonly waitedMs: number; +} + +/** A call waiting for an answer, as `pending()` lists it. */ +export interface WaitingCall { + readonly id: string; + readonly seq: number; + readonly payload: unknown; + readonly since: number; } interface JournalEntry { readonly seq: number; readonly key: string; readonly label: string; - readonly state: "executing" | "applied"; + // An interrupted call is `waiting` until it's answered, then `answered` + // until the resumed run makes it again. + readonly state: "executing" | "applied" | "waiting" | "answered"; readonly response: string | null; + readonly answer: string | null; readonly at: number | null; readonly ordinal: number | null; } @@ -105,6 +120,8 @@ export type JournalStep = readonly seq: number; /** The call was in flight when the run was interrupted. */ readonly rerun: boolean; + /** The answer to the call's interrupt, which it now returns. */ + readonly answer?: { readonly value: unknown }; } | { readonly kind: "diverge"; readonly message: string }; @@ -130,6 +147,8 @@ export function createResumeTables(db: Database): void { execution_id TEXT NOT NULL, on_interrupt TEXT NOT NULL, attempt INTEGER NOT NULL, + waiting_since INTEGER, + waited_ms INTEGER NOT NULL DEFAULT 0, calls_at_attempt INTEGER NOT NULL DEFAULT 0, PRIMARY KEY (backend, execution_id) ) @@ -152,6 +171,9 @@ export function createResumeTables(db: Database): void { label TEXT NOT NULL, state TEXT NOT NULL, response TEXT, + interrupt_payload TEXT, + waiting_since INTEGER, + answer TEXT, journaled_at INTEGER, journal_order INTEGER, PRIMARY KEY (backend, execution_id, seq) @@ -209,8 +231,14 @@ export function saveResume( } export function loadResume(db: Database, backend: string, id: string): ResumeState | undefined { - const row = db.one<{ on_interrupt: string; attempt: number; calls_at_attempt: number }>( - `SELECT on_interrupt, attempt, calls_at_attempt FROM workspace_runtime_resume + const row = db.one<{ + on_interrupt: string; + attempt: number; + calls_at_attempt: number; + waiting_since: number | null; + waited_ms: number; + }>( + `SELECT on_interrupt, attempt, calls_at_attempt, waiting_since, waited_ms FROM workspace_runtime_resume WHERE backend = ? AND execution_id = ?`, backend, id, @@ -230,6 +258,8 @@ export function loadResume(db: Database, backend: string, id: string): ResumeSta attempt: Number(row.attempt), callsAtAttempt: Number(row.calls_at_attempt), start, + waitingSince: row.waiting_since === null ? null : Number(row.waiting_since), + waitedMs: Number(row.waited_ms), }; } catch { return undefined; @@ -253,6 +283,67 @@ export function recordAttempt( ); } +/** + * Answer a waiting call. Returns how many of the run's calls are still + * waiting, or `undefined` if this one wasn't. Once none are, the run's + * wait ends and its timeout counts again. + */ +export function answerCall( + db: Database, + backend: string, + id: string, + seq: number, + value: unknown, +): number | undefined { + const answered = db.all( + `UPDATE workspace_runtime_journal SET state = 'answered', answer = ? + WHERE backend = ? AND execution_id = ? AND seq = ? AND state = 'waiting' + RETURNING seq`, + encodeLimited(value, "answer"), + backend, + id, + seq, + ); + if (answered.length === 0) return undefined; + const waiting = Number( + db.one<{ count: number }>( + `SELECT COUNT(*) AS count FROM workspace_runtime_journal + WHERE backend = ? AND execution_id = ? AND state = 'waiting'`, + backend, + id, + )?.count ?? 0, + ); + if (waiting === 0) { + db.run( + `UPDATE workspace_runtime_resume + SET waited_ms = waited_ms + MAX(0, ? - waiting_since), waiting_since = NULL + WHERE backend = ? AND execution_id = ? AND waiting_since IS NOT NULL`, + Date.now(), + backend, + id, + ); + } + return waiting; +} + +/** Every call of this backend's runs that's waiting for an answer. */ +export function waitingCalls(db: Database, backend: string): WaitingCall[] { + return db + .all<{ execution_id: string; seq: number; interrupt_payload: string; waiting_since: number }>( + `SELECT execution_id, seq, interrupt_payload, waiting_since + FROM workspace_runtime_journal + WHERE backend = ? AND state = 'waiting' + ORDER BY waiting_since, execution_id, seq`, + backend, + ) + .map((row) => ({ + id: row.execution_id, + seq: Number(row.seq), + payload: decodeValue(row.interrupt_payload), + since: Number(row.waiting_since), + })); +} + /** How many calls a run has journaled. */ export function journaledCalls(db: Database, backend: string, id: string): number { const row = db.one<{ calls: number }>( @@ -325,10 +416,12 @@ export class ExecutionJournal implements BridgeJournal { label: string; state: string; response: string | null; + answer: string | null; journaled_at: number | null; journal_order: number | null; }>( - `SELECT seq, call_key, label, state, response, journaled_at, journal_order FROM workspace_runtime_journal + `SELECT seq, call_key, label, state, response, answer, journaled_at, journal_order + FROM workspace_runtime_journal WHERE backend = ? AND execution_id = ? ORDER BY seq`, backend, id, @@ -337,9 +430,10 @@ export class ExecutionJournal implements BridgeJournal { seq: Number(row.seq), key: row.call_key, label: row.label, - state: row.state === "applied" && row.response !== null ? "applied" : "executing", + state: journalState(row.state, row.response), // An empty response marks one stored in chunks. response: row.response === "" ? this.#chunkedResponse(Number(row.seq)) : row.response, + answer: row.answer, at: row.journaled_at === null ? null : Number(row.journaled_at), ordinal: row.journal_order === null ? null : Number(row.journal_order), })); @@ -441,9 +535,23 @@ export class ExecutionJournal implements BridgeJournal { ordinal: entry.ordinal ?? 0, }; } + const answer = + entry.answer === null ? undefined : { value: decodeValue(entry.answer) as unknown }; + // An answered call runs again by design, so it isn't a rerun until + // it's under way. + if (entry.state === "answered") { + this.#db.run( + `UPDATE workspace_runtime_journal SET state = 'executing' + WHERE backend = ? AND execution_id = ? AND seq = ?`, + this.#backend, + this.#id, + entry.seq, + ); + return { kind: "live", seq: entry.seq, rerun: false, answer }; + } // The call was in flight when the execution was interrupted, so its // response never reached the code. Run it again. - return { kind: "live", seq: entry.seq, rerun: true }; + return { kind: "live", seq: entry.seq, rerun: true, answer }; } if (this.#unanswered > 0) { return { @@ -500,6 +608,34 @@ export class ExecutionJournal implements BridgeJournal { return { at, ordinal: this.#lastOrdinal }; } + /** + * Mark a live call as waiting for an answer, and return when it + * started waiting. Its response is never committed: the run is + * released, and the call runs again once answered. The run's own + * waiting_since is when its first unanswered call started waiting, + * which is what its timeout leaves out. + */ + interrupt(seq: number, payload: unknown): number { + const now = Date.now(); + this.#db.run( + `UPDATE workspace_runtime_journal SET state = 'waiting', interrupt_payload = ?, waiting_since = ? + WHERE backend = ? AND execution_id = ? AND seq = ?`, + encodeLimited(payload, "payload"), + now, + this.#backend, + this.#id, + seq, + ); + this.#db.run( + `UPDATE workspace_runtime_resume SET waiting_since = COALESCE(waiting_since, ?) + WHERE backend = ? AND execution_id = ?`, + now, + this.#backend, + this.#id, + ); + return now; + } + diverged(message: string): void { this.#onDiverge(message); } @@ -579,6 +715,24 @@ function cyrb53(text: string, seed: number): string { return (4294967296 * (2097151 & h2) + (h1 >>> 0)).toString(36).padStart(11, "0"); } +function journalState(state: string, response: string | null): JournalEntry["state"] { + if (state === "applied") return response === null ? "executing" : "applied"; + return state === "waiting" || state === "answered" ? state : "executing"; +} + +// Interrupt payloads and answers are stored in one row each. +const MAX_WAIT_CHARACTERS = CHUNK_CHARACTERS; + +function encodeLimited(value: unknown, what: string): string { + const text = encodeValue(value); + if (text.length > MAX_WAIT_CHARACTERS) { + throw new Error( + `An interrupt's ${what} must encode to at most ${MAX_WAIT_CHARACTERS} characters.`, + ); + } + return text; +} + const BYTES_TAG = "\u0000bytes"; export function encodeValue(value: unknown): string { diff --git a/packages/computer/src/backends/worker-javascript/worker-javascript.test.ts b/packages/computer/src/backends/worker-javascript/worker-javascript.test.ts index 91b5471c0..5d4d85726 100644 --- a/packages/computer/src/backends/worker-javascript/worker-javascript.test.ts +++ b/packages/computer/src/backends/worker-javascript/worker-javascript.test.ts @@ -2086,6 +2086,28 @@ describe("WorkerJavaScriptBackend interrupted executions", () => { for (const workspace of [...workspaces, last]) await workspace.close(); }); + it("starts a run's attempts again once its wait is answered", async () => { + const storage = new SQLiteTestStorage(); + const db = new Database(storage); + const workspaces = [await interrupt(storage)]; + for (let attempt = 0; attempt < 2; attempt += 1) { + const workspace = open(storage); + await workspace.runtime.recover(); + workspaces.push(workspace); + } + expect(loadResume(db, "worker-javascript", "run")?.attempt).toBe(3); + // The run's next call waits for an answer. + const journal = new ExecutionJournal(db, "worker-javascript", "run"); + const step = journal.begin("host/ws:test.approve", []); + if (step.kind !== "live") throw new Error("expected a live call"); + journal.interrupt(step.seq, { kind: "approval" }); + const answering = open(storage); + workspaces.push(answering); + await answering.runtime.respond("run", `run:${step.seq}`, true); + expect(loadResume(db, "worker-javascript", "run")?.attempt).toBe(1); + for (const workspace of workspaces) await workspace.close(); + }); + it("asks for a REPLAY_VERSION check when the runtime's code changes", () => { // The code the runtime loads into the isolate changed. If a run that // started on the old code could replay differently on the new code, diff --git a/packages/computer/src/backends/worker-javascript/worker-javascript.ts b/packages/computer/src/backends/worker-javascript/worker-javascript.ts index 43436fd08..26262a5e3 100644 --- a/packages/computer/src/backends/worker-javascript/worker-javascript.ts +++ b/packages/computer/src/backends/worker-javascript/worker-javascript.ts @@ -18,10 +18,12 @@ import type { WorkspaceRuntimeEvent, WorkspaceRuntimeInterruptPolicy, WorkspaceRuntimeLoader, + WorkspaceRuntimePendingCall, WorkspaceRuntimeValue, } from "../../runtime/types.js"; import { decodeRuntimeFrames, type RuntimeFrame } from "./frames.js"; import { + answerCall, clearResume, createResumeTables, RESUME_TABLES, @@ -38,6 +40,7 @@ import { REPLAY_VERSION, saveResume, toBase64, + waitingCalls, } from "./journal.js"; import { assertHostModuleExports, @@ -115,6 +118,16 @@ export interface WorkerJavaScriptBackendOptions { * release. */ onInterrupt?: WorkspaceRuntimeInterruptPolicy; + /** + * Called when a host module call starts waiting in `ctx.interrupt()`, + * with the fields `runtime.pending()` lists, for example to ask someone + * for the answer. It's called once the wait is saved, and at most once + * per wait: a crash in between loses the call. Treat `pending()` as the + * source of truth, for example by checking it when the object starts. + * What it throws is logged and otherwise ignored. Experimental: it may + * change in any release. + */ + onWait?: (call: WorkspaceRuntimePendingCall) => void | Promise; /** * How often to schedule a timer while any execution is running, in * milliseconds. Set it to 0 to turn the timer off. Defaults to 60 @@ -256,6 +269,9 @@ interface ExecutionRecord { // Output a replayed run already recorded. Its frames must repeat these // bytes, which the record checks and doesn't append again. replayed?: Record<"stdout" | "stderr", { bytes: Uint8Array; offset: number }>; + // Set while a run's isolate is being released for an interrupt, and + // once it is, until the run resumes. + waiting?: "releasing" | "released"; } export class WorkerJavaScriptBackend implements WorkspaceModuleBackend { @@ -489,24 +505,11 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { const record = this.#loadRecord(id); if (record?.status !== "running") return; this.#records.set(id, record); - const fail = (message: string) => - this.#finish(record, "failed", [ - { - id, - seq: record.events.length + 1, - name: "stderr", - value: new TextEncoder().encode(`${message}\n`), - }, - { id, seq: record.events.length + 2, name: "exit", code: 1 }, - ]); - let saved: ReturnType; - try { - saved = loadResume(this.#host.db, this.#backendId, id); - } catch { - saved = undefined; - } - if (saved === undefined) { - fail("Execution was interrupted when its Workspace runtime restarted."); + const saved = this.#loadResume(record); + if (saved === undefined) return; + // A run waiting for an answer resumes once it has every one. + if (saved.waitingSince !== null) { + record.waiting = "released"; return; } // Attempts count from the last call the run journaled, so a long run @@ -515,26 +518,142 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { const progressed = journaledCalls(this.#host.db, this.#backendId, id) > saved.callsAtAttempt; const attempt = progressed ? 1 : saved.attempt; if (attempt >= MAX_INTERRUPT_ATTEMPTS) { - fail( + this.#failInterrupted( + record, `Execution was interrupted ${attempt} times without making progress and won't be started again.`, ); return; } + this.#queueRelaunch(record, { ...saved, attempt }, true); + } + + #loadResume(record: ExecutionRecord) { + let saved: ReturnType; + try { + saved = loadResume(this.#host.db, this.#backendId, record.id); + } catch { + saved = undefined; + } + if (saved === undefined) { + this.#failInterrupted( + record, + "Execution was interrupted when its Workspace runtime restarted.", + ); + } + return saved; + } + + #failInterrupted(record: ExecutionRecord, message: string) { + this.#finish(record, "failed", [ + { + id: record.id, + seq: record.events.length + 1, + name: "stderr", + value: new TextEncoder().encode(`${message}\n`), + }, + { id: record.id, seq: record.events.length + 2, name: "exit", code: 1 }, + ]); + } + + // Queue a run to start again, on the runtime code it started on. + // Resuming after an answer isn't an attempt. + #queueRelaunch( + record: ExecutionRecord, + saved: NonNullable>, + counts: boolean, + ) { const { start } = saved; if (start.version !== JOURNAL_VERSION || start.fingerprint !== this.#fingerprint()) { - fail( + this.#failInterrupted( + record, "Execution was interrupted and can't resume, because the JavaScript runtime changed since it started.", ); return; } - const resumable = { ...saved, attempt }; this.#recoveryQueue.push(() => { if (record.status !== "running" || record.finalization) return; - this.#recovering.add(id); - this.#relaunch(record, resumable, fail); + this.#recovering.add(record.id); + this.#relaunch(record, saved, counts); }); } + // A call interrupted to wait for an answer. Let the run's other calls + // finish and be journaled, then release its isolate. The run stays + // running and resumes once every waiting call is answered. + async #release(record: ExecutionRecord) { + const { bridge } = record; + if (record.waiting !== undefined || bridge === undefined) return; + record.waiting = "releasing"; + await bridge.quiesce(); + // It may have settled meanwhile, for example by failing on its own. + if (record.status !== "running" || record.finalization) return; + record.bridge = undefined; + record.control?.cancel(); + await record.control?.completion.catch(() => undefined); + await bridge.cancelAndDrain().catch(() => undefined); + record.control = undefined; + record.journal = undefined; + record.waiting = "released"; + this.#markSettled(record.id); + this.#resumeIfAnswered(record); + } + + #notifyWait(record: ExecutionRecord, call: { seq: number; payload: unknown; since: number }) { + const { onWait } = this.#options; + if (onWait === undefined) return; + const pending: WorkspaceRuntimePendingCall = { + id: record.id, + backend: this.#backendId, + callId: `${record.id}:${call.seq}`, + payload: call.payload as WorkspaceRuntimeValue, + since: call.since, + }; + // After the current call, so the hook can't hold up the run's release. + void Promise.resolve() + .then(() => onWait(pending)) + .catch((error: unknown) => { + console.error({ message: "onWait failed", id: record.id, error: String(error) }); + }); + } + + #resumeIfAnswered(record: ExecutionRecord) { + if (record.status !== "running" || record.waiting !== "released") return; + const saved = this.#loadResume(record); + if (saved === undefined || saved.waitingSince !== null) return; + this.#queueRelaunch(record, saved, false); + this.#drainRecoveries(); + } + + async pending(): Promise[]> { + return waitingCalls(this.#host.db, this.#backendId).map((call) => ({ + id: call.id, + callId: `${call.id}:${call.seq}`, + payload: call.payload as WorkspaceRuntimeValue, + since: call.since, + })); + } + + async respond(input: { id: string; callId: string; answer: WorkspaceRuntimeValue }) { + assertExecutionId(input.id); + assertRuntimeValue(input.answer); + const record = this.#recordFor(input.id); + const prefix = `${input.id}:`; + const seq = input.callId.startsWith(prefix) ? Number(input.callId.slice(prefix.length)) : NaN; + const waiting = + record?.status === "running" && Number.isSafeInteger(seq) + ? this.#host.db.transactionSync(() => + answerCall(this.#host.db, this.#backendId, input.id, seq, input.answer), + ) + : undefined; + if (record === undefined || waiting === undefined) { + throw runtimeError( + "ENOTWAITING", + `execution ${input.id} has no call ${input.callId} waiting for an answer`, + ); + } + if (waiting === 0) this.#resumeIfAnswered(record); + } + // A Durable Object can run only so many Dynamic Workers at once, and // new executions need some of them, so interrupted runs start a few at // a time. Each one that settles lets the next start. @@ -548,23 +667,27 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { #relaunch( record: ExecutionRecord, - { policy, attempt, start }: NonNullable>, - fail: (message: string) => void, + { policy, attempt, start, waitedMs }: NonNullable>, + counts: boolean, ) { const id = record.id; + const fail = (message: string) => this.#failInterrupted(record, message); // A run's timeout counts from when it first started, including the - // time it spent interrupted or waiting to resume. - const remainingMs = start.startedAt + start.timeoutMs - Date.now(); + // time it spent interrupted or waiting to resume, but not the time it + // waited for answers. + const remainingMs = start.startedAt + start.timeoutMs + waitedMs - Date.now(); if (remainingMs <= 0) { fail("Execution was interrupted and its timeout passed before it could resume."); return; } + record.waiting = undefined; try { + // An answer counts as progress, so the run's attempts start again. recordAttempt( this.#host.db, this.#backendId, id, - attempt + 1, + counts ? attempt + 1 : 1, journaledCalls(this.#host.db, this.#backendId, id), ); let journal: ExecutionJournal | undefined; @@ -818,6 +941,10 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { maxResultBytes: this.#options.maxResultBytes, onAttachOutput: (readable) => this.#pumpFrames(record, readable), callIdPrefix: record.id, + onWait: (call) => { + this.#notifyWait(record, call); + void this.#release(record); + }, ...hostEgressFetch(this.#options.egress), ...(spec.journal === undefined ? {} : { journal: spec.journal }), }); @@ -1030,6 +1157,9 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { // as it arrives. Resolves when the stream closes, which is the // signal the runner's evaluate awaits before returning. async #pumpFrames(record: ExecutionRecord, readable: ReadableStream) { + // A run released for an interrupt has moved on from this attempt, and + // whatever its isolate writes on the way out isn't the run's output. + const attempt = record.bridge; const reader = decodeRuntimeFrames(readable).getReader(); try { while (true) { @@ -1050,7 +1180,7 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { if (record.status !== "running") break; throw error; } - if (next.done) break; + if (next.done || record.bridge !== attempt) break; this.#ingestFrame(record, next.value); } } finally { @@ -1109,6 +1239,8 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { #finalize(record: ExecutionRecord, errorMessage?: string): Promise { if (record.finalization) return record.finalization; if (record.status !== "running") return Promise.resolve(); + // Its isolate ending for an interrupt isn't the run ending. + if (record.waiting === "released") return Promise.resolve(); const finalization = this.#finalizeOnce(record, errorMessage); record.finalization = finalization; return finalization; diff --git a/packages/computer/src/index.ts b/packages/computer/src/index.ts index f8bd47dd2..cdbd71f91 100644 --- a/packages/computer/src/index.ts +++ b/packages/computer/src/index.ts @@ -108,6 +108,7 @@ export type { WorkspaceRuntimeKillOptions, WorkspaceRuntimeLoader, WorkspaceRuntimeOutputOptions, + WorkspaceRuntimePendingCall, WorkspaceRuntimeResult, WorkspaceRuntimeStatus, WorkspaceRuntimeTruncation, diff --git a/packages/computer/src/modules/artifacts.test.ts b/packages/computer/src/modules/artifacts.test.ts index b22e36c09..ec2b75423 100644 --- a/packages/computer/src/modules/artifacts.test.ts +++ b/packages/computer/src/modules/artifacts.test.ts @@ -43,6 +43,7 @@ function callContext(access: "read" | "read-write" = "read-write"): WorkspaceMod deadline: Date.now() + 60_000, access, callId: "test-call", + interrupt: () => Promise.reject(new Error("not used")), resolvePath: async (path) => path, }; } diff --git a/packages/computer/src/modules/assets.test.ts b/packages/computer/src/modules/assets.test.ts index dd2f88b5d..eb564ea5f 100644 --- a/packages/computer/src/modules/assets.test.ts +++ b/packages/computer/src/modules/assets.test.ts @@ -41,6 +41,7 @@ function callContext( deadline: Date.now() + 60_000, access: "read-write", callId: "test-call", + interrupt: () => Promise.reject(new Error("not used")), resolvePath: async (path) => (path.startsWith("/") ? path : `/workspace/${path}`), ...overrides, }; diff --git a/packages/computer/src/modules/container.test.ts b/packages/computer/src/modules/container.test.ts index e9947e3e0..1ab720520 100644 --- a/packages/computer/src/modules/container.test.ts +++ b/packages/computer/src/modules/container.test.ts @@ -90,6 +90,7 @@ function callContext( deadline: Date.now() + 60_000, access: "read-write", callId: "test-call", + interrupt: () => Promise.reject(new Error("not used")), resolvePath: async (path) => path, ...overrides, }; diff --git a/packages/computer/src/modules/tools.test.ts b/packages/computer/src/modules/tools.test.ts index 6ccf4d4a6..f7db6e3b8 100644 --- a/packages/computer/src/modules/tools.test.ts +++ b/packages/computer/src/modules/tools.test.ts @@ -18,6 +18,7 @@ function callContext( deadline: Date.now() + 60_000, access: "read-write", callId: "test-call", + interrupt: () => Promise.reject(new Error("not used")), resolvePath: async (path) => path, ...overrides, }; diff --git a/packages/computer/src/runtime/bridge.test.ts b/packages/computer/src/runtime/bridge.test.ts index c74f1ec5c..a73405583 100644 --- a/packages/computer/src/runtime/bridge.test.ts +++ b/packages/computer/src/runtime/bridge.test.ts @@ -249,7 +249,7 @@ describe("WorkspaceRuntimeBridge assertResult", () => { }); describe("WorkspaceRuntimeBridge journal", () => { - function journaled(steps: ReturnType[]) { + function journaled(steps: ReturnType[], maxTotalResponseBytes = 4) { const commits: number[] = []; const responses: unknown[] = []; const journal: BridgeJournal = { @@ -263,6 +263,7 @@ describe("WorkspaceRuntimeBridge journal", () => { if (!Array.isArray(events)) throw new Error("malformed event log"); logs.push([position, events]); }, + interrupt: () => 0, diverged: (message) => diverged.push(message), stalled: () => diverged.push("stalled"), }; @@ -270,7 +271,7 @@ describe("WorkspaceRuntimeBridge journal", () => { const logs: unknown[] = []; const contexts: WorkspaceModuleCallContext[] = []; const target = new WorkspaceRuntimeBridge({} as WorkspaceRuntimeCapability, { - maxTotalResponseBytes: 4, + maxTotalResponseBytes, hostModules: new Map([ [ "ws:test", @@ -280,6 +281,10 @@ describe("WorkspaceRuntimeBridge journal", () => { return "ok"; }, charge: unsafeToRepeat(async () => "c"), + twice: async (_args: unknown, context: WorkspaceModuleCallContext) => [ + await context.interrupt("first"), + await context.interrupt("second"), + ], }, ], ]), @@ -349,6 +354,28 @@ describe("WorkspaceRuntimeBridge journal", () => { expect(logs).toEqual([[0, [-1, -2]]]); }); + it("runs an unsafe call again once answered, but not if it was cut off after", async () => { + const { target, diverged } = journaled([ + { kind: "live", seq: 1, rerun: false, answer: { value: true } }, + { kind: "live", seq: 1, rerun: true, answer: { value: true } }, + ]); + await expect(target.call("host/ws:test.charge", args)).resolves.toMatchObject({ result: "c" }); + await expect(message(target.call("host/ws:test.charge", args))).resolves.toContain( + "isn't safe to repeat", + ); + expect(diverged).toHaveLength(1); + }); + + it("lets a call interrupt only once", async () => { + const { target } = journaled( + [{ kind: "live", seq: 1, rerun: false, answer: { value: "yes" } }], + 4096, + ); + await expect(message(target.call("host/ws:test.twice", args))).resolves.toContain( + "ctx.interrupt() can only be called once per host module call.", + ); + }); + it("passes a stalled replay to the journal", () => { const { target, diverged } = journaled([]); target.reportStall(); diff --git a/packages/computer/src/runtime/bridge.ts b/packages/computer/src/runtime/bridge.ts index 84a331a40..c1828438f 100644 --- a/packages/computer/src/runtime/bridge.ts +++ b/packages/computer/src/runtime/bridge.ts @@ -28,6 +28,15 @@ export class WorkspaceRuntimeBridge extends RpcTarget { #fetchesRunning = 0; readonly #fetchQueue: Array<() => void> = []; readonly #abortControllers = new Set(); + // Responses on their way to the isolate, by their call's controller. + readonly #responses = new Map, AbortController>(); + // Calls that interrupted, by their controller. + readonly #waiting = new Set(); + readonly #onWait?: (call: { seq: number; payload: unknown; since: number }) => void; + // Set once a call interrupts: new calls are held, never started. + #holding = false; + // Wakes quiesce() when another call starts waiting. + #waited?: () => void; #cancelled = false; #callTimedOut = false; #calls = 0; @@ -52,6 +61,8 @@ export class WorkspaceRuntimeBridge extends RpcTarget { fetch?: (request: Request) => Promise; // Prefix of every call's callId, such as the execution id. callIdPrefix?: string; + // Called when a call interrupts, to release the run. + onWait?: (call: { seq: number; payload: unknown; since: number }) => void; } = {}, ) { super(); @@ -68,6 +79,7 @@ export class WorkspaceRuntimeBridge extends RpcTarget { this.#journal = integrations.journal; this.#fetch = integrations.fetch; this.#callIdPrefix = integrations.callIdPrefix ?? crypto.randomUUID(); + this.#onWait = integrations.onWait; } // Drain the runner's framed output stream. The isolate passes the @@ -112,6 +124,9 @@ export class WorkspaceRuntimeBridge extends RpcTarget { return Promise.resolve(boundedError(error, this.#maxPayloadBytes)); } } + // The run is being released for an interrupt. A call it makes now + // runs live when it resumes. + if (this.#holding) return new Promise(() => undefined); let requestBytes: number; try { requestBytes = measureValue(args, this.#maxPayloadBytes, "request"); @@ -181,12 +196,23 @@ export class WorkspaceRuntimeBridge extends RpcTarget { const abort = new AbortController(); this.#abortControllers.add(abort); const deadline = Date.now() + this.#maxCallDurationMs; + // The journal holds one answer per call, so a call can wait only once. + let interrupted = false; const operation = respond(async () => { if (name.startsWith("host/")) { return this.#callHostModule(name, args, { signal: abort.signal, deadline, callId, + interrupt: (payload) => { + if (interrupted) { + return Promise.reject( + new Error("ctx.interrupt() can only be called once per host module call."), + ); + } + interrupted = true; + return this.#interrupt(step, abort, payload); + }, access: this.#capability.access, resolvePath: (path, options) => this.#capability.resolveConfined(path, options?.allowMissing ?? false), @@ -258,6 +284,8 @@ export class WorkspaceRuntimeBridge extends RpcTarget { } }, this.#maxPayloadBytes); const call = withDeadline(operation, this.#maxCallDurationMs, this.#maxPayloadBytes, () => { + // A call waiting for an answer isn't late: the run is released. + if (this.#waiting.has(abort)) return; this.#callTimedOut = true; abort.abort(new Error("Workspace capability call timed out.")); }); @@ -274,7 +302,7 @@ export class WorkspaceRuntimeBridge extends RpcTarget { // Errors count against the response budget too. The budget error // itself is short and not counted, and `maxCalls` bounds how many // the isolate can see. - return call + const delivered = call .then((response) => { if (isFetch) return "result" in response ? { result: response.result } : response; const bytes = "result" in response ? response.bytes : errorBytes(response); @@ -293,6 +321,8 @@ export class WorkspaceRuntimeBridge extends RpcTarget { return "result" in response ? { result: response.result } : response; }) .then((response) => { + // An interrupted call answers nothing: it runs again once answered. + if (this.#waiting.has(abort)) return new Promise(() => undefined); // Write before deliver: the journal holds the response before the // isolate can act on it. A filesystem change is synchronous SQL on // the same database, with no I/O between it and here, so it and @@ -300,6 +330,60 @@ export class WorkspaceRuntimeBridge extends RpcTarget { if (step?.kind !== "live" || this.#journal === undefined) return response; return { ...response, ...this.#journal.commit(step.seq, response) }; }); + this.#responses.set(delivered, abort); + void delivered.finally(() => this.#responses.delete(delivered)).catch(() => undefined); + return delivered; + } + + #interrupt( + step: ReturnType | undefined, + abort: AbortController, + payload: unknown, + ): Promise { + if (step?.kind !== "live" || this.#journal === undefined) { + return Promise.reject( + new Error( + 'ctx.interrupt() needs a replayable run. Run the execution with onInterrupt: "replay", or make that the backend\'s default.', + ), + ); + } + if (step.answer !== undefined) return Promise.resolve(step.answer.value as never); + if (this.#cancelled) + return Promise.reject(new Error("Workspace execution is being cancelled.")); + if (!this.#waiting.has(abort)) { + let since: number; + try { + since = this.#journal.interrupt(step.seq, payload); + } catch (error) { + return Promise.reject(error); + } + this.#waiting.add(abort); + this.#waited?.(); + this.#onWait?.({ seq: step.seq, payload, since }); + } + // Settles only when the run is released and the call aborted. + return new Promise((_, reject) => { + abort.signal.addEventListener("abort", () => reject(abort.signal.reason), { once: true }); + }); + } + + /** + * Hold new calls and wait until each one in flight has been answered + * and journaled, or is waiting itself. The run can then be released + * without any call running twice when it resumes. + */ + async quiesce(): Promise { + this.#holding = true; + while (true) { + const answering = [...this.#responses].filter(([, abort]) => !this.#waiting.has(abort)); + if (answering.length === 0) return; + // Look again once one is answered or starts waiting. + await new Promise((resolve) => { + this.#waited = resolve; + const wake = () => resolve(); + void Promise.race(answering.map(([response]) => response)).then(wake, wake); + }); + } } // The isolate is waiting for a journaled call its replay never made. @@ -539,10 +623,17 @@ export interface BridgeJournal { readonly at: number; readonly ordinal: number; } - | { readonly kind: "live"; readonly seq: number; readonly rerun: boolean } + | { + readonly kind: "live"; + readonly seq: number; + readonly rerun: boolean; + readonly answer?: { readonly value: unknown }; + } | { readonly kind: "diverge"; readonly message: string }; /** Record a live call's response, and return its time and ordinal. */ commit(seq: number, response: BridgeResponse): { at: number; ordinal: number }; + /** Mark a live call as waiting for an answer, and return when it started. */ + interrupt(seq: number, payload: unknown): number; /** Save a chunk of the isolate's event log. Throws on a malformed one. */ record(position: unknown, events: unknown): void; /** Called when a call doesn't match the journal. */ diff --git a/packages/computer/src/runtime/runtime.ts b/packages/computer/src/runtime/runtime.ts index ef4acb5b5..1a3869ffd 100644 --- a/packages/computer/src/runtime/runtime.ts +++ b/packages/computer/src/runtime/runtime.ts @@ -14,7 +14,9 @@ import { type WorkspaceRuntimeExecOptions, type WorkspaceRuntimeGetOptions, type WorkspaceRuntimeKillOptions, + type WorkspaceRuntimePendingCall, type WorkspaceRuntimeResult, + type WorkspaceRuntimeValue, } from "./types.js"; interface WorkspaceRuntimeRouterOptions { @@ -172,6 +174,43 @@ export class WorkspaceRuntime { } } + /** + * Host module calls waiting for an answer, from `ctx.interrupt()`, + * across every backend that supports it, oldest wait first. + * Experimental: it may change in any release. + */ + async pending(): Promise { + const calls: WorkspaceRuntimePendingCall[] = []; + for (const [id, backend] of this.#options.backends) { + if (!isModuleBackend(backend)) continue; + const handle = await this.#options.backendHandle(id); + for (const call of (await handle.pending?.()) ?? []) calls.push({ ...call, backend: id }); + } + return calls.sort((a, b) => a.since - b.since); + } + + /** + * Answer a call waiting in `ctx.interrupt()`. The run resumes once + * none of its calls are still waiting, and the call gets `answer` as + * what `interrupt()` returns. Throws `ENOTWAITING` if the call isn't + * waiting, for example because it was already answered or the run + * was killed. Experimental: it may change in any release. + */ + async respond( + id: string, + callId: string, + answer: WorkspaceRuntimeValue, + options: { backend?: string } = {}, + ): Promise { + assertExecutionId(id); + const backend = this.#backend(options.backend); + const handle = await this.#options.backendHandle(backend); + if (handle.respond === undefined) { + throw new Error(`Backend ${backend} doesn't run calls that wait for an answer.`); + } + await handle.respond({ id, callId, answer }); + } + async killExec(id: string, options: WorkspaceRuntimeKillOptions = {}): Promise { assertExecutionId(id); const backend = this.#backend(options.backend); diff --git a/packages/computer/src/runtime/types.ts b/packages/computer/src/runtime/types.ts index 050873fc1..7c898b883 100644 --- a/packages/computer/src/runtime/types.ts +++ b/packages/computer/src/runtime/types.ts @@ -28,6 +28,28 @@ export interface WorkspaceModuleCallContext { * the same id, so a function can use it as an idempotency key. */ readonly callId: string; + /** + * Pause the run until a person or another system answers, and return + * the answer. The run is released while it waits, survives a deploy, + * and resumes once `workspace.runtime.respond()` answers this call: + * it replays to this call, and this function runs again with the same + * `callId`, this time getting the answer straight away. Anything the + * function does before `interrupt()` therefore runs twice, so keep it + * safe to repeat. Only a run with `onInterrupt: "replay"` can wait; + * anywhere else this throws. A call can interrupt once, and a second + * `interrupt()` throws. Experimental: it may change in any release. + * + * ```ts + * postComment: async ([issue, body], ctx) => { + * const { approved } = await ctx.interrupt({ kind: "approval", issue, body }); + * if (!approved) throw new Error("The comment was rejected."); + * return github.comment(issue, body, { idempotencyKey: ctx.callId }); + * } + * ``` + * + * @param payload What the answer is for, as `pending()` lists it. + */ + interrupt(payload: WorkspaceRuntimeValue): Promise; } /** @@ -283,6 +305,22 @@ export interface ModuleExecutionInput { */ export type WorkspaceRuntimeInterruptPolicy = "fail" | "restart" | "replay"; +/** + * A host module call waiting for an answer, from `ctx.interrupt()`. + * Experimental: it may change in any release. + */ +export interface WorkspaceRuntimePendingCall { + /** The execution the call belongs to. */ + readonly id: string; + readonly backend: string; + /** Pass this to `respond()` to answer the call. */ + readonly callId: string; + /** What the call passed to `ctx.interrupt()`. */ + readonly payload: WorkspaceRuntimeValue; + /** When the call started waiting, in milliseconds since the epoch. */ + readonly since: number; +} + export interface ModuleExecutionEnvelope { id: string; // Identity of the process-local backend runtime that owns this @@ -313,6 +351,9 @@ export interface WorkspaceModuleBackendHandle { // Settle the runs a restart interrupted. A backend that recovers // when it connects needs nothing more here. recover?(): Promise; + // Calls waiting for an answer, and answering one. + pending?(): Promise[]>; + respond?(input: { id: string; callId: string; answer: WorkspaceRuntimeValue }): Promise; // Tear down a backend-owned transport. The command adapter omits // it: a command backend's transport is closed through its // BackendHandle, not through the adapter the runtime consumes. diff --git a/packages/computer/src/stub.test.ts b/packages/computer/src/stub.test.ts index 6f847ff0c..b8f17469d 100644 --- a/packages/computer/src/stub.test.ts +++ b/packages/computer/src/stub.test.ts @@ -13,7 +13,7 @@ import { enableStubTracking, stubSnapshot } from "@cloudflare/computer-rpc/debug"; import { SQLiteTestStorage } from "@cloudflare/dofs/testing"; -import { beforeAll, describe, expect, it } from "vitest"; +import { beforeAll, describe, expect, it, vi } from "vitest"; import type { BackendHandle, WorkspaceBackend } from "./backend.js"; import { createGitClient } from "./git/index.js"; @@ -357,6 +357,17 @@ describe("WorkspaceStub", () => { } }); + it("passes runtime.pending and runtime.respond through", async () => { + await withStub(async (ws) => { + const call = { id: "run", backend: "b", callId: "run:1", payload: { kind: "x" }, since: 1 }; + vi.spyOn(ws.runtime, "pending").mockResolvedValue([call]); + const respond = vi.spyOn(ws.runtime, "respond").mockResolvedValue(); + await expect(ws.stub().runtime.pending()).resolves.toEqual([call]); + await ws.stub().runtime.respond("run", "run:1", { ok: true }, { backend: "b" }); + expect(respond).toHaveBeenCalledWith("run", "run:1", { ok: true }, { backend: "b" }); + }); + }); + it("runtime.exec rejects when a backend ignores the requested id", async () => { const shellRpc: import("@cloudflare/computer-rpc").ShellRPC = { async exec() { diff --git a/packages/computer/src/stub.ts b/packages/computer/src/stub.ts index a49ad8ed2..58dcf505b 100644 --- a/packages/computer/src/stub.ts +++ b/packages/computer/src/stub.ts @@ -75,7 +75,9 @@ import type { WorkspaceRuntimeExecOptions, WorkspaceRuntimeGetOptions, WorkspaceRuntimeKillOptions, + WorkspaceRuntimePendingCall, WorkspaceRuntimeResult, + WorkspaceRuntimeValue, } from "./runtime/types.js"; import { encodeRuntimeEvent } from "./runtime/wire.js"; import type { Workspace } from "./workspace.js"; @@ -470,6 +472,20 @@ export class WorkspaceRuntimeStub extends RpcTarget { disposeExec(id: string, options: { backend?: string } = {}): Promise { return this.#ws.runtime.disposeExec(id, options); } + + // Plain data both ways, so nothing here needs disposing. + pending(): Promise { + return this.#ws.runtime.pending(); + } + + respond( + id: string, + callId: string, + answer: WorkspaceRuntimeValue, + options: { backend?: string } = {}, + ): Promise { + return this.#ws.runtime.respond(id, callId, answer, options); + } } // Assets half. Pure value returns: `publish(path, { expiresAfter }) diff --git a/packages/computer/src/tools/ai-sdk/bindings.test.ts b/packages/computer/src/tools/ai-sdk/bindings.test.ts index 233fe7c69..5fd1e0fd4 100644 --- a/packages/computer/src/tools/ai-sdk/bindings.test.ts +++ b/packages/computer/src/tools/ai-sdk/bindings.test.ts @@ -15,6 +15,7 @@ function callContext(): WorkspaceModuleCallContext { deadline: Date.now() + 60_000, access: "read-write", callId: "test-call", + interrupt: () => Promise.reject(new Error("not used")), resolvePath: async (path) => path, }; } diff --git a/packages/computer/src/tools/pi-ai/bindings.test.ts b/packages/computer/src/tools/pi-ai/bindings.test.ts index 3894c1d33..93f679a65 100644 --- a/packages/computer/src/tools/pi-ai/bindings.test.ts +++ b/packages/computer/src/tools/pi-ai/bindings.test.ts @@ -18,6 +18,7 @@ function callContext( deadline: Date.now() + 60_000, access, callId: "test-call", + interrupt: () => Promise.reject(new Error("not used")), resolvePath: async (path) => path, }; } diff --git a/packages/computer/tests/script-runner-worker.ts b/packages/computer/tests/script-runner-worker.ts index 7cc3dc370..b6db9afba 100644 --- a/packages/computer/tests/script-runner-worker.ts +++ b/packages/computer/tests/script-runner-worker.ts @@ -6,7 +6,11 @@ import { type WorkerJavaScriptBackendOptions, } from "../src/backends/worker-javascript/index.js"; import { createGitClient } from "../src/git/index.js"; -import type { WorkspaceRuntimeValue, WorkspaceStub } from "../src/index.js"; +import type { + WorkspaceRuntimePendingCall, + WorkspaceRuntimeValue, + WorkspaceStub, +} from "../src/index.js"; import { unsafeToRepeat, Workspace } from "../src/index.js"; import { createArtifactsModule } from "../src/modules/artifacts.js"; import { createContainerModule } from "../src/modules/container.js"; @@ -61,6 +65,8 @@ function fakeContainerBackend(): WorkspaceBackend { // state outlives the object, so it counts host calls across the restart // and holds the gate a run waits on until the object restarts. const durableCalls = new Map(); +// Each call onWait was told about. +const waits: WorkspaceRuntimePendingCall[] = []; let gateOpen = false; // Named gates a run can wait on after the first, for a second restart. const opened = new Set(); @@ -90,6 +96,7 @@ function javascriptBackend( ...overrides, maxStdioBytes: overrides.maxStdioBytes ?? 64, maxCapabilityBytes: 1024, + onWait: (call) => void waits.push(call), maxConcurrentCapabilityCalls: 2, modules: { "math-kit": "export const double = (value) => value * 2;", @@ -149,6 +156,33 @@ function javascriptBackend( await new Promise((resolve) => setTimeout(resolve, Number(args[0]))); return countCall(String(args[1] ?? "wait")); }, + // Counts when it starts and again when it settles, and stops early + // if its call is aborted. + async slowTick(args, ctx) { + countCall(`start:${String(args[1])}`); + await new Promise((resolve, reject) => { + const timer = setTimeout(resolve, Number(args[0])); + ctx.signal.addEventListener("abort", () => { + clearTimeout(timer); + reject(ctx.signal.reason); + }); + }); + return countCall(String(args[1])); + }, + // Waits for an answer, then returns it. + async approve(args, ctx) { + const action = String(args[0]); + countCall(`ask:${action}`); + const answer = await ctx.interrupt({ kind: "approval", action }); + countCall(`answered:${action}`); + return answer; + }, + // Like approve, but awaits first, as a real lookup would. + async approveLater(args, ctx) { + const action = String(args[0]); + await new Promise((resolve) => setTimeout(resolve, 50)); + return ctx.interrupt({ kind: "approval", action }); + }, async marker() { return { __workspace_codec__: { version: 1, type: "bytes", data: [1] }, @@ -281,6 +315,41 @@ export class HostDO extends DurableObject { return handle.id; } + async startWaiting(input: { + source: string; + id: string; + onInterrupt?: "fail" | "restart" | "replay"; + timeoutMs?: number; + }) { + durableCalls.clear(); + waits.length = 0; + const handle = await this.#workspace.runtime.exec(input.source, { + backend: "worker-javascript", + id: input.id, + onInterrupt: input.onInterrupt ?? "replay", + timeoutMs: input.timeoutMs, + }); + void handle.result().catch(() => undefined); + return handle.id; + } + + pending() { + return this.#workspace.runtime.pending(); + } + + waits() { + return waits; + } + + async respond(id: string, callId: string, value: WorkspaceRuntimeValue) { + try { + await this.#workspace.runtime.respond(id, callId, value); + return { ok: true as const }; + } catch (error) { + return { ok: false as const, code: (error as { code?: string }).code }; + } + } + // Restart the object, as a deploy or crash would. restart(open?: string) { gateOpen = true; @@ -544,6 +613,36 @@ export default class extends WorkerEntrypoint { return new Response(null, { status: 204 }); } + if (url.pathname === "/interrupt-start") { + return Response.json({ + id: await stub.startWaiting( + (await request.json()) as { + source: string; + id: string; + onInterrupt?: "fail" | "restart" | "replay"; + timeoutMs?: number; + }, + ), + }); + } + + if (url.pathname === "/interrupt-waits") { + return Response.json(await stub.waits()); + } + + if (url.pathname === "/interrupt-pending") { + return Response.json(await stub.pending()); + } + + if (url.pathname === "/interrupt-respond") { + const body = (await request.json()) as { + id: string; + callId: string; + value: WorkspaceRuntimeValue; + }; + return Response.json(await stub.respond(body.id, body.callId, body.value)); + } + if (url.pathname === "/durable-calls") { return Response.json(await stub.durableCalls()); } diff --git a/packages/computer/tests/script-runner.test.ts b/packages/computer/tests/script-runner.test.ts index 56dc8bb5b..44eb96774 100644 --- a/packages/computer/tests/script-runner.test.ts +++ b/packages/computer/tests/script-runner.test.ts @@ -330,6 +330,8 @@ describe("WorkspaceRuntime", () => { const text = await response.text(); expect(response.status, text).toBe(200); expect(JSON.parse(text).result.value).toEqual([ + "approve", + "approveLater", "delete", "echo", "gate", @@ -338,6 +340,7 @@ describe("WorkspaceRuntime", () => { "largeError", "marker", "slow", + "slowTick", "sum", "tick", "unsafeGate", @@ -1341,3 +1344,224 @@ describe("durable executions", () => { expect(await result("durable-recover")).toMatchObject({ status: "completed" }); }); }); + +describe("interrupts", () => { + type Pending = { id: string; backend: string; callId: string; payload: unknown; since: number }; + + async function start(id: string, source: string, options: Record = {}) { + const response = await SELF.fetch("https://example.test/interrupt-start", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ id, source, ...options }), + }); + expect(response.status, await response.clone().text()).toBe(200); + } + + async function pending(id: string): Promise { + const response = await SELF.fetch("https://example.test/interrupt-pending"); + return ((await response.json()) as Pending[]).filter((call) => call.id === id); + } + + // The first of a run's calls to wait, other than `skip`, once one does. + async function waitingCall(id: string, skip?: string): Promise { + for (let attempt = 0; attempt < 200; attempt += 1) { + const call = (await pending(id)).find((item) => item.callId !== skip); + if (call !== undefined) return call; + await new Promise((resolve) => setTimeout(resolve, 20)); + } + throw new Error(`no call of ${id} is waiting`); + } + + async function respond(id: string, callId: string, value: unknown) { + const response = await SELF.fetch("https://example.test/interrupt-respond", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ id, callId, value }), + }); + return (await response.json()) as { ok: boolean; code?: string }; + } + + async function result(id: string) { + const response = await SELF.fetch( + `https://example.test/runtime-get?id=${encodeURIComponent(id)}`, + ); + return (await response.json()) as { + status: string; + stdout: string; + stderr: string; + value?: unknown; + }; + } + + async function calls() { + return (await (await SELF.fetch("https://example.test/durable-calls")).json()) as Record< + string, + number + >; + } + + const approval = ` + import { approve, tick } from "ws:test-host"; + export default async function () { + console.log("before", await tick("a")); + const answer = await approve("deploy"); + console.log("after", answer.ok); + return answer; + } + `; + + it("waits for an answer, then resumes where it was", async () => { + await start("wait-answer", approval); + const call = await waitingCall("wait-answer"); + expect(call).toMatchObject({ + backend: "worker-javascript", + callId: expect.stringMatching(/^wait-answer:\d+$/), + payload: { kind: "approval", action: "deploy" }, + }); + await expect(respond("wait-answer", call.callId, { ok: true })).resolves.toEqual({ ok: true }); + expect(await result("wait-answer")).toMatchObject({ + status: "completed", + stdout: "before 1\nafter true\n", + value: { ok: true }, + }); + // The function ran again to get its answer; nothing else repeated. + expect(await calls()).toEqual({ a: 1, "ask:deploy": 2, "answered:deploy": 1 }); + expect(await pending("wait-answer")).toEqual([]); + await expect(respond("wait-answer", call.callId, { ok: true })).resolves.toEqual({ + ok: false, + code: "ENOTWAITING", + }); + }); + + it("keeps waiting across a restart", async () => { + await start("wait-restart", approval); + await waitingCall("wait-restart"); + await SELF.fetch("https://example.test/durable-restart", { method: "POST" }); + const call = await waitingCall("wait-restart"); + expect(call.payload).toEqual({ kind: "approval", action: "deploy" }); + await respond("wait-restart", call.callId, { ok: false }); + expect(await result("wait-restart")).toMatchObject({ + status: "completed", + stdout: "before 1\nafter false\n", + }); + expect(await calls()).toEqual({ a: 1, "ask:deploy": 2, "answered:deploy": 1 }); + }); + + it("asks calls made together one after another", async () => { + await start( + "wait-both", + ` + import { approve } from "ws:test-host"; + export default async () => (await Promise.all([approve("one"), approve("two")])).map((a) => a.v); + `, + ); + // The second call reaches the host once the run is being released, + // so it waits until the first is answered and the run resumes. + const first = await waitingCall("wait-both"); + expect(first.payload).toEqual({ kind: "approval", action: "one" }); + await respond("wait-both", first.callId, { v: "first" }); + const second = await waitingCall("wait-both", first.callId); + expect(second.payload).toEqual({ kind: "approval", action: "two" }); + await respond("wait-both", second.callId, { v: "second" }); + expect(await result("wait-both")).toMatchObject({ + status: "completed", + value: ["first", "second"], + }); + expect(await calls()).toEqual({ + "ask:one": 2, + "answered:one": 1, + "ask:two": 2, + "answered:two": 1, + }); + }); + + it("waits on calls in flight that interrupt after the first", async () => { + await start( + "wait-later", + ` + import { approveLater } from "ws:test-host"; + export default async () => Promise.all([approveLater("one"), approveLater("two")]); + `, + ); + const first = await waitingCall("wait-later"); + const second = await waitingCall("wait-later", first.callId); + await respond("wait-later", first.callId, 1); + await respond("wait-later", second.callId, 2); + expect(await result("wait-later")).toMatchObject({ status: "completed", value: [1, 2] }); + }); + + it("lets other calls in flight finish before it releases the run", async () => { + await start( + "wait-inflight", + ` + import { approve, slowTick } from "ws:test-host"; + // slowTick reaches the host first, so it's in flight when approve interrupts. + export default async () => Promise.all([slowTick(300, "slow"), approve("x")]); + `, + ); + const call = await waitingCall("wait-inflight"); + await respond("wait-inflight", call.callId, { ok: true }); + expect(await result("wait-inflight")).toMatchObject({ + status: "completed", + value: [1, { ok: true }], + }); + // Journaled when it finished, so the resumed run didn't start it again. + expect(await calls()).toMatchObject({ "start:slow": 1, slow: 1 }); + }); + + it("resumes when the answer arrives before the run is released", async () => { + await start( + "wait-early", + ` + import { approve, slowTick } from "ws:test-host"; + export default async () => Promise.all([slowTick(1000, "early"), approve("x")]); + `, + ); + const call = await waitingCall("wait-early"); + await respond("wait-early", call.callId, { ok: true }); + // slowTick is still running, so the run hadn't been released yet. + expect(await calls()).not.toHaveProperty("early"); + expect(await result("wait-early")).toMatchObject({ + status: "completed", + value: [1, { ok: true }], + }); + }); + + it("doesn't count waiting toward the run's timeout", async () => { + await start("wait-timeout", approval, { timeoutMs: 1500 }); + const call = await waitingCall("wait-timeout"); + await new Promise((resolve) => setTimeout(resolve, 2000)); + await respond("wait-timeout", call.callId, { ok: true }); + expect(await result("wait-timeout")).toMatchObject({ status: "completed" }); + }); + + it("tells onWait about a wait once, even across a restart and resume", async () => { + await start("wait-notify", approval); + const call = await waitingCall("wait-notify"); + await SELF.fetch("https://example.test/durable-restart", { method: "POST" }); + await waitingCall("wait-notify"); + await respond("wait-notify", call.callId, { ok: true }); + expect(await result("wait-notify")).toMatchObject({ status: "completed" }); + const waits = await (await SELF.fetch("https://example.test/interrupt-waits")).json(); + expect(waits).toEqual([call]); + }); + + it("can be killed while it waits", async () => { + await start("wait-kill", approval); + const call = await waitingCall("wait-kill"); + await SELF.fetch("https://example.test/runtime-kill?id=wait-kill", { method: "POST" }); + expect(await result("wait-kill")).toMatchObject({ status: "cancelled" }); + expect(await pending("wait-kill")).toEqual([]); + await expect(respond("wait-kill", call.callId, { ok: true })).resolves.toMatchObject({ + code: "ENOTWAITING", + }); + }); + + it("throws in a run that can't be released", async () => { + await start("wait-fail", approval, { onInterrupt: "fail" }); + const settled = await result("wait-fail"); + expect(settled.status).toBe("failed"); + // The test backend keeps 64 bytes of standard error. + expect(settled.stderr).toContain("ctx.interrupt() needs a replayable run"); + }); +});