diff --git a/.changeset/durable-javascript.md b/.changeset/durable-javascript.md new file mode 100644 index 00000000..d798a293 --- /dev/null +++ b/.changeset/durable-javascript.md @@ -0,0 +1,5 @@ +--- +"@cloudflare/computer": minor +--- + +Add experimental `onInterrupt: "restart" | "replay"` to `WorkerJavaScriptBackend`, so runs interrupted by a deploy or crash can resume. Also adds `runtime.recover()`, `unsafeToRepeat()` and `callId` on host module call contexts. diff --git a/docs/17_isolate_javascript.md b/docs/17_isolate_javascript.md index 0bf86598..14b4a1cd 100644 --- a/docs/17_isolate_javascript.md +++ b/docs/17_isolate_javascript.md @@ -143,7 +143,60 @@ A run also needs its Durable Object to stay in memory, and that depends on the h CPU time is limited separately from wall-clock time. Each run's CPU limit is the smaller of its timeout and `maxCpuMs`, which defaults to `maxTimeoutMs`. CPU time adds up over the whole run, however much of it is spent waiting, and Cloudflare currently stops a Dynamic Worker call at about five minutes of CPU whatever the limit. A run that mostly waits on I/O uses little of that and can last hours. A run that computes heavily fails with `Worker exceeded CPU time limit.` once it reaches the cap. Every execution gets a fresh Dynamic Worker and a fresh CPU budget, so heavy work can be split across executions or moved to a container. The platform's subrequest limit, 10,000 by default on the Workers Paid plan, also counts every `fetch` over the whole run. A long run also holds one of the ten Dynamic Workers a Durable Object can have in flight for as long as it lasts. -Nothing keeps a run going across a deploy or a crash. A deploy shuts the old instance down a few minutes later, and a run that was in flight is marked as failed the next time the backend connects. +A deploy or a crash still ends a run. A deploy shuts the old instance down a few minutes later. What happens next depends on `onInterrupt`, described below. + +## Interrupted executions + +A run that was in flight when its Durable Object restarted is settled by its `onInterrupt` policy. Set a default on the backend and override it per run. The `"restart"` and `"replay"` policies are [experimental](../packages/computer/README.md#status): their behavior and stored format may change in any release, and a run interrupted across such an upgrade fails rather than resuming. + +```ts +const backend = new WorkerJavaScriptBackend({ loader: env.LOADER, onInterrupt: "replay" }); + +await workspace.runtime.exec(source, { onInterrupt: "fail" }); +``` + +| Policy | After a restart | +|---|---| +| `"fail"` (default) | The run fails with exit 1. | +| `"restart"` | The run starts again from the beginning. Its output so far stays, followed by a note on standard error, so output repeats. Calls it already made run again. | +| `"replay"` | The run starts again from the beginning, but every filesystem and host module call it already made is answered from a journal instead of running again. Output it already produced isn't repeated. Once the journal runs out, the run carries on live. | + +Nothing changes in the code that runs. The backend recovers interrupted runs when it connects after the restart, which the next runtime call or `workspace.ready()` triggers. A caller waiting on a run, by polling it or reconnecting to its output, triggers that itself. Nothing else wakes a Durable Object after a deploy, so a run nobody is waiting on stays interrupted until something calls the object, and fails if its timeout passes first. A host that wants such runs to resume on their own can call `workspace.runtime.recover()` from its alarm. The Workspace never sets an alarm itself, since a Durable Object has only one. + +To make this possible, a run under `"restart"` or `"replay"` saves its module graph, input, environment, working directory, standard input and limits when it starts. A replayed run also journals each call: the host writes the call and its response to SQLite before the isolate sees the response. A filesystem call's change and its journal entry commit in the same transaction, so after a restart either both are there or neither is. The saved data is deleted when the run settles. + +A replay shows the code its events in the order it saw them the first time. While a run is live, the isolate logs the order it sees responses arrive and timers fire in, and sends the log with its next call or output, so the host saves it before anything that came of those events. On replay the host answers journaled calls at once, and the isolate holds each response and timer back until the log says it's its turn, so `Promise.race`, first-done-wins and a timer racing a call go the way they went before. Whatever the run did after the last thing it sent, nobody saw, so the replay doesn't have to repeat it. To keep the rest of the run on the same path, a replayable run gets: + +- A virtual clock. `Date.now()`, `new Date()` and `performance.now()` start at the time the run started, move to the time each journaled response was committed, and move to a timer's due time when it fires. Like the Workers clock, the virtual clock doesn't move while code computes. +- Replayable timers. `setTimeout`, `setInterval`, `scheduler.wait()` and `AbortSignal.timeout()` are logged like responses. While a run replays, they fire as soon as their turn comes, so it doesn't wait out sleeps it already waited. When the replay ends, timers still pending wait out the rest of their delay. +- Seeded random numbers. `Math.random()`, `crypto.getRandomValues()` and `crypto.randomUUID()` draw from a ChaCha20 stream over a 256-bit seed saved with the run, so they repeat on replay and stay unpredictable to anyone without the seed. +- Journaled `fetch`. The isolate's `fetch` goes through the host, which makes the request through the backend's egress and journals the response, so a replay doesn't send it again. + +A replay has to make the same calls with the same arguments as the first attempt, and print the same output. A call the journal doesn't hold, while journaled calls are still unanswered, output that differs from what was recorded, or a run that finishes or stops before making the calls it made before, fails the run with a message saying it took a different path. + +A host module call still in flight at the restart never delivered its response, so it runs again on replay. Every call gets a `context.callId` that stays the same when it runs again, so a function can use it as an idempotency key. A function that can't deduplicate and mustn't run twice, such as one that charges a card, can be marked with `unsafeToRepeat()`. An interruption during a marked call fails the run instead: + +```ts +import { unsafeToRepeat } from "@cloudflare/computer"; + +modules: { + "ws:billing": { + charge: unsafeToRepeat(async ([order, cents]) => billing.charge(order, cents)), + refund: async ([order], { callId }) => billing.refund(order, { idempotencyKey: callId }), + }, +} +``` + +Some limits apply: + +- A run gets three attempts. An interruption after the third fails it. +- A run resumes only within its original timeout, counted from when it first started. +- A run resumes only on the same runtime code. A deploy that changes the backend's generated modules, compatibility date or flags fails the runs it interrupted rather than replaying them on different code. +- After a restart, interrupted runs start four at a time, since a Durable Object runs at most ten Dynamic Workers at once. The rest wait their turn within their timeouts. +- A `fetch` in flight at the restart is sent again, like a host module call. +- A replayable run's `fetch` is buffered rather than streamed: each request and response body must fit within `maxCapabilityBytes`, 1 MiB by default. It doesn't count toward `maxCapabilityCalls` or the capability byte budgets, and at most six are open at once, with the rest waiting their turn, as in any Worker. Every response stays in the journal until the run settles. +- Some sources of nondeterminism aren't covered: keys from `crypto.subtle.generateKey()`, `node:crypto` randomness, `node:timers`, and TCP sockets from `cloudflare:sockets`. A run that uses them for anything it passes to a call or prints diverges on replay. +- The journal counts toward the Durable Object's storage while the run lasts: roughly the size of every call's response, plus the saved module graph. ## Environment, standard input, and the `process` shim diff --git a/examples/rlm/worker/capability.test.ts b/examples/rlm/worker/capability.test.ts index d0838100..b17848eb 100644 --- a/examples/rlm/worker/capability.test.ts +++ b/examples/rlm/worker/capability.test.ts @@ -39,6 +39,7 @@ function context(signal = new AbortController().signal): WorkspaceModuleCallCont signal, deadline: Date.now() + 1_000, access: "read", + callId: "test-call", resolvePath: async (path) => path, }; } diff --git a/packages/computer/README.md b/packages/computer/README.md index d5539d43..fb518a2e 100644 --- a/packages/computer/README.md +++ b/packages/computer/README.md @@ -31,6 +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 | | `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/durable-runtime.test.ts b/packages/computer/src/backends/worker-javascript/durable-runtime.test.ts new file mode 100644 index 00000000..54ccd867 --- /dev/null +++ b/packages/computer/src/backends/worker-javascript/durable-runtime.test.ts @@ -0,0 +1,259 @@ +import vm from "node:vm"; + +import { describe, expect, it, vi } from "vitest"; + +import { durableRuntimeModule } from "./durable-runtime.js"; + +interface Payload { + result?: unknown; + at?: number; + ordinal?: number; +} + +interface DurableExports { + install( + durable: { seed: string; startedAt: number; schedule: number[] }, + host: { reportStall(): unknown; recordLog(log: unknown): unknown }, + call: (...args: unknown[]) => Promise, + ): void; + request(): void; + failed(): void; + deliver(payload: Payload): Promise | Payload; + takeLog(): { from: number; events: number[] } | undefined; +} + +// Load the module into its own realm, as the isolate would, so the +// globals it replaces aren't this process's. +function load( + schedule: number[], + seed = "0".repeat(64), + call: (...args: unknown[]) => Promise = async () => null, +) { + const context = vm.createContext({ + setTimeout, + clearTimeout, + crypto: {}, + performance: {}, + AbortSignal: {}, + AbortController, + scheduler: { wait: () => undefined }, + DOMException, + Request, + Response, + }); + const source = durableRuntimeModule().replace(/export function/g, "function"); + const durable = vm.runInContext( + `${source}\n({ install, request, failed, deliver, takeLog });`, + context, + ) as DurableExports; + const reportStall = vi.fn<() => void>(); + const recordLog = vi.fn<(log: unknown) => void>(); + durable.install({ seed, startedAt: 1000, schedule }, { reportStall, recordLog }, call); + const run = (code: string) => vm.runInContext(code, context) as T; + return { durable, reportStall, recordLog, run, context }; +} + +const settle = (ms = 20) => new Promise((resolve) => setTimeout(resolve, ms)); + +function track(durable: DurableExports, payload: Payload, seen: unknown[]) { + durable.request(); + void Promise.resolve(durable.deliver(payload)).then((value) => seen.push(value.result)); +} + +describe("durable runtime", () => { + it("releases replayed responses in the order the log gives", async () => { + const { durable, run } = load([2, 1]); + const seen: unknown[] = []; + track(durable, { result: "a", ordinal: 1, at: 1010 }, seen); + track(durable, { result: "b", ordinal: 2, at: 1020 }, seen); + await settle(); + expect(seen).toEqual(["b", "a"]); + expect(run("Date.now()")).toBe(1020); + }); + + it("fires a timer where the log placed it, whatever its delay", async () => { + // Set first, so it's timer 1. It fired after both responses, as a + // timer does when the code computes past its due time. + const { durable, run } = load([1, 2, -1]); + const seen: unknown[] = []; + run("globalThis.fired = []; setTimeout(() => fired.push(Date.now()), 0)"); + track(durable, { result: "a", ordinal: 1, at: 1010 }, seen); + track(durable, { result: "b", ordinal: 2, at: 1100 }, seen); + await settle(); + expect(seen).toEqual(["a", "b"]); + expect(run("fired")).toEqual([1100]); + }); + + it("fires a timer between the responses the log put it between", async () => { + const { durable, run } = load([1, -1, 2]); + const seen: unknown[] = []; + run("globalThis.order = []; setTimeout(() => order.push('timer'), 500)"); + durable.request(); + void Promise.resolve(durable.deliver({ result: "a", ordinal: 1, at: 1010 })).then(() => + run("order.push('a')"), + ); + track(durable, { result: "b", ordinal: 2, at: 1020 }, seen); + await settle(); + expect(run("order")).toEqual(["a", "timer"]); + expect(seen).toEqual(["b"]); + // The timer's due time is past the second response's, and the clock + // never runs backwards. + expect(run("Date.now()")).toBe(1500); + }); + + it("holds a response the log doesn't place until the replay is done", async () => { + const { durable } = load([1]); + const seen: unknown[] = []; + track(durable, { result: "live", ordinal: 5, at: 1005 }, seen); + await settle(); + expect(seen).toEqual([]); + track(durable, { result: "replayed", ordinal: 1, at: 1010 }, seen); + await settle(); + expect(seen).toEqual(["replayed", "live"]); + }); + + it("passes an unjournaled response straight through", async () => { + const { durable } = load([1]); + const seen: unknown[] = []; + track(durable, { result: "refused" }, seen); + await settle(); + expect(seen).toEqual(["refused"]); + }); + + it("logs the order a live run sees its events in", async () => { + const { durable, run } = load([]); + run("setTimeout(() => {}, 10)"); + track(durable, { result: "a", ordinal: 3, at: 1001 }, []); + await settle(40); + expect(durable.takeLog()).toEqual({ from: 0, events: [3, -1] }); + expect(durable.takeLog()).toBeUndefined(); + }); + + it("logs new events after the ones it replayed", async () => { + const { durable } = load([1]); + track(durable, { result: "a", ordinal: 1, at: 1010 }, []); + track(durable, { result: "b", ordinal: 2, at: 1020 }, []); + await settle(); + expect(durable.takeLog()).toEqual({ from: 1, events: [2] }); + }); + + it("sends a long log on its own rather than holding it for the next call", async () => { + const { durable, recordLog, run } = load([]); + run("for (let i = 0; i < 5000; i++) setTimeout(() => {}, 0)"); + await settle(100); + expect(recordLog).toHaveBeenCalledOnce(); + const [chunk] = recordLog.mock.calls[0] as [{ from: number; events: number[] }]; + expect(chunk.from).toBe(0); + expect(chunk.events).toHaveLength(4096); + expect(durable.takeLog()).toMatchObject({ from: 4096 }); + }); + + it("gives timers left after the replay their remaining real delay", async () => { + const { durable, run } = load([1]); + run("globalThis.fired = []; setTimeout(() => fired.push(Date.now()), 40)"); + track(durable, { result: "a", ordinal: 1, at: 1010 }, []); + await settle(5); + expect(run("fired")).toEqual([]); + await settle(80); + expect(run("fired")).toEqual([1040]); + expect(durable.takeLog()).toEqual({ from: 1, events: [-1] }); + }); + + it("reports a replay that stops making the calls it made before", async () => { + const { reportStall } = load([1010]); + await settle(1300); + expect(reportStall).toHaveBeenCalledOnce(); + }); + + it("doesn't report a stall while a call is in flight", async () => { + const { durable, reportStall } = load([1010]); + durable.request(); + await settle(1300); + expect(reportStall).not.toHaveBeenCalled(); + }); + + it("draws random bytes from ChaCha20 over the seed", () => { + // RFC 8439 A.1, test vector 1: an all-zero key, nonce and counter. + const { run } = load([]); + expect(run("[...crypto.getRandomValues(new Uint8Array(8))]")).toEqual([ + 0x76, 0xb8, 0xe0, 0xad, 0xa0, 0xf1, 0x3d, 0x90, + ]); + }); + + it("repeats random numbers for the same seed", () => { + const seed = "a1".repeat(32); + const draw = () => load([], seed).run("[Math.random(), crypto.randomUUID()].join(' ')"); + const first = draw(); + expect(draw()).toBe(first); + expect(first).toMatch( + /^0\.\d+ [0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/, + ); + expect(load([], "b2".repeat(32)).run("Math.random()")).not.toBe(Number(first.split(" ")[0])); + }); + + it("hands back Node timer handles when the real timers do", async () => { + // Node's own setTimeout returns Timeout objects, as workerd's does + // under nodejs_compat. + const { run } = load([]); + run(` + globalThis.fired = []; + const cleared = setTimeout(() => fired.push("cleared"), 5); + cleared.unref(); + clearTimeout(cleared); + setTimeout(() => fired.push("refreshed"), 5).refresh(); + globalThis.handle = setTimeout(() => {}, 0); + `); + await settle(); + expect(run("fired")).toEqual(["refreshed"]); + expect(run("handle.hasRef() && typeof Number(handle)")).toBe("number"); + }); + + it("sends fetch through the host with its redirect mode", async () => { + const call = vi.fn<(...args: unknown[]) => Promise>(async () => ({ + status: 302, + statusText: "Found", + headers: [], + body: null, + })); + const { run } = load([], undefined, call); + const status = await run>( + "fetch('https://example.test/a', { redirect: 'manual' }).then((response) => response.status)", + ); + expect(status).toBe(302); + expect(call).toHaveBeenCalledWith("net", "fetch", [ + expect.objectContaining({ url: "https://example.test/a", method: "GET", redirect: "manual" }), + ]); + }); + + it("lets a signal cancel scheduler.wait", async () => { + const { run } = load([]); + run(` + globalThis.waits = []; + const early = new AbortController(); + early.abort(new DOMException("early", "AbortError")); + scheduler.wait(5, { signal: early.signal }).catch((error) => waits.push(error.message)); + const later = new AbortController(); + scheduler.wait(30, { signal: later.signal }).then( + () => waits.push("resolved"), + (error) => waits.push(error.message), + ); + setTimeout(() => later.abort(new DOMException("later", "AbortError")), 5); + `); + await settle(60); + expect(run("waits")).toEqual(["early", "later"]); + }); + + it("keeps Date subclasses working", () => { + const { run } = load([]); + expect( + run("class Day extends Date {}; const day = new Day(); day instanceof Day && day.getTime()"), + ).toBe(1000); + }); + + it("measures performance.now from the run's start", async () => { + const { durable, run } = load([1]); + track(durable, { result: "a", ordinal: 1, at: 1250 }, []); + await settle(); + expect(run("performance.now()")).toBe(250); + }); +}); diff --git a/packages/computer/src/backends/worker-javascript/durable-runtime.ts b/packages/computer/src/backends/worker-javascript/durable-runtime.ts new file mode 100644 index 00000000..34162f9c --- /dev/null +++ b/packages/computer/src/backends/worker-javascript/durable-runtime.ts @@ -0,0 +1,431 @@ +// The isolate half of replayable runs, as module source. +// +// A replay has to show the code its events in the order it saw them the +// first time: the same responses, the same timer firings between them, +// the same clock readings and the same random numbers. While a run is +// live, this module logs each event as the code sees it: a response by +// the ordinal the host gave it, a timer by the order it was set in. The +// log rides along with the next capability call or output frame, so the +// host saves it before anything that came of those events. Whatever the +// code did after the last thing it sent, nobody saw, so a replay is free +// to order it differently. +// +// On replay the host answers journaled calls at once, and this module +// holds each response back, and each timer, until the log says it's its +// turn. Once the log runs out, the run carries on live: responses nobody +// saw the first time are released, and pending timers wait out the rest +// of their delay. +// +// The clock is virtual. It starts when the run did, moves to the time +// each journaled response was committed and to a timer's due time when it +// fires, and doesn't move otherwise, like the Workers clock between I/O. +// Random numbers come from a ChaCha20 stream over the run's seed. `fetch` +// goes through the host, which journals it like any other call. +// +// Nothing here runs unless the runner installs it for a replayable run. + +export const DURABLE_MODULE = "workspace-durable.js"; + +export function durableRuntimeModule(): string { + return ` + let active = false; + let host; + let virtualNow = 0; + let startedAt = 0; + // The order the run saw its events in before it was interrupted, and + // how far the replay has followed it. + let schedule = []; + let cursor = 0; + // Ordinals of the responses the rest of the schedule still places. + const expected = new Set(); + // Events since the log was last sent, and the position of the first. + let pending = []; + let logged = 0; + // A run that sets many timers between calls and output sends its log + // on its own once this many events are waiting. + const LOG_FLUSH = 4096; + const received = new Map(); + // Responses the schedule doesn't place, released once it's done. + const held = []; + let inFlight = 0; + const timers = new Map(); + let nextTimerId = 1; + // Node's global timers hand back Timeout objects rather than numbers. + // When the real ones do, these do too, mapped back to their timer. + let nodeHandles = false; + const handleIds = new WeakMap(); + let realSetTimeout; + let realClearTimeout; + let pumpScheduled = false; + let stallCheck; + let stalledSince; + let stallReported = false; + // How long the replay may wait for an event it saw before, with + // nothing else left to do, before it counts as a different path. + const STALL_MS = 1000; + + const replaying = () => cursor < schedule.length; + const advance = (time) => { + if (typeof time === "number" && time > virtualNow) virtualNow = time; + }; + + export function install(durable, bridge, call) { + active = true; + host = bridge; + startedAt = durable.startedAt; + virtualNow = startedAt; + schedule = durable.schedule || []; + logged = schedule.length; + for (const event of schedule) if (event > 0) expected.add(event); + realSetTimeout = globalThis.setTimeout; + realClearTimeout = globalThis.clearTimeout; + installRandom(durable.seed); + installClock(); + installTimers(); + installFetch(call); + if (replaying()) schedulePump(); + } + + function logEvent(event) { + pending.push(event); + if (pending.length >= LOG_FLUSH) { + Promise.resolve(host.recordLog(takeLog())).catch(() => {}); + } + } + + // The events logged since the last call, for the host to save. + export function takeLog() { + if (!active || pending.length === 0) return undefined; + const chunk = { from: logged, events: pending }; + logged += pending.length; + pending = []; + return chunk; + } + + // Called as a capability call leaves for the host. + export function request() { + if (active) inFlight += 1; + } + + // Called when a capability call failed without a response. + export function failed() { + if (active) inFlight -= 1; + } + + // Resolves with the response once the code may see it. + export function deliver(payload) { + if (!active) return payload; + inFlight -= 1; + return new Promise((resolve) => { + const ordinal = payload.ordinal; + const release = (replayed) => { + advance(payload.at); + if (!replayed && typeof ordinal === "number") logEvent(ordinal); + resolve(payload); + }; + // Not journaled, such as a limit the bridge refused, so it goes + // straight back as it did the first time. + if (typeof ordinal !== "number" || !replaying()) return release(false); + if (expected.has(ordinal)) { + received.set(ordinal, release); + schedulePump(); + } else { + held.push({ ordinal, release }); + } + }); + } + + // One event per turn, so the code reacts to each before the next. + function schedulePump() { + if (pumpScheduled) return; + pumpScheduled = true; + realSetTimeout(pump, 0); + } + + function pump() { + pumpScheduled = false; + if (!replaying()) return; + const event = schedule[cursor]; + if (event > 0 ? received.has(event) : timers.has(-event)) { + stalledSince = undefined; + cursor += 1; + if (event > 0) { + const release = received.get(event); + received.delete(event); + expected.delete(event); + release(true); + } else { + fire(-event, true); + } + if (replaying()) schedulePump(); + else finishReplay(); + return; + } + if (inFlight > 0) { + // The code asked for something; wait for it to come back. + stalledSince = undefined; + return; + } + // The code isn't waiting on the host, and the next event hasn't + // come. Async work the host doesn't see, such as hashing, can still + // be running, so wait a little first. + const now = realNow(); + stalledSince ??= now; + if (now - stalledSince < STALL_MS) { + // Check again soon. Anything that arrives first pumps at once. + stallCheck ??= realSetTimeout(() => { + stallCheck = undefined; + schedulePump(); + }, 50); + return; + } + if (!stallReported) { + stallReported = true; + Promise.resolve(host.reportStall()).catch(() => {}); + } + } + + // The log has run out, so the run goes live. + function finishReplay() { + held.sort((a, b) => a.ordinal - b.ordinal); + for (const { release } of held.splice(0)) release(false); + for (const [id, timer] of timers) { + if (timer.handle === undefined) { + timer.handle = realSetTimeout(() => fire(id, false), Math.max(0, timer.due - virtualNow)); + } + } + } + + function arm(id, timer) { + timer.handle = undefined; + timers.set(id, timer); + // While replaying, a timer fires when the log says it did. + if (replaying()) return schedulePump(); + timer.handle = realSetTimeout(() => fire(id, false), Math.max(0, timer.due - virtualNow)); + } + + function addTimer(callback, delay, args, repeat) { + const id = nextTimerId++; + const step = Math.max(0, Number(delay) || 0); + arm(id, { callback, args, repeat, step, due: virtualNow + step }); + if (!nodeHandles) return id; + const handle = { + ref() { return this; }, + unref() { return this; }, + hasRef() { return true; }, + refresh() { + const timer = timers.get(id); + if (timer !== undefined) { + if (timer.handle !== undefined) realClearTimeout(timer.handle); + arm(id, { ...timer, due: virtualNow + timer.step }); + } + return this; + }, + close() { + cancelTimer(id); + return this; + }, + [Symbol.toPrimitive]: () => id, + }; + handleIds.set(handle, id); + return handle; + } + + function fire(id, replayed) { + const timer = timers.get(id); + if (timer === undefined) return; + timers.delete(id); + advance(timer.due); + if (!replayed) logEvent(-id); + if (timer.repeat) arm(id, { ...timer, due: timer.due + Math.max(timer.step, 1) }); + if (typeof timer.callback !== "function") return; + try { + timer.callback(...timer.args); + } catch (error) { + // Report it as a throwing timer would, without stopping the pump. + realSetTimeout(() => { + throw error; + }, 0); + } + } + + function cancelTimer(handle) { + const id = typeof handle === "object" && handle !== null ? handleIds.get(handle) : handle; + const timer = timers.get(id); + if (timer === undefined) { + if (handle !== undefined) realClearTimeout(handle); + return; + } + timers.delete(id); + if (timer.handle !== undefined) realClearTimeout(timer.handle); + } + + function installTimers() { + const probe = realSetTimeout(() => {}, 0); + realClearTimeout(probe); + nodeHandles = typeof probe === "object"; + globalThis.setTimeout = (callback, delay, ...args) => addTimer(callback, delay, args, false); + globalThis.setInterval = (callback, delay, ...args) => addTimer(callback, delay, args, true); + globalThis.clearTimeout = cancelTimer; + globalThis.clearInterval = cancelTimer; + if (typeof globalThis.setImmediate === "function") { + globalThis.setImmediate = (callback, ...args) => addTimer(callback, 0, args, false); + globalThis.clearImmediate = cancelTimer; + } + if (globalThis.scheduler && typeof globalThis.scheduler.wait === "function") { + globalThis.scheduler.wait = (delay, options) => + new Promise((resolve, reject) => { + const signal = options && options.signal; + if (signal && signal.aborted) return reject(signal.reason); + const timer = addTimer(resolve, delay, [], false); + if (signal) { + signal.addEventListener("abort", () => { + cancelTimer(timer); + reject(signal.reason); + }, { once: true }); + } + }); + } + Object.defineProperty(AbortSignal, "timeout", { + configurable: true, + writable: true, + value(delay) { + const controller = new AbortController(); + addTimer(() => controller.abort(new DOMException("The operation timed out.", "TimeoutError")), delay, [], false); + return controller.signal; + }, + }); + } + + // The real clock, for the stall grace period only. + let RealDate = Date; + const realNow = () => RealDate.now(); + + function installClock() { + RealDate = Date; + function DurableDate(...args) { + if (!new.target) return new RealDate(virtualNow).toString(); + return Reflect.construct(RealDate, args.length === 0 ? [virtualNow] : args, new.target); + } + DurableDate.prototype = RealDate.prototype; + DurableDate.now = () => virtualNow; + DurableDate.parse = RealDate.parse; + DurableDate.UTC = RealDate.UTC; + globalThis.Date = DurableDate; + if (globalThis.performance) { + Object.defineProperty(globalThis.performance, "now", { + configurable: true, + writable: true, + value: () => virtualNow - startedAt, + }); + } + } + + function installRandom(seed) { + const stream = chacha20(seed); + Math.random = () => { + const b = stream(7); + return ((b[0] & 0x1f) * 2 ** 48 + b[1] * 2 ** 40 + b[2] * 2 ** 32 + + b[3] * 2 ** 24 + b[4] * 2 ** 16 + b[5] * 2 ** 8 + b[6]) / 2 ** 53; + }; + Object.defineProperty(crypto, "getRandomValues", { + configurable: true, + writable: true, + value(array) { + if (!ArrayBuffer.isView(array) || array instanceof Float32Array || + array instanceof Float64Array || array instanceof DataView) { + throw new TypeError("getRandomValues takes an integer typed array."); + } + if (array.byteLength > 65536) { + throw new DOMException("getRandomValues takes at most 65536 bytes.", "QuotaExceededError"); + } + new Uint8Array(array.buffer, array.byteOffset, array.byteLength).set(stream(array.byteLength)); + return array; + }, + }); + Object.defineProperty(crypto, "randomUUID", { + configurable: true, + writable: true, + value() { + const b = stream(16); + b[6] = (b[6] & 0x0f) | 0x40; + b[8] = (b[8] & 0x3f) | 0x80; + const hex = [...b].map((byte) => byte.toString(16).padStart(2, "0")).join(""); + return hex.slice(0, 8) + "-" + hex.slice(8, 12) + "-" + hex.slice(12, 16) + "-" + + hex.slice(16, 20) + "-" + hex.slice(20); + }, + }); + } + + // ChaCha20 keystream over a 256-bit key: returns n bytes at a time. + function chacha20(seed) { + const state = new Uint32Array(16); + state.set([0x61707865, 0x3320646e, 0x79622d32, 0x6b206574]); + for (let i = 0; i < 8; i++) state[4 + i] = parseInt(seed.slice(i * 8, i * 8 + 8), 16); + const block = new Uint32Array(16); + const bytes = new Uint8Array(block.buffer); + const x = new Uint32Array(16); + let offset = 64; + const rotl = (v, n) => (v << n) | (v >>> (32 - n)); + const quarter = (a, b, c, d) => { + x[a] += x[b]; x[d] = rotl(x[d] ^ x[a], 16); + x[c] += x[d]; x[b] = rotl(x[b] ^ x[c], 12); + x[a] += x[b]; x[d] = rotl(x[d] ^ x[a], 8); + x[c] += x[d]; x[b] = rotl(x[b] ^ x[c], 7); + }; + const refill = () => { + x.set(state); + for (let round = 0; round < 10; round++) { + quarter(0, 4, 8, 12); quarter(1, 5, 9, 13); quarter(2, 6, 10, 14); quarter(3, 7, 11, 15); + quarter(0, 5, 10, 15); quarter(1, 6, 11, 12); quarter(2, 7, 8, 13); quarter(3, 4, 9, 14); + } + for (let i = 0; i < 16; i++) block[i] = x[i] + state[i]; + state[12] += 1; + if (state[12] === 0) state[13] += 1; + offset = 0; + }; + return (n) => { + const out = new Uint8Array(n); + for (let i = 0; i < n; i++) { + if (offset === 64) refill(); + out[i] = bytes[offset++]; + } + return out; + }; + } + + // fetch goes through the host as a capability call, so it's journaled + // and its response replays like any other. + function installFetch(call) { + const nullBody = new Set([101, 103, 204, 205, 304]); + globalThis.fetch = async (input, init) => { + const request = new Request(input, init); + const signal = init && init.signal ? init.signal : input instanceof Request ? input.signal : undefined; + if (signal && signal.aborted) throw signal.reason; + const body = request.method === "GET" || request.method === "HEAD" + ? null + : new Uint8Array(await request.arrayBuffer()); + const pending = call("net", "fetch", [{ + url: request.url, + method: request.method, + redirect: request.redirect, + headers: [...request.headers], + body, + }]); + const result = await (signal + ? Promise.race([pending, new Promise((_, reject) => { + signal.addEventListener("abort", () => reject(signal.reason), { once: true }); + })]) + : pending); + const response = new Response(nullBody.has(result.status) ? null : result.body, { + status: result.status, + statusText: result.statusText, + headers: result.headers, + }); + Object.defineProperty(response, "url", { value: result.url }); + Object.defineProperty(response, "redirected", { value: result.redirected }); + return response; + }; + } + `; +} diff --git a/packages/computer/src/backends/worker-javascript/frames.test.ts b/packages/computer/src/backends/worker-javascript/frames.test.ts index 64c6fb77..62d5cd5c 100644 --- a/packages/computer/src/backends/worker-javascript/frames.test.ts +++ b/packages/computer/src/backends/worker-javascript/frames.test.ts @@ -51,6 +51,21 @@ describe("parseRuntimeFrame", () => { expect(frame).toEqual({ name: "exit", code: 0, result: null }); }); + it("carries a durable run's event log alongside a frame", () => { + expect( + parseRuntimeFrame( + JSON.stringify({ name: "stdout", b64: "aGk=", log: { from: 2, events: [-1] } }), + ), + ).toEqual({ + name: "stdout", + value: new TextEncoder().encode("hi"), + log: { from: 2, events: [-1] }, + }); + expect(() => parseRuntimeFrame(JSON.stringify({ name: "exit", code: 0, log: "x" }))).toThrow( + "malformed event log", + ); + }); + it("rejects invalid JSON", () => { expect(() => parseRuntimeFrame("not json")).toThrow("received an invalid execution frame"); }); diff --git a/packages/computer/src/backends/worker-javascript/frames.ts b/packages/computer/src/backends/worker-javascript/frames.ts index 6b738238..b0622403 100644 --- a/packages/computer/src/backends/worker-javascript/frames.ts +++ b/packages/computer/src/backends/worker-javascript/frames.ts @@ -1,9 +1,13 @@ import type { WorkspaceRuntimeValue } from "../../runtime/types.js"; -export type RuntimeFrame = +export type RuntimeFrame = ( | { name: "stdout"; value: Uint8Array } | { name: "stderr"; value: Uint8Array } - | { name: "exit"; code: number; result?: WorkspaceRuntimeValue }; + | { name: "exit"; code: number; result?: WorkspaceRuntimeValue } +) & { + // A durable run's events since it last sent any, saved with the frame. + log?: { from: unknown; events: unknown }; +}; export function parseRuntimeFrame(line: string): RuntimeFrame { let record: Record; @@ -12,6 +16,16 @@ export function parseRuntimeFrame(line: string): RuntimeFrame { } catch { throw new Error("WorkerJavaScriptBackend received an invalid execution frame"); } + const frame = parseFrameBody(record); + const log = record.log as { from?: unknown; events?: unknown } | undefined; + if (log === undefined) return frame; + if (log === null || typeof log !== "object") { + throw new Error("WorkerJavaScriptBackend received a malformed event log"); + } + return { ...frame, log: { from: log.from, events: log.events } }; +} + +function parseFrameBody(record: Record): RuntimeFrame { const name = record.name; if (name === "stdout" || name === "stderr") { if (typeof record.b64 !== "string") { diff --git a/packages/computer/src/backends/worker-javascript/journal.test.ts b/packages/computer/src/backends/worker-javascript/journal.test.ts new file mode 100644 index 00000000..2596e2d4 --- /dev/null +++ b/packages/computer/src/backends/worker-javascript/journal.test.ts @@ -0,0 +1,182 @@ +import { Database } from "@cloudflare/dofs"; +import { SQLiteTestStorage } from "@cloudflare/dofs/testing"; +import { describe, expect, it, vi } from "vitest"; + +import { + callKey, + createDurableTables, + decodeValue, + describeCall, + encodeValue, + ExecutionJournal, +} from "./journal.js"; + +function database() { + const db = new Database(new SQLiteTestStorage()); + createDurableTables(db); + return db; +} + +describe("callKey", () => { + it("ignores object key order", () => { + expect(callKey("fs.writeFile", [{ a: 1, b: 2 }])).toBe( + callKey("fs.writeFile", [{ b: 2, a: 1 }]), + ); + }); + + it("tells calls apart by name, value and type", () => { + const key = callKey("fs.writeFile", ["/a", "x"]); + expect(callKey("fs.readFile", ["/a", "x"])).not.toBe(key); + expect(callKey("fs.writeFile", ["/a", "y"])).not.toBe(key); + expect(callKey("fs.writeFile", ["/a", new TextEncoder().encode("x")])).not.toBe(key); + }); +}); + +describe("describeCall", () => { + it("shows arguments, summarizing bytes and cutting long values", () => { + expect(describeCall("fs.writeFile", ["/a", new Uint8Array(3), "x".repeat(100)])).toBe( + `fs.writeFile("/a", <3 bytes>, "${"x".repeat(46)}…)`, + ); + }); +}); + +describe("encodeValue", () => { + it("round-trips byte arrays inside responses", () => { + const value = { result: { data: new Uint8Array([0, 1, 255]), text: "t" } }; + expect(decodeValue(encodeValue(value))).toEqual(value); + }); +}); + +describe("ExecutionJournal", () => { + function record(db: Database) { + const journal = new ExecutionJournal(db, "js", "run"); + for (const path of ["/a", "/b"]) { + const step = journal.begin("fs.readFile", [path]); + if (step.kind !== "live") throw new Error("expected a live call"); + journal.commit(step.seq, { result: `contents of ${path}` }); + } + } + + it("answers recorded calls in any order, then goes live", () => { + const db = database(); + record(db); + const replay = new ExecutionJournal(db, "js", "run"); + expect(replay.begin("fs.readFile", ["/b"])).toMatchObject({ + kind: "replay", + response: { result: "contents of /b" }, + }); + expect(replay.expectedCall()).toBe('fs.readFile("/a")'); + expect(replay.begin("fs.readFile", ["/a"])).toMatchObject({ + kind: "replay", + response: { result: "contents of /a" }, + }); + expect(replay.expectedCall()).toBeUndefined(); + expect(replay.begin("fs.readFile", ["/c"])).toEqual({ kind: "live", seq: 3, rerun: false }); + }); + + it("reports a call the journal doesn't hold while recorded calls remain", () => { + const db = database(); + record(db); + const onDiverge = vi.fn<(message: string) => void>(); + const replay = new ExecutionJournal(db, "js", "run", onDiverge); + const step = replay.begin("fs.rm", ["/a"]); + expect(step).toMatchObject({ kind: "diverge" }); + expect(step.kind === "diverge" && step.message).toContain( + 'it called fs.rm("/a"), but the next call it made before was fs.readFile("/a").', + ); + replay.diverged("diverged"); + expect(onDiverge).toHaveBeenCalledWith("diverged"); + }); + + it("runs a call again when it was in flight at the interruption", () => { + const db = database(); + const first = new ExecutionJournal(db, "js", "run"); + expect(first.begin("host/ws:test.slow", [])).toEqual({ kind: "live", seq: 1, rerun: false }); + const replay = new ExecutionJournal(db, "js", "run"); + expect(replay.begin("host/ws:test.slow", [])).toEqual({ kind: "live", seq: 1, rerun: true }); + expect(replay.expectedCall()).toBeUndefined(); + }); + + it("numbers responses in the order they were committed", () => { + const db = database(); + const first = new ExecutionJournal(db, "js", "run"); + const a = first.begin("fs.readFile", ["/a"]); + const b = first.begin("fs.readFile", ["/b"]); + if (a.kind !== "live" || b.kind !== "live") throw new Error("expected live calls"); + expect(first.commit(b.seq, { result: "b" })).toMatchObject({ ordinal: 1 }); + expect(first.commit(a.seq, { result: "a" })).toMatchObject({ ordinal: 2 }); + const replay = new ExecutionJournal(db, "js", "run"); + expect(replay.begin("fs.readFile", ["/a"])).toMatchObject({ kind: "replay", ordinal: 2 }); + expect(replay.begin("fs.readFile", ["/b"])).toMatchObject({ kind: "replay", ordinal: 1 }); + // Live responses after the replay carry on the numbering. + const c = replay.begin("fs.readFile", ["/c"]); + if (c.kind !== "live") throw new Error("expected a live call"); + expect(replay.commit(c.seq, { result: "c" })).toMatchObject({ ordinal: 3 }); + }); + + it("splits a large response across rows", () => { + const db = database(); + const first = new ExecutionJournal(db, "js", "run"); + const step = first.begin("net.fetch", [{ url: "https://example.test/big" }]); + if (step.kind !== "live") throw new Error("expected a live call"); + const body = new Uint8Array(600 * 1024).map((_, index) => index % 251); + first.commit(step.seq, { result: { status: 200, body } }); + expect( + db.one<{ count: number }>("SELECT count(*) AS count FROM workspace_runtime_journal_chunks") + ?.count, + ).toBeGreaterThan(1); + const replay = new ExecutionJournal(db, "js", "run"); + expect(replay.begin("net.fetch", [{ url: "https://example.test/big" }])).toMatchObject({ + kind: "replay", + response: { result: { status: 200, body } }, + }); + }); + + it("saves the event log in chunks that can arrive out of order", () => { + const db = database(); + const first = new ExecutionJournal(db, "js", "run"); + first.record(2, [-1, 3]); + first.record(0, [1, 2]); + // Sent again by nobody, but harmless if it were. + first.record(0, [1, 2]); + expect(new ExecutionJournal(db, "js", "run").schedule()).toEqual([1, 2, -1, 3]); + }); + + it("drops log chunks past a gap", () => { + const db = database(); + const first = new ExecutionJournal(db, "js", "run"); + first.record(0, [1]); + first.record(3, [-2]); + const replay = new ExecutionJournal(db, "js", "run"); + expect(replay.schedule()).toEqual([1]); + // The replay logs those positions afresh, and the stale chunk is gone. + replay.record(1, [2, 3]); + replay.record(3, [-1]); + expect(new ExecutionJournal(db, "js", "run").schedule()).toEqual([1, 2, 3, -1]); + }); + + it("refuses a malformed event log", () => { + const journal = new ExecutionJournal(database(), "js", "run"); + for (const [position, events] of [ + [-1, [1]], + [0, []], + [0, [0]], + [0, [1.5]], + ["0", [1]], + ] as const) { + expect(() => journal.record(position, events)).toThrow("malformed event log"); + } + }); + + it("reports a replay that stops before a recorded call", () => { + const db = database(); + record(db); + const onDiverge = vi.fn<(message: string) => void>(); + const replay = new ExecutionJournal(db, "js", "run", onDiverge); + replay.begin("fs.readFile", ["/a"]); + replay.stalled(); + expect(onDiverge).toHaveBeenCalledWith( + expect.stringContaining('it stopped before calling fs.readFile("/b").'), + ); + }); +}); diff --git a/packages/computer/src/backends/worker-javascript/journal.ts b/packages/computer/src/backends/worker-javascript/journal.ts new file mode 100644 index 00000000..202bdf52 --- /dev/null +++ b/packages/computer/src/backends/worker-javascript/journal.ts @@ -0,0 +1,571 @@ +// Durable execution state for the JavaScript backend. +// +// An execution started with `onInterrupt: "restart"` or `"replay"` saves +// what it needs to start again: its module graph, input, context and +// limits. A replayed execution also journals every capability call it +// makes, writing each response before the isolate sees it. After the +// object restarts, the execution runs again from the top and the bridge +// answers its calls from the journal instead of running them, until the +// journal runs out and the execution carries on live. +// +// Each response is numbered when it's committed. The isolate logs the +// order it saw its events in, each response by number and each timer by +// the order it was set, and sends the log along with its next call or +// output, so the log is saved before anything that depends on it. When +// the run replays, the isolate gets the log and sees its events in that +// order again. A call the journal doesn't hold, while journaled calls are +// still unanswered, means the code took a different path anyway, and the +// execution fails rather than continue on a state the journal can't +// vouch for. + +import type { Database } from "@cloudflare/dofs"; + +import type { BridgeJournal, BridgeResponse } from "../../runtime/bridge.js"; +import type { WorkspaceRuntimeValue } from "../../runtime/types.js"; + +export type InterruptPolicy = "fail" | "restart" | "replay"; + +export const INTERRUPT_POLICIES: readonly InterruptPolicy[] = ["fail", "restart", "replay"]; + +/** Bump when the saved start or journal format changes. */ +export const JOURNAL_VERSION = 1; + +/** How many attempts an execution gets before an interruption fails it. */ +export const MAX_INTERRUPT_ATTEMPTS = 3; + +/** + * How many interrupted runs restart at once. A Durable Object runs at + * most ten Dynamic Workers at a time, and new executions need some. + */ +export const MAX_CONCURRENT_RECOVERIES = 4; + +// Saved start data and large responses are split across rows, since a +// Durable Object's SQLite holds at most 2 MB in a row. +const CHUNK_CHARACTERS = 256 * 1024; + +export interface DurableStart { + readonly version: number; + readonly modules: Record; + readonly entryName: string; + readonly input: WorkspaceRuntimeValue; + readonly env: Record; + readonly cwd: string; + /** Base64. */ + readonly stdin: string; + readonly timeoutMs: number; + readonly cpuMs: number; + readonly startedAt: number; + /** 64 hex digits that seed the run's random numbers. */ + readonly seed: string; + /** Identifies the runtime code the execution started on. */ + readonly fingerprint: string; +} + +export interface DurableRecord { + readonly policy: Exclude; + readonly attempt: number; + readonly start: DurableStart; +} + +interface JournalEntry { + readonly seq: number; + readonly key: string; + readonly label: string; + readonly state: "executing" | "applied"; + readonly response: string | null; + readonly at: number | null; + readonly ordinal: number | null; +} + +export type JournalStep = + | { + readonly kind: "replay"; + readonly response: BridgeResponse; + readonly at: number; + /** Where this response came in the order responses were delivered. */ + readonly ordinal: number; + } + | { + readonly kind: "live"; + readonly seq: number; + /** The call was in flight when the run was interrupted. */ + readonly rerun: boolean; + } + | { readonly kind: "diverge"; readonly message: string }; + +export const DURABLE_TABLES = [ + "workspace_runtime_durable", + "workspace_runtime_durable_chunks", + "workspace_runtime_journal", + "workspace_runtime_journal_chunks", + "workspace_runtime_schedule", +] as const; + +/** Most events one log chunk may carry. */ +const MAX_LOG_CHUNK = 65536; + +// The tables hold state for running executions only. A change to their +// shape should rename them rather than migrate: every backend creates +// them on connect, whether or not it uses them, and CREATE TABLE IF NOT +// EXISTS leaves an older table as it was. +export function createDurableTables(db: Database): void { + db.run(` + CREATE TABLE IF NOT EXISTS workspace_runtime_durable ( + backend TEXT NOT NULL, + execution_id TEXT NOT NULL, + policy TEXT NOT NULL, + attempt INTEGER NOT NULL, + PRIMARY KEY (backend, execution_id) + ) + `); + db.run(` + CREATE TABLE IF NOT EXISTS workspace_runtime_durable_chunks ( + backend TEXT NOT NULL, + execution_id TEXT NOT NULL, + idx INTEGER NOT NULL, + data TEXT NOT NULL, + PRIMARY KEY (backend, execution_id, idx) + ) + `); + db.run(` + CREATE TABLE IF NOT EXISTS workspace_runtime_journal ( + backend TEXT NOT NULL, + execution_id TEXT NOT NULL, + seq INTEGER NOT NULL, + call_key TEXT NOT NULL, + label TEXT NOT NULL, + state TEXT NOT NULL, + response TEXT, + at INTEGER, + ordinal INTEGER, + PRIMARY KEY (backend, execution_id, seq) + ) + `); + db.run(` + CREATE TABLE IF NOT EXISTS workspace_runtime_journal_chunks ( + backend TEXT NOT NULL, + execution_id TEXT NOT NULL, + seq INTEGER NOT NULL, + idx INTEGER NOT NULL, + data TEXT NOT NULL, + PRIMARY KEY (backend, execution_id, seq, idx) + ) + `); + db.run(` + CREATE TABLE IF NOT EXISTS workspace_runtime_schedule ( + backend TEXT NOT NULL, + execution_id TEXT NOT NULL, + position INTEGER NOT NULL, + events TEXT NOT NULL, + PRIMARY KEY (backend, execution_id, position) + ) + `); +} + +export function saveDurableStart( + db: Database, + backend: string, + id: string, + policy: Exclude, + start: DurableStart, +): void { + const text = JSON.stringify(start); + db.transactionSync(() => { + clearDurable(db, backend, id); + db.run( + `INSERT INTO workspace_runtime_durable (backend, execution_id, policy, attempt) + VALUES (?, ?, ?, 1)`, + backend, + id, + policy, + ); + for (let offset = 0, idx = 0; offset < text.length; offset += CHUNK_CHARACTERS, idx += 1) { + db.run( + `INSERT INTO workspace_runtime_durable_chunks (backend, execution_id, idx, data) + VALUES (?, ?, ?, ?)`, + backend, + id, + idx, + text.slice(offset, offset + CHUNK_CHARACTERS), + ); + } + }); +} + +export function loadDurable(db: Database, backend: string, id: string): DurableRecord | undefined { + const row = db.one<{ policy: string; attempt: number }>( + `SELECT policy, attempt FROM workspace_runtime_durable WHERE backend = ? AND execution_id = ?`, + backend, + id, + ); + if (!row || (row.policy !== "restart" && row.policy !== "replay")) return undefined; + const chunks = db.all<{ data: string }>( + `SELECT data FROM workspace_runtime_durable_chunks + WHERE backend = ? AND execution_id = ? ORDER BY idx`, + backend, + id, + ); + try { + const start = JSON.parse(chunks.map((chunk) => chunk.data).join("")) as DurableStart; + return { policy: row.policy, attempt: Number(row.attempt), start }; + } catch { + return undefined; + } +} + +export function recordAttempt(db: Database, backend: string, id: string, attempt: number): void { + db.run( + `UPDATE workspace_runtime_durable SET attempt = ? WHERE backend = ? AND execution_id = ?`, + attempt, + backend, + id, + ); +} + +export function clearJournal(db: Database, backend: string, id: string): void { + for (const table of [ + "workspace_runtime_journal", + "workspace_runtime_journal_chunks", + "workspace_runtime_schedule", + ]) { + db.run(`DELETE FROM ${table} WHERE backend = ? AND execution_id = ?`, backend, id); + } +} + +export function clearDurable(db: Database, backend: string, id: string): void { + db.run( + `DELETE FROM workspace_runtime_durable WHERE backend = ? AND execution_id = ?`, + backend, + id, + ); + db.run( + `DELETE FROM workspace_runtime_durable_chunks WHERE backend = ? AND execution_id = ?`, + backend, + id, + ); + clearJournal(db, backend, id); +} + +/** + * The capability calls of one execution, backed by SQLite. + * + * A journal opened over an earlier attempt's entries answers matching + * calls from them. Calls match by name and arguments, in the order they + * were made, rather than by position, so independent calls that finish + * in a different order on replay still match. + */ +export class ExecutionJournal implements BridgeJournal { + readonly #db: Database; + readonly #backend: string; + readonly #id: string; + readonly #entries: JournalEntry[]; + readonly #consumed = new Set(); + readonly #onDiverge: (message: string) => void; + #nextSeq: number; + #lastOrdinal: number; + #unanswered: number; + readonly #log: number[]; + + constructor( + db: Database, + backend: string, + id: string, + onDiverge: (message: string) => void = () => undefined, + ) { + this.#db = db; + this.#onDiverge = onDiverge; + this.#backend = backend; + this.#id = id; + this.#entries = db + .all<{ + seq: number; + call_key: string; + label: string; + state: string; + response: string | null; + at: number | null; + ordinal: number | null; + }>( + `SELECT seq, call_key, label, state, response, at, ordinal FROM workspace_runtime_journal + WHERE backend = ? AND execution_id = ? ORDER BY seq`, + backend, + id, + ) + .map((row) => ({ + seq: Number(row.seq), + key: row.call_key, + label: row.label, + state: row.state === "applied" && row.response !== null ? "applied" : "executing", + // An empty response marks one stored in chunks. + response: row.response === "" ? this.#chunkedResponse(Number(row.seq)) : row.response, + at: row.at === null ? null : Number(row.at), + ordinal: row.ordinal === null ? null : Number(row.ordinal), + })); + this.#nextSeq = (this.#entries.at(-1)?.seq ?? 0) + 1; + const applied = this.#entries.filter((entry) => entry.state === "applied"); + this.#unanswered = applied.length; + this.#lastOrdinal = applied.reduce((last, entry) => Math.max(last, entry.ordinal ?? 0), 0); + this.#log = this.#loadLog(); + } + + /** + * The order the run saw its events in so far: a positive number is the + * response with that ordinal, a negative one the timer set in that + * place firing. The isolate replays its events in this order. + */ + schedule(): number[] { + return [...this.#log]; + } + + /** + * Save part of the isolate's event log. Chunks can arrive out of order, + * since calls and output travel separately, so each says where it goes. + */ + record(position: unknown, events: unknown): void { + if ( + !Number.isSafeInteger(position) || + (position as number) < 0 || + !Array.isArray(events) || + events.length === 0 || + events.length > MAX_LOG_CHUNK || + !events.every((event) => Number.isSafeInteger(event) && event !== 0) + ) { + throw new Error("Workspace execution sent a malformed event log."); + } + this.#db.run( + `INSERT OR IGNORE INTO workspace_runtime_schedule (backend, execution_id, position, events) + VALUES (?, ?, ?, ?)`, + this.#backend, + this.#id, + position as number, + JSON.stringify(events), + ); + } + + #chunkedResponse(seq: number): string { + return this.#db + .all<{ data: string }>( + `SELECT data FROM workspace_runtime_journal_chunks + WHERE backend = ? AND execution_id = ? AND seq = ? ORDER BY idx`, + this.#backend, + this.#id, + seq, + ) + .map((chunk) => chunk.data) + .join(""); + } + + // The log as far as it runs without a gap. A chunk that arrived after + // one lost to the interruption describes events nobody saw, so it's + // dropped, and the replay logs those positions afresh. + #loadLog(): number[] { + const rows = this.#db.all<{ position: number; events: string }>( + `SELECT position, events FROM workspace_runtime_schedule + WHERE backend = ? AND execution_id = ? ORDER BY position`, + this.#backend, + this.#id, + ); + const log: number[] = []; + for (const row of rows) { + if (Number(row.position) !== log.length) break; + log.push(...(JSON.parse(row.events) as number[])); + } + this.#db.run( + `DELETE FROM workspace_runtime_schedule + WHERE backend = ? AND execution_id = ? AND position >= ?`, + this.#backend, + this.#id, + log.length, + ); + return log; + } + + /** + * Decide how to answer one call. A live call is journaled as executing + * before it runs, and `commit()` records its response. + */ + begin(name: string, args: readonly unknown[]): JournalStep { + const key = callKey(name, args); + const label = describeCall(name, args); + const entry = this.#entries.find((item) => !this.#consumed.has(item.seq) && item.key === key); + if (entry !== undefined) { + this.#consumed.add(entry.seq); + if (entry.state === "applied") { + this.#unanswered -= 1; + return { + kind: "replay", + response: decodeValue(entry.response ?? "null") as BridgeResponse, + at: entry.at ?? 0, + ordinal: entry.ordinal ?? 0, + }; + } + // 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 }; + } + if (this.#unanswered > 0) { + return { + kind: "diverge", + message: + "The execution took a different path when it resumed after an interruption: " + + `it called ${label}, but the next call it made before was ${this.expectedCall()}.`, + }; + } + const seq = this.#nextSeq++; + this.#db.run( + `INSERT INTO workspace_runtime_journal + (backend, execution_id, seq, call_key, label, state, response, at) + VALUES (?, ?, ?, ?, ?, 'executing', NULL, NULL)`, + this.#backend, + this.#id, + seq, + key, + label, + ); + return { kind: "live", seq, rerun: false }; + } + + /** Record a live call's response, before the isolate sees it. */ + commit(seq: number, response: BridgeResponse): { at: number; ordinal: number } { + const at = Date.now(); + this.#lastOrdinal += 1; + const text = encodeValue(response); + const chunked = text.length > CHUNK_CHARACTERS; + if (chunked) { + for (let offset = 0, idx = 0; offset < text.length; offset += CHUNK_CHARACTERS, idx += 1) { + this.#db.run( + `INSERT OR REPLACE INTO workspace_runtime_journal_chunks + (backend, execution_id, seq, idx, data) + VALUES (?, ?, ?, ?, ?)`, + this.#backend, + this.#id, + seq, + idx, + text.slice(offset, offset + CHUNK_CHARACTERS), + ); + } + } + this.#db.run( + `UPDATE workspace_runtime_journal SET state = 'applied', response = ?, at = ?, ordinal = ? + WHERE backend = ? AND execution_id = ? AND seq = ?`, + chunked ? "" : text, + at, + this.#lastOrdinal, + this.#backend, + this.#id, + seq, + ); + return { at, ordinal: this.#lastOrdinal }; + } + + diverged(message: string): void { + this.#onDiverge(message); + } + + /** The isolate is waiting for a call the replay never made. */ + stalled(): void { + const expected = this.expectedCall(); + const where = + expected === undefined + ? "before reaching the point it was interrupted at" + : `before calling ${expected}`; + this.#onDiverge( + `The execution took a different path when it resumed after an interruption: it stopped ${where}.`, + ); + } + + /** The first journaled call not answered yet. */ + expectedCall(): string | undefined { + return this.#entries.find((item) => item.state === "applied" && !this.#consumed.has(item.seq)) + ?.label; + } +} + +/** A short, readable form of a call for messages: its name and arguments. */ +export function describeCall(name: string, args: readonly unknown[]): string { + const text = `${name}(${args.map(describeValue).join(", ")})`; + return text.length > 160 ? `${text.slice(0, 159)}…` : text; +} + +function describeValue(value: unknown): string { + if (value instanceof Uint8Array) return `<${value.byteLength} bytes>`; + const text = JSON.stringify(value) ?? "undefined"; + return text.length > 48 ? `${text.slice(0, 47)}…` : text; +} + +/** A stable identity for a call: its name and its arguments. */ +export function callKey(name: string, args: readonly unknown[]): string { + return digest(`${name}\n${stableStringify(args)}`); +} + +/** Identify runtime code by the text of its generated modules. */ +export function fingerprint(parts: readonly string[]): string { + return `v${JOURNAL_VERSION}:${digest(parts.join("\u0000"))}`; +} + +// JSON with sorted object keys and tagged byte arrays, so equal values +// always produce equal text. +function stableStringify(value: unknown): string { + if (value instanceof Uint8Array) return JSON.stringify({ [BYTES_TAG]: toBase64(value) }); + if (Array.isArray(value)) return `[${value.map((item) => stableStringify(item)).join(",")}]`; + if (value !== null && typeof value === "object") { + const record = value as Record; + const fields = Object.keys(record) + .filter((key) => record[key] !== undefined) + .sort() + .map((key) => `${JSON.stringify(key)}:${stableStringify(record[key])}`); + return `{${fields.join(",")}}`; + } + return JSON.stringify(value) ?? "null"; +} + +// Two 53-bit cyrb53 hashes with different seeds, plus the length. +function digest(text: string): string { + return `${cyrb53(text, 0)}${cyrb53(text, 0x9e3779b9)}:${text.length}`; +} + +function cyrb53(text: string, seed: number): string { + let h1 = 0xdeadbeef ^ seed; + let h2 = 0x41c6ce57 ^ seed; + for (let index = 0; index < text.length; index += 1) { + const code = text.charCodeAt(index); + h1 = Math.imul(h1 ^ code, 2654435761); + h2 = Math.imul(h2 ^ code, 1597334677); + } + h1 = Math.imul(h1 ^ (h1 >>> 16), 2246822507) ^ Math.imul(h2 ^ (h2 >>> 13), 3266489909); + h2 = Math.imul(h2 ^ (h2 >>> 16), 2246822507) ^ Math.imul(h1 ^ (h1 >>> 13), 3266489909); + return (4294967296 * (2097151 & h2) + (h1 >>> 0)).toString(36).padStart(11, "0"); +} + +const BYTES_TAG = "\u0000bytes"; + +export function encodeValue(value: unknown): string { + return JSON.stringify(value, (_key, item: unknown) => + item instanceof Uint8Array ? { [BYTES_TAG]: toBase64(item) } : item, + ); +} + +export function decodeValue(text: string): unknown { + return JSON.parse(text, (_key, item: unknown) => { + if (item !== null && typeof item === "object" && !Array.isArray(item)) { + const tagged = (item as Record)[BYTES_TAG]; + if (typeof tagged === "string" && Object.keys(item).length === 1) return fromBase64(tagged); + } + return item; + }); +} + +export function toBase64(bytes: Uint8Array): string { + let binary = ""; + for (let offset = 0; offset < bytes.length; offset += 0x8000) { + binary += String.fromCharCode(...bytes.subarray(offset, offset + 0x8000)); + } + return btoa(binary); +} + +export function fromBase64(text: string): Uint8Array { + const binary = atob(text); + const bytes = new Uint8Array(binary.length); + for (let index = 0; index < binary.length; index += 1) bytes[index] = binary.charCodeAt(index); + return bytes; +} diff --git a/packages/computer/src/backends/worker-javascript/module-graph.ts b/packages/computer/src/backends/worker-javascript/module-graph.ts index e6522e95..45b0df36 100644 --- a/packages/computer/src/backends/worker-javascript/module-graph.ts +++ b/packages/computer/src/backends/worker-javascript/module-graph.ts @@ -7,6 +7,7 @@ import type { WorkspaceModuleFunctions, WorkspaceRuntimeLoader, } from "../../runtime/types.js"; +import { DURABLE_MODULE, durableRuntimeModule } from "./durable-runtime.js"; export type JavaScriptModuleMap = WorkspaceRuntimeLoader extends { load(code: { modules: infer Modules }): unknown; @@ -263,6 +264,7 @@ export async function buildModuleGraph(options: BuildModuleGraphOptions) { const entryName = moduleName(entryPath); const modules: Record = Object.assign(Object.create(null), { [CAPABILITIES_MODULE]: capabilitiesModule(options.maxCapabilityBytes), + [DURABLE_MODULE]: durableRuntimeModule(), }); const seen = new Set(); let totalBytes = new TextEncoder().encode(options.source).byteLength; @@ -297,7 +299,7 @@ export async function buildModuleGraph(options: BuildModuleGraphOptions) { if (node !== specifier) edits.push({ ...site, specifier: node }); continue; } - if (specifier === CAPABILITIES_MODULE) { + if (specifier === CAPABILITIES_MODULE || specifier === DURABLE_MODULE) { throw new Error(`Module ${JSON.stringify(specifier)} is reserved for Workspace internals.`); } if (options.emptyHostModules?.has(specifier)) { @@ -510,19 +512,35 @@ function directoryName(name: string) { function isInternalModuleName(name: string) { return ( name === CAPABILITIES_MODULE || + name === DURABLE_MODULE || name === RUNNER_MODULE || name === ENTRY_BASENAME || name === MODULES_DIRECTORY || name.startsWith(`${MODULES_DIRECTORY}/`) || name.endsWith(`/${CAPABILITIES_MODULE}`) || + name.endsWith(`/${DURABLE_MODULE}`) || name.endsWith(`/${RUNNER_MODULE}`) || name.split("/").at(-1)?.startsWith("ws:") === true ); } +/** + * The generated modules every graph carries. A durable execution + * compares them to tell whether the runtime changed under it. + */ +export function internalModuleSources(maxCapabilityBytes: number): string[] { + return [ + capabilitiesModule(maxCapabilityBytes), + durableRuntimeModule(), + nodeFsPromisesModule(), + nodeFsModule(), + ]; +} + function capabilitiesModule(maxCapabilityBytes: number) { const requestTooLargeMessage = `Workspace capability request exceeds ${maxCapabilityBytes} bytes.`; return ` + import { deliver, failed, request, takeLog } from "${DURABLE_MODULE}"; let host; const callKey = Symbol.for("cloudflare.workspace.runtime.call"); const filesystemMethods = new Set([ @@ -553,9 +571,11 @@ function capabilitiesModule(maxCapabilityBytes: number) { throw new Error(${JSON.stringify(requestTooLargeMessage)}); } let payload; + request(); try { - payload = await host.call(namespace + "." + method, args); + payload = await host.call(namespace + "." + method, args, takeLog()); } catch (error) { + failed(); // A module is evaluated outside any request, and the runtime // refuses I/O there with an error about request handlers, which // this code doesn't have. Name the call and the fix instead. @@ -570,6 +590,8 @@ function capabilitiesModule(maxCapabilityBytes: number) { } throw error; } + // A replayable run sees responses in the order it first saw them. + payload = await deliver(payload); if (payload.error !== undefined) { const error = new Error(payload.error.message); if (payload.error.code !== undefined) error.code = payload.error.code; 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 7e5c76e4..dc2636de 100644 --- a/packages/computer/src/backends/worker-javascript/worker-javascript.test.ts +++ b/packages/computer/src/backends/worker-javascript/worker-javascript.test.ts @@ -1943,3 +1943,245 @@ describe("WorkerJavaScriptBackend", () => { }); }); }); + +describe("WorkerJavaScriptBackend interrupted executions", () => { + const hanging = { + load() { + return { + getEntrypoint() { + return { evaluate: () => new Promise(() => undefined) }; + }, + }; + }, + }; + + function open( + storage: SQLiteTestStorage, + options: Partial[0]> = {}, + ) { + return new Workspace({ + storage, + backends: [new WorkerJavaScriptBackend({ loader: hanging, ...options })], + }); + } + + async function interrupt(storage: SQLiteTestStorage, timeoutMs?: number) { + const workspace = open(storage); + await workspace.fs.mkdir("/workspace", { recursive: true }); + await workspace.runtime.exec("export default 1", { + backend: "worker-javascript", + id: "run", + onInterrupt: "replay", + ...(timeoutMs === undefined ? {} : { timeoutMs }), + }); + return workspace; + } + + async function settle(workspace: Workspace) { + const handle = await workspace.runtime.getExec("run", { + backend: "worker-javascript", + encoding: "utf8", + }); + return handle.result(); + } + + it("saves the event log an output frame carries along with the output", async () => { + const storage = new SQLiteTestStorage(); + let whileRunning: unknown[] = []; + const workspace = new Workspace({ + storage, + backends: [ + new WorkerJavaScriptBackend({ + onInterrupt: "replay", + loader: { + load() { + return { + getEntrypoint() { + return { + async evaluate( + _input: unknown, + host: { attachOutput(readable: ReadableStream): Promise }, + ) { + const encoder = new TextEncoder(); + let output!: ReadableStreamDefaultController; + const drained = host.attachOutput( + new ReadableStream({ start: (controller) => void (output = controller) }), + ); + const frame = { + name: "stdout", + b64: btoa("tick\n"), + log: { from: 0, events: [-1] }, + }; + output.enqueue(encoder.encode(`${JSON.stringify(frame)}\n`)); + await new Promise((resolve) => setTimeout(resolve, 10)); + const db = new Database(storage); + whileRunning = [ + db.all("SELECT position, events FROM workspace_runtime_schedule"), + db.all("SELECT name FROM workspace_runtime_events"), + ]; + output.enqueue( + encoder.encode(`${JSON.stringify({ name: "exit", code: 0 })}\n`), + ); + output.close(); + await drained; + }, + }; + }, + }; + }, + }, + }), + ], + }); + await workspace.fs.mkdir("/workspace", { recursive: true }); + const handle = await workspace.runtime.exec("export default 1", { + backend: "worker-javascript", + }); + await expect(handle.result()).resolves.toMatchObject({ status: "completed" }); + expect(whileRunning).toEqual([[{ position: 0, events: "[-1]" }], [{ name: "stdout" }]]); + await workspace.close(); + }); + + it("fails a run after three attempts", async () => { + const storage = new SQLiteTestStorage(); + const workspaces = [await interrupt(storage)]; + for (let attempt = 0; attempt < 2; attempt += 1) { + const workspace = open(storage); + await workspace.runtime.recover(); + workspaces.push(workspace); + } + const last = open(storage); + await expect(settle(last)).resolves.toMatchObject({ + status: "failed", + stderr: expect.stringContaining("interrupted 3 times"), + }); + for (const workspace of [...workspaces, last]) await workspace.close(); + }); + + it("fails a run when the runtime changed under it", async () => { + const storage = new SQLiteTestStorage(); + const first = await interrupt(storage); + const changed = open(storage, { maxStdioBytes: 2048 }); + await expect(settle(changed)).resolves.toMatchObject({ + status: "failed", + stderr: expect.stringContaining("runtime changed since it started"), + }); + await changed.close(); + await first.close(); + }); + + it("fails a run whose timeout passed while it was interrupted", async () => { + const storage = new SQLiteTestStorage(); + const first = await interrupt(storage, 1000); + const now = Date.now(); + const clock = vi.spyOn(Date, "now").mockReturnValue(now + 5000); + try { + const later = open(storage); + await expect(settle(later)).resolves.toMatchObject({ + status: "failed", + stderr: expect.stringContaining("timeout passed"), + }); + await later.close(); + } finally { + clock.mockRestore(); + } + await first.close(); + }); + + it("journals a run's calls and removes them once it settles", async () => { + const storage = new SQLiteTestStorage(); + let whileRunning: unknown[] = []; + const workspace = new Workspace({ + storage, + backends: [ + new WorkerJavaScriptBackend({ + onInterrupt: "replay", + loader: { + load() { + return { + getEntrypoint() { + return { + async evaluate( + _input: unknown, + host: Parameters[0] & { + call(name: string, args: unknown[]): Promise; + }, + ) { + await host.call("fs.writeFile", ["/workspace/out.txt", "x"]); + whileRunning = new Database(storage).all( + "SELECT label, state FROM workspace_runtime_journal", + ); + await evaluateResult(host, 1); + }, + }; + }, + }; + }, + }, + }), + ], + }); + await workspace.fs.mkdir("/workspace", { recursive: true }); + const handle = await workspace.runtime.exec("export default 1", { + backend: "worker-javascript", + }); + await expect(handle.result()).resolves.toMatchObject({ status: "completed", value: 1 }); + expect(whileRunning).toEqual([ + { label: 'fs.writeFile("/workspace/out.txt", "x")', state: "applied" }, + ]); + const db = new Database(storage); + for (const table of [ + "workspace_runtime_durable", + "workspace_runtime_durable_chunks", + "workspace_runtime_journal", + ]) { + expect(db.all(`SELECT * FROM ${table}`), table).toEqual([]); + } + await workspace.close(); + }); + + it("restarts at most four interrupted runs at once", async () => { + const storage = new SQLiteTestStorage(); + const first = open(storage); + await first.fs.mkdir("/workspace", { recursive: true }); + for (let index = 0; index < 6; index += 1) { + await first.runtime.exec("export default 1", { + backend: "worker-javascript", + id: `run-${index}`, + onInterrupt: "replay", + }); + } + let launched = 0; + const later = open(storage, { + loader: { + load() { + launched += 1; + return hanging.load(); + }, + }, + }); + await later.runtime.recover(); + await vi.waitFor(() => expect(launched).toBe(4)); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(launched).toBe(4); + // Each run that settles lets a waiting one start. + await later.runtime.killExec("run-0", { backend: "worker-javascript" }); + await vi.waitFor(() => expect(launched).toBe(5)); + await later.close(); + await first.close(); + }); + + it("rejects an unknown onInterrupt policy", async () => { + expect( + () => new WorkerJavaScriptBackend({ loader: hanging, onInterrupt: "maybe" as never }), + ).toThrow('onInterrupt must be "fail", "restart" or "replay".'); + const workspace = open(new SQLiteTestStorage()); + await expect( + workspace.runtime.exec("export default 1", { + backend: "worker-javascript", + onInterrupt: "maybe" as never, + }), + ).rejects.toThrow("onInterrupt must be"); + await workspace.close(); + }); +}); diff --git a/packages/computer/src/backends/worker-javascript/worker-javascript.ts b/packages/computer/src/backends/worker-javascript/worker-javascript.ts index 3a8ee6e3..28ab5f38 100644 --- a/packages/computer/src/backends/worker-javascript/worker-javascript.ts +++ b/packages/computer/src/backends/worker-javascript/worker-javascript.ts @@ -1,7 +1,11 @@ import type { WorkspaceBackendHost } from "../../backend.js"; import { WorkspaceRuntimeBridge } from "../../runtime/bridge.js"; import { assertRuntimeValue, WorkspaceRuntimeCapability } from "../../runtime/capability.js"; -import { dynamicWorkerEgress, type WorkspaceEgressPolicy } from "../../runtime/egress.js"; +import { + dynamicWorkerEgress, + hostEgressFetch, + type WorkspaceEgressPolicy, +} from "../../runtime/egress.js"; import type { ModuleExecutionEnvelope, ModuleExecutionInput, @@ -12,14 +16,32 @@ import type { WorkspaceModuleFunctions, WorkspaceRuntimeAccess, WorkspaceRuntimeEvent, + WorkspaceRuntimeInterruptPolicy, WorkspaceRuntimeLoader, WorkspaceRuntimeValue, } from "../../runtime/types.js"; import { decodeRuntimeFrames, type RuntimeFrame } from "./frames.js"; +import { + clearDurable, + createDurableTables, + DURABLE_TABLES, + ExecutionJournal, + fingerprint, + fromBase64, + INTERRUPT_POLICIES, + JOURNAL_VERSION, + loadDurable, + MAX_CONCURRENT_RECOVERIES, + MAX_INTERRUPT_ATTEMPTS, + recordAttempt, + saveDurableStart, + toBase64, +} from "./journal.js"; import { assertHostModuleExports, buildModuleGraph, hasNodeModules, + internalModuleSources, NODE_MODULES_DESCRIPTION, type ParsedModules, parseModules, @@ -65,6 +87,31 @@ export interface WorkerJavaScriptBackendOptions { * effect. */ maxCpuMs?: number; + /** + * What happens to a run if the Durable Object restarts under it, as on + * a deploy or crash. A run's own `onInterrupt` overrides it. Defaults + * to `"fail"`. + * + * - `"fail"` settles the run as failed. + * - `"restart"` runs it again from the start, once the backend + * reconnects. Its output so far stays, so the run's output repeats. + * - `"replay"` runs it again from the start, but answers every + * filesystem, host module and `fetch` call it already made from a + * journal instead of running it again, in the order the responses + * first arrived. Its clock, timers and random numbers replay too. + * Output already recorded isn't repeated. If the run takes a + * different path, it fails. + * + * A run is resumed when the backend reconnects, which the next runtime + * call or `ready()` triggers, or `runtime.recover()`. It gets three + * attempts and must finish within its original timeout. A call in + * flight at the interruption runs again, unless it's marked with + * `unsafeToRepeat()`. + * + * `"restart"` and `"replay"` are experimental and may change in any + * release. + */ + onInterrupt?: WorkspaceRuntimeInterruptPolicy; /** * 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 @@ -133,6 +180,7 @@ type ResolvedWorkerJavaScriptBackendOptions = Required< | "defaultTimeoutMs" | "maxTimeoutMs" | "maxCpuMs" + | "onInterrupt" | "keepAliveIntervalMs" | "maxSourceBytes" | "maxInputBytes" @@ -163,6 +211,10 @@ interface WorkspaceExecutionContext { env: Record; cwd: string; stdin: Uint8Array; + // Set for a replayable run: the seed for its random numbers, the time + // its clock starts at, and the order it saw its events in before it was + // interrupted, which is empty on the first attempt. + durable?: { seed: string; startedAt: number; schedule: number[] }; } interface JavaScriptEntrypoint { @@ -197,6 +249,10 @@ interface ExecutionRecord { result?: WorkspaceRuntimeValue; hasResult?: boolean; exitCode?: number; + journal?: ExecutionJournal; + // 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 }>; } export class WorkerJavaScriptBackend implements WorkspaceModuleBackend { @@ -218,6 +274,8 @@ export class WorkerJavaScriptBackend implements WorkspaceModuleBackend { assertPositiveFinite(defaultTimeoutMs, "defaultTimeoutMs"); const maxCpuMs = options.maxCpuMs ?? maxTimeoutMs; assertPositiveFinite(maxCpuMs, "maxCpuMs"); + const onInterrupt = options.onInterrupt ?? "fail"; + assertInterruptPolicy(onInterrupt); const keepAliveIntervalMs = options.keepAliveIntervalMs ?? 60_000; // Pending work protects a Durable Object for at most 15 minutes, so a // longer interval would leave gaps. @@ -282,6 +340,7 @@ export class WorkerJavaScriptBackend implements WorkspaceModuleBackend { defaultTimeoutMs, maxTimeoutMs, maxCpuMs, + onInterrupt, keepAliveIntervalMs, maxSourceBytes: options.maxSourceBytes ?? 1024 * 1024, maxInputBytes: options.maxInputBytes ?? 1024 * 1024, @@ -341,6 +400,8 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { // Running executions, each with the time after which it stops counting // towards the keep-alive. readonly #running = new Map(); + readonly #recoveryQueue: Array<() => void> = []; + readonly #recovering = new Set(); #keepAlive: ReturnType | undefined; // Zero when the timer is off or can't keep this object in memory. readonly #keepAliveIntervalMs: number; @@ -405,26 +466,162 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { PRIMARY KEY (backend, execution_id, seq) ) `); + createDurableTables(host.db); this.#prune(); const interrupted = host.db.all<{ id: string }>( `SELECT id FROM workspace_runtime_executions WHERE backend = ? AND status = 'running'`, this.#backendId, ); - for (const { id } of interrupted) { - const record = this.#recordFor(id); - if (record?.status !== "running") continue; + for (const { id } of interrupted) this.#resumeInterrupted(id); + this.#drainRecoveries(); + } + + // Recovery runs when the backend connects, so there's nothing left to + // do by the time the runtime asks. + async recover(): Promise {} + + // Start a run the restart interrupted again, if it asked for that and + // still can, or settle it as failed. + #resumeInterrupted(id: string) { + 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( - "Execution was interrupted when its Workspace runtime restarted.\n", - ), + value: new TextEncoder().encode(`${message}\n`), }, { id, seq: record.events.length + 2, name: "exit", code: 1 }, ]); + let saved: ReturnType; + try { + saved = loadDurable(this.#host.db, this.#backendId, id); + } catch { + saved = undefined; + } + if (saved === undefined) { + fail("Execution was interrupted when its Workspace runtime restarted."); + return; + } + const { attempt, start } = saved; + if (attempt >= MAX_INTERRUPT_ATTEMPTS) { + fail(`Execution was interrupted ${attempt} times and won't be started again.`); + return; } + if (start.version !== JOURNAL_VERSION || start.fingerprint !== this.#fingerprint(start)) { + fail( + "Execution was interrupted and can't resume, because the JavaScript runtime changed since it started.", + ); + return; + } + const resumable = saved; + this.#recoveryQueue.push(() => { + if (record.status !== "running" || record.finalization) return; + this.#recovering.add(id); + this.#relaunch(record, resumable, fail); + }); + } + + // 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. + #drainRecoveries() { + while (this.#recovering.size < MAX_CONCURRENT_RECOVERIES) { + const next = this.#recoveryQueue.shift(); + if (next === undefined) return; + next(); + } + } + + #relaunch( + record: ExecutionRecord, + { policy, attempt, start }: NonNullable>, + fail: (message: string) => void, + ) { + const id = record.id; + // 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(); + if (remainingMs <= 0) { + fail("Execution was interrupted and its timeout passed before it could resume."); + return; + } + try { + recordAttempt(this.#host.db, this.#backendId, id, attempt + 1); + let journal: ExecutionJournal | undefined; + let durable: WorkspaceExecutionContext["durable"]; + if (policy === "replay") { + journal = this.#journalFor(record); + durable = { seed: start.seed, startedAt: start.startedAt, schedule: journal.schedule() }; + record.replayed = { + stdout: { bytes: recordedOutput(record.events, "stdout"), offset: 0 }, + stderr: { bytes: recordedOutput(record.events, "stderr"), offset: 0 }, + }; + } else { + this.#append(record, { + id, + seq: record.events.length + 1, + name: "stderr", + value: new TextEncoder().encode( + "Execution restarted from the beginning after its Workspace runtime was interrupted.\n", + ), + }); + } + void this.#launch(record, { + capability: this.#capability(), + modules: start.modules, + entryName: start.entryName, + input: start.input, + context: { + env: start.env, + cwd: start.cwd, + stdin: fromBase64(start.stdin), + ...(durable === undefined ? {} : { durable }), + }, + timeoutMs: remainingMs, + cpuMs: start.cpuMs, + journal, + }); + } catch (error) { + fail(error instanceof Error ? error.message : String(error)); + } + } + + // The generated code a run depends on. A run resumes only on the same. + #fingerprint(start: { entryName: string }): string { + return fingerprint([ + runtimeWorkerModule(start.entryName, this.#options.maxStdioBytes), + ...internalModuleSources(this.#options.maxCapabilityBytes), + this.#options.compatibilityDate, + this.#options.compatibilityFlags.join(","), + ]); + } + + #journalFor(record: ExecutionRecord): ExecutionJournal { + return new ExecutionJournal(this.#host.db, this.#backendId, record.id, (message) => + this.#diverge(record, message), + ); + } + + // A replayed run called something its journal doesn't hold. Stop it + // now, before it acts on a state the journal can't account for. + #diverge(record: ExecutionRecord, message: string) { + if (record.status !== "running" || record.finalization) return; + record.control?.cancel(); + void this.#finalize(record, message); + } + + #capability(): WorkspaceRuntimeCapability { + return new WorkspaceRuntimeCapability( + this.#host.fs, + this.#options.root, + this.#options.access, + this.#options.maxCapabilityBytes, + this.#options.maxDirectoryEntries, + ); } // Parsing a large bundle on every run would be wasteful, so source @@ -466,6 +663,8 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { throw new Error(`Workspace runtime stdin exceeds ${this.#options.maxStdinBytes} bytes.`); } assertEnv(input.env, this.#options.maxEnvBytes); + const policy = input.onInterrupt ?? this.#options.onInterrupt; + assertInterruptPolicy(policy); if (new TextEncoder().encode(input.source).byteLength > this.#options.maxSourceBytes) { throw new Error(`Workspace runtime source exceeds ${this.#options.maxSourceBytes} bytes.`); } @@ -477,13 +676,7 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { try { if (this.#options.access === "read-write") await ensureDirectory(this.#host.fs, this.#options.root); - const capability = new WorkspaceRuntimeCapability( - this.#host.fs, - this.#options.root, - this.#options.access, - this.#options.maxCapabilityBytes, - this.#options.maxDirectoryEntries, - ); + const capability = this.#capability(); const graph = await buildModuleGraph({ source: input.source, cwd: input.cwd ?? this.#options.root, @@ -522,52 +715,52 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { throw error; } this.#records.set(id, record); - this.#markRunning(id, timeoutMs); + const context: WorkspaceExecutionContext = { + env: input.env ?? {}, + cwd: input.cwd ?? this.#options.root, + stdin: stdinBytes, + }; + const cpuMs = Math.min(timeoutMs, this.#options.maxCpuMs); + let journal: ExecutionJournal | undefined; try { - const bridge = new WorkspaceRuntimeBridge(capability, { - hostModules: this.#hostModuleFunctions, - maxPayloadBytes: this.#options.maxCapabilityBytes, - maxCallDurationMs: this.#options.maxHostCallMs, - maxConcurrentCalls: this.#options.maxConcurrentCapabilityCalls, - maxCalls: this.#options.maxCapabilityCalls, - maxTotalRequestBytes: this.#options.maxCapabilityRequestBytes, - maxTotalResponseBytes: this.#options.maxCapabilityResponseBytes, - maxResultBytes: this.#options.maxResultBytes, - onAttachOutput: (readable) => this.#pumpFrames(record, readable), - }); - record.bridge = bridge; - record.control = startJavaScriptExecution({ - loader: this.#options.loader, - modules: graph.modules, - entryName: graph.entryName, - input: inputValue, - context: { - env: input.env ?? {}, - cwd: input.cwd ?? this.#options.root, - stdin: stdinBytes, - }, - bridge, - timeoutMs, - cpuMs: Math.min(timeoutMs, this.#options.maxCpuMs), - egress: this.#options.egress, - compatibilityDate: this.#options.compatibilityDate, - compatibilityFlags: this.#options.compatibilityFlags, - maxStdioBytes: this.#options.maxStdioBytes, - maxSourceBytes: this.#options.maxSourceBytes, - onComplete: () => this.#finalize(record), - onError: (message) => this.#finalize(record, message), - }); - // The completion promise drives finalize (which writes the - // terminal SQL) and is no longer handed to a host lifetime - // hook. Observe it so a finalize rejection surfaces as a - // swallowed error rather than an unhandled rejection in the - // Durable Object. - void record.control.completion.catch(() => undefined); + if (policy !== "fail") { + const startedAt = Date.now(); + const seed = [...crypto.getRandomValues(new Uint8Array(32))] + .map((byte) => byte.toString(16).padStart(2, "0")) + .join(""); + saveDurableStart(this.#host.db, this.#backendId, id, policy, { + version: JOURNAL_VERSION, + modules: graph.modules, + entryName: graph.entryName, + input: inputValue, + env: context.env, + cwd: context.cwd, + stdin: toBase64(stdinBytes), + timeoutMs, + cpuMs, + startedAt, + seed, + fingerprint: this.#fingerprint(graph), + }); + if (policy === "replay") { + journal = this.#journalFor(record); + context.durable = { seed, startedAt, schedule: [] }; + } + } } catch (error) { - record.control?.cancel(); - await record.control?.completion.catch(() => undefined); await this.#finalize(record, error instanceof Error ? error.message : String(error)); + return { id, events: this.#stream(record) }; } + await this.#launch(record, { + capability, + modules: graph.modules, + entryName: graph.entryName, + input: inputValue, + context, + timeoutMs, + cpuMs, + journal, + }); return { id, events: this.#stream(record) }; } finally { this.#pendingIds.delete(id); @@ -579,6 +772,68 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { } } + // Start a run's Dynamic Worker. A failure to start settles the run. + async #launch( + record: ExecutionRecord, + spec: { + capability: WorkspaceRuntimeCapability; + modules: Record; + entryName: string; + input: WorkspaceRuntimeValue; + context: WorkspaceExecutionContext; + timeoutMs: number; + cpuMs: number; + journal: ExecutionJournal | undefined; + }, + ): Promise { + this.#markRunning(record.id, spec.timeoutMs); + try { + const bridge = new WorkspaceRuntimeBridge(spec.capability, { + hostModules: this.#hostModuleFunctions, + maxPayloadBytes: this.#options.maxCapabilityBytes, + maxCallDurationMs: this.#options.maxHostCallMs, + maxConcurrentCalls: this.#options.maxConcurrentCapabilityCalls, + maxCalls: this.#options.maxCapabilityCalls, + maxTotalRequestBytes: this.#options.maxCapabilityRequestBytes, + maxTotalResponseBytes: this.#options.maxCapabilityResponseBytes, + maxResultBytes: this.#options.maxResultBytes, + onAttachOutput: (readable) => this.#pumpFrames(record, readable), + callIdPrefix: record.id, + ...hostEgressFetch(this.#options.egress), + ...(spec.journal === undefined ? {} : { journal: spec.journal }), + }); + record.bridge = bridge; + record.journal = spec.journal; + record.control = startJavaScriptExecution({ + loader: this.#options.loader, + modules: spec.modules, + entryName: spec.entryName, + input: spec.input, + context: spec.context, + bridge, + timeoutMs: spec.timeoutMs, + cpuMs: spec.cpuMs, + egress: this.#options.egress, + compatibilityDate: this.#options.compatibilityDate, + compatibilityFlags: this.#options.compatibilityFlags, + maxStdioBytes: this.#options.maxStdioBytes, + maxSourceBytes: this.#options.maxSourceBytes, + onComplete: () => this.#finalize(record), + onError: (message) => this.#finalize(record, message), + }); + // The completion promise drives finalize (which writes the + // terminal SQL) and is no longer handed to a host lifetime + // hook. Observe it so a finalize rejection surfaces as a + // swallowed error rather than an unhandled rejection in the + // Durable Object. + void record.control.completion.catch(() => undefined); + } catch (error) { + record.control?.cancel(); + await record.control?.completion.catch(() => undefined); + await this.#finalize(record, error instanceof Error ? error.message : String(error)); + } + } + async getExec(input: { id: string; after?: number | "tail" }): Promise { assertExecutionId(input.id); this.#prune(); @@ -647,6 +902,7 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { #markSettled(id: string) { this.#running.delete(id); this.#syncKeepAlive(); + if (this.#recovering.delete(id)) this.#drainRecoveries(); } // Hold the Durable Object in memory while any execution runs. Scheduling @@ -711,30 +967,12 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { #recordFor(id: string): ExecutionRecord | undefined { const cached = this.#records.get(id); if (cached) return cached; - const row = this.#host.db.one<{ status: ExecutionRecord["status"] }>( - `SELECT status FROM workspace_runtime_executions WHERE backend = ? AND id = ?`, - this.#backendId, - id, - ); - if (!row) return undefined; - const events = this.#host.db - .all<{ seq: number; name: WorkspaceRuntimeEvent["name"]; payload: Uint8Array }>( - `SELECT seq, name, payload FROM workspace_runtime_events - WHERE backend = ? AND execution_id = ? ORDER BY seq`, - this.#backendId, - id, - ) - .map((event) => decodeEvent(id, event.seq, event.name, event.payload)); - const record: ExecutionRecord = { - id, - status: row.status, - events, - subscribers: new Set(), - }; - if (record.status === "running") { - // Every live execution is cached before its Dynamic Worker starts. A - // durable running row without a cached record is therefore an orphan - // left by restart or a failed terminal transaction. + const record = this.#loadRecord(id); + if (record?.status === "running") { + // Every live execution is cached before its Dynamic Worker starts, + // and recovery caches the ones it resumes. A durable running row + // without a cached record is therefore an orphan left by a failed + // terminal transaction. this.#records.set(id, record); this.#finish(record, "failed", [ { @@ -751,6 +989,24 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { return record; } + #loadRecord(id: string): ExecutionRecord | undefined { + const row = this.#host.db.one<{ status: ExecutionRecord["status"] }>( + `SELECT status FROM workspace_runtime_executions WHERE backend = ? AND id = ?`, + this.#backendId, + id, + ); + if (!row) return undefined; + const events = this.#host.db + .all<{ seq: number; name: WorkspaceRuntimeEvent["name"]; payload: Uint8Array }>( + `SELECT seq, name, payload FROM workspace_runtime_events + WHERE backend = ? AND execution_id = ? ORDER BY seq`, + this.#backendId, + id, + ) + .map((event) => decodeEvent(id, event.seq, event.name, event.payload)); + return { id, status: row.status, events, subscribers: new Set() }; + } + // Drain the runner's framed output stream live, ingesting each frame // as it arrives. Resolves when the stream closes, which is the // signal the runner's evaluate awaits before returning. @@ -788,12 +1044,39 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { // stream but only settle the terminal state in #finalize. #ingestFrame(record: ExecutionRecord, frame: RuntimeFrame) { if (record.status !== "running") return; + // Saved with the output in one transaction, so output never outlives + // the log of the events that led to it. + if (frame.log !== undefined) { + try { + record.journal?.record(frame.log.from, frame.log.events); + } catch (error) { + this.#diverge(record, error instanceof Error ? error.message : String(error)); + return; + } + } if (frame.name === "stdout" || frame.name === "stderr") { + let value = frame.value; + const replayed = record.replayed?.[frame.name]; + if (replayed !== undefined && replayed.offset < replayed.bytes.byteLength) { + const end = Math.min(replayed.bytes.byteLength, replayed.offset + value.byteLength); + const expected = replayed.bytes.subarray(replayed.offset, end); + if (!bytesEqual(value.subarray(0, expected.byteLength), expected)) { + this.#diverge( + record, + "The execution took a different path when it resumed after an interruption: " + + `it wrote different ${frame.name} from the first time.`, + ); + return; + } + replayed.offset = end; + value = value.subarray(expected.byteLength); + if (value.byteLength === 0) return; + } this.#append(record, { id: record.id, seq: record.events.length + 1, name: frame.name, - value: frame.value, + value, }); } else { record.exitCode = frame.code; @@ -813,6 +1096,21 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { } async #finalizeOnce(record: ExecutionRecord, errorMessage?: string) { + const expected = record.journal?.expectedCall(); + if (errorMessage === undefined && expected !== undefined) { + errorMessage = + "The execution took a different path when it resumed after an interruption: " + + `it finished without calling ${expected} again.`; + } + const unwritten = (["stdout", "stderr"] as const).find((name) => { + const replayed = record.replayed?.[name]; + return replayed !== undefined && replayed.offset < replayed.bytes.byteLength; + }); + if (errorMessage === undefined && unwritten !== undefined) { + errorMessage = + "The execution took a different path when it resumed after an interruption: " + + `it finished without writing all the ${unwritten} it wrote the first time.`; + } try { await record.bridge?.cancelAndDrain(); } catch (error) { @@ -937,6 +1235,7 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { record.id, ); for (const event of events) this.#persistEvent(record.id, event); + clearDurable(this.#host.db, this.#backendId, record.id); }); } catch { // The execution must settle even if durable storage is unavailable. @@ -1039,6 +1338,18 @@ class JavaScriptBackendHandle implements WorkspaceModuleBackendHandle { this.#backendId, cutoff, ); + // Saved start data and journals belong to running executions only. + // Anything else was left by a terminal state that wasn't persisted. + for (const table of DURABLE_TABLES) { + this.#host.db.run( + `DELETE FROM ${table} + WHERE backend = ? AND execution_id NOT IN ( + SELECT id FROM workspace_runtime_executions WHERE backend = ? AND status = 'running' + )`, + this.#backendId, + this.#backendId, + ); + } const excess = this.#host.db.all<{ id: string }>( `SELECT id FROM workspace_runtime_executions WHERE backend = ? AND status != 'running' @@ -1207,7 +1518,8 @@ function startJavaScriptExecution(options: { function runtimeWorkerModule(entryName: string, maxStdioBytes: number) { return ` import { WorkerEntrypoint } from "cloudflare:workers"; - import { install, moduleScopeRefusals } from "workspace-capabilities.js"; + import { call, install, moduleScopeRefusals } from "workspace-capabilities.js"; + import { install as installDurable, takeLog } from "workspace-durable.js"; // A promise nobody awaits would otherwise fail without a trace while // the run reports success. Track the rejections nothing handled, and @@ -1286,6 +1598,8 @@ function runtimeWorkerModule(entryName: string, maxStdioBytes: number) { // task awaits writeChain before closing. let writeChain = Promise.resolve(); const enqueue = (frame) => { + const log = takeLog(); + if (log !== undefined) frame.log = log; const bytes = encoder.encode(JSON.stringify(frame) + "\\n"); writeChain = writeChain.then(() => writer.write(bytes)).catch(() => {}); }; @@ -1356,6 +1670,11 @@ function runtimeWorkerModule(entryName: string, maxStdioBytes: number) { return ""; }; install(host); + // A replayable run gets a clock, timers, random numbers and fetch + // that replay the way they went the first time. + // The runner keeps the real timer for itself. + const realSetTimeout = globalThis.setTimeout; + if (context && context.durable) installDurable(context.durable, host, call); // Hand the readable end to the host, which drains it live while // this call stays in flight. Keeping evaluate in flight is what // holds the host bridge stub alive for the whole run; frames @@ -1368,7 +1687,7 @@ function runtimeWorkerModule(entryName: string, maxStdioBytes: number) { ? await module.default(input) : module.default ?? null; // Rejections are reported after the microtask queue drains. - await new Promise((resolve) => setTimeout(resolve, 0)); + await new Promise((resolve) => realSetTimeout(resolve, 0)); if (moduleScopeRefusals.length > 0) throw moduleScopeRefusals[0]; for (const reason of unhandled.values()) { const message = reason instanceof Error ? reason.message : String(reason); @@ -1485,6 +1804,12 @@ function assertPositiveInteger(value: number, name: string) { } } +function assertInterruptPolicy(value: unknown): asserts value is WorkspaceRuntimeInterruptPolicy { + if (!INTERRUPT_POLICIES.includes(value as WorkspaceRuntimeInterruptPolicy)) { + throw new Error('onInterrupt must be "fail", "restart" or "replay".'); + } +} + function assertExecutionId(id: string) { if (typeof id !== "string" || id.length === 0) { throw new Error("Workspace runtime execution id must be a non-empty string."); @@ -1517,3 +1842,23 @@ function disposeQuietly(value: { [Symbol.dispose]?: () => void }) { value[Symbol.dispose]?.(); } catch {} } + +function recordedOutput( + events: readonly WorkspaceRuntimeEvent[], + name: "stdout" | "stderr", +): Uint8Array { + const chunks = events.flatMap((event) => (event.name === name ? [event.value] : [])); + const bytes = new Uint8Array(chunks.reduce((total, chunk) => total + chunk.byteLength, 0)); + let offset = 0; + for (const chunk of chunks) { + bytes.set(chunk, offset); + offset += chunk.byteLength; + } + return bytes; +} + +function bytesEqual(a: Uint8Array, b: Uint8Array): boolean { + if (a.byteLength !== b.byteLength) return false; + for (let i = 0; i < a.byteLength; i++) if (a[i] !== b[i]) return false; + return true; +} diff --git a/packages/computer/src/index.ts b/packages/computer/src/index.ts index 2910db30..f8bd47dd 100644 --- a/packages/computer/src/index.ts +++ b/packages/computer/src/index.ts @@ -104,6 +104,7 @@ export type { WorkspaceRuntimeExecHandle, WorkspaceRuntimeExecOptions, WorkspaceRuntimeGetOptions, + WorkspaceRuntimeInterruptPolicy, WorkspaceRuntimeKillOptions, WorkspaceRuntimeLoader, WorkspaceRuntimeOutputOptions, @@ -112,6 +113,7 @@ export type { WorkspaceRuntimeTruncation, WorkspaceRuntimeValue, } from "./runtime/types.js"; +export { unsafeToRepeat } from "./runtime/replay-safety.js"; export { decodeRuntimeEvents, encodeRuntimeEvent } from "./runtime/wire.js"; export { type RawShellValue, type ShellValue, sh, shellQuote } from "./sh.js"; export type { ExecEncoding, ExecSyncResult, KillSignal } from "./shell.js"; diff --git a/packages/computer/src/modules/artifacts.test.ts b/packages/computer/src/modules/artifacts.test.ts index 90fd8ab3..b22e36c0 100644 --- a/packages/computer/src/modules/artifacts.test.ts +++ b/packages/computer/src/modules/artifacts.test.ts @@ -42,6 +42,7 @@ function callContext(access: "read" | "read-write" = "read-write"): WorkspaceMod signal: new AbortController().signal, deadline: Date.now() + 60_000, access, + callId: "test-call", resolvePath: async (path) => path, }; } diff --git a/packages/computer/src/modules/assets.test.ts b/packages/computer/src/modules/assets.test.ts index 170f33b6..dd2f88b5 100644 --- a/packages/computer/src/modules/assets.test.ts +++ b/packages/computer/src/modules/assets.test.ts @@ -40,6 +40,7 @@ function callContext( signal: new AbortController().signal, deadline: Date.now() + 60_000, access: "read-write", + callId: "test-call", 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 592418a8..e9947e3e 100644 --- a/packages/computer/src/modules/container.test.ts +++ b/packages/computer/src/modules/container.test.ts @@ -89,6 +89,7 @@ function callContext( signal: new AbortController().signal, deadline: Date.now() + 60_000, access: "read-write", + callId: "test-call", resolvePath: async (path) => path, ...overrides, }; diff --git a/packages/computer/src/modules/tools.test.ts b/packages/computer/src/modules/tools.test.ts index f8b9709a..6ccf4d4a 100644 --- a/packages/computer/src/modules/tools.test.ts +++ b/packages/computer/src/modules/tools.test.ts @@ -17,6 +17,7 @@ function callContext( signal: new AbortController().signal, deadline: Date.now() + 60_000, access: "read-write", + callId: "test-call", resolvePath: async (path) => path, ...overrides, }; diff --git a/packages/computer/src/runtime/bridge.test.ts b/packages/computer/src/runtime/bridge.test.ts index 075e2d26..909045bf 100644 --- a/packages/computer/src/runtime/bridge.test.ts +++ b/packages/computer/src/runtime/bridge.test.ts @@ -1,7 +1,9 @@ import { describe, expect, it, vi } from "vitest"; -import { type BridgeResponse, WorkspaceRuntimeBridge } from "./bridge.js"; +import { type BridgeJournal, type BridgeResponse, WorkspaceRuntimeBridge } from "./bridge.js"; import type { WorkspaceRuntimeCapability } from "./capability.js"; +import { unsafeToRepeat } from "./replay-safety.js"; +import type { WorkspaceModuleCallContext } from "./types.js"; const encoder = new TextEncoder(); const args = ["value"]; @@ -245,3 +247,218 @@ describe("WorkspaceRuntimeBridge assertResult", () => { ); }); }); + +describe("WorkspaceRuntimeBridge journal", () => { + function journaled(steps: ReturnType[]) { + const commits: number[] = []; + const journal: BridgeJournal = { + begin: () => steps.shift() ?? { kind: "live", seq: 99, rerun: false }, + commit: (seq) => { + commits.push(seq); + return { at: 1234, ordinal: 7 }; + }, + record: (position, events) => { + if (!Array.isArray(events)) throw new Error("malformed event log"); + logs.push([position, events]); + }, + diverged: (message) => diverged.push(message), + stalled: () => diverged.push("stalled"), + }; + const diverged: string[] = []; + const logs: unknown[] = []; + const contexts: WorkspaceModuleCallContext[] = []; + const target = new WorkspaceRuntimeBridge({} as WorkspaceRuntimeCapability, { + maxTotalResponseBytes: 4, + hostModules: new Map([ + [ + "ws:test", + { + run: async (_args: unknown, context: WorkspaceModuleCallContext) => { + contexts.push(context); + return "ok"; + }, + charge: unsafeToRepeat(async () => "c"), + }, + ], + ]), + journal, + callIdPrefix: "exec-1", + }); + return { target, commits, diverged, contexts, logs }; + } + + it("answers a replayed call from the journal and journals a live one", async () => { + const { target, commits } = journaled([ + { kind: "replay", response: { result: "no" }, at: 5, ordinal: 1 }, + { kind: "live", seq: 2, rerun: false }, + ]); + await expect(target.call("host/ws:test.run", args)).resolves.toEqual({ + result: "no", + at: 5, + ordinal: 1, + }); + await expect(target.call("host/ws:test.run", args)).resolves.toEqual({ + result: "ok", + at: 1234, + ordinal: 7, + }); + expect(commits).toEqual([2]); + }); + + it("saves the event log a call carries before making the call", async () => { + const { target, logs } = journaled([{ kind: "live", seq: 1, rerun: false }]); + await target.call("host/ws:test.run", args, { from: 3, events: [4, -1] }); + expect(logs).toEqual([[3, [4, -1]]]); + await expect( + message(target.call("host/ws:test.run", args, { from: 5, events: "x" as never })), + ).resolves.toContain("malformed event log"); + }); + + it("gives a journaled call the same id when it runs again", async () => { + const { target, contexts } = journaled([ + { kind: "live", seq: 7, rerun: false }, + { kind: "live", seq: 7, rerun: true }, + ]); + await target.call("host/ws:test.run", args); + await target.call("host/ws:test.run", args); + expect(contexts.map((context) => context.callId)).toEqual(["exec-1:7", "exec-1:7"]); + }); + + it("refuses to run an unsafe call again after an interruption", async () => { + const { target, diverged } = journaled([ + { kind: "live", seq: 1, rerun: false }, + { kind: "live", seq: 1, rerun: true }, + { kind: "live", seq: 2, rerun: 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( + "interrupted while ws:test.charge was running, which isn't safe to repeat", + ); + expect(diverged).toHaveLength(1); + // A function that isn't marked runs again. + await expect(target.call("host/ws:test.run", args)).resolves.toMatchObject({ result: "ok" }); + }); + + it("saves an event log sent on its own", () => { + const { target, logs } = journaled([]); + target.recordLog({ from: 0, events: [-1, -2] }); + expect(logs).toEqual([[0, [-1, -2]]]); + }); + + it("passes a stalled replay to the journal", () => { + const { target, diverged } = journaled([]); + target.reportStall(); + expect(diverged).toEqual(["stalled"]); + }); + + it("counts replayed responses against the cumulative response budget", async () => { + const { target } = journaled([ + { kind: "replay", response: { result: "abc" }, at: 5 }, + { kind: "live", seq: 2 }, + ]); + await target.call("host/ws:test.run", args); + await expect(message(target.call("host/ws:test.run", args))).resolves.toContain( + "responses exceed 4 bytes", + ); + }); + + it("doesn't count a replayed budget error, as the first attempt didn't", async () => { + const { target } = journaled([ + { + kind: "replay", + response: { + error: { message: "Workspace execution capability responses exceed 4 bytes." }, + }, + at: 5, + }, + { kind: "live", seq: 2, rerun: false }, + ]); + await target.call("host/ws:test.run", args); + await expect(message(target.call("host/ws:test.run", args))).resolves.toBeUndefined(); + }); +}); + +describe("WorkspaceRuntimeBridge fetch", () => { + it("keeps fetch out of the capability call budgets", async () => { + const target = new WorkspaceRuntimeBridge({} as WorkspaceRuntimeCapability, { + maxCalls: 1, + maxTotalResponseBytes: 8, + hostModules: new Map([["ws:test", { run: async () => "ok" }]]), + fetch: async () => new Response("a response well over eight bytes"), + }); + for (let index = 0; index < 3; index += 1) { + await expect(message(target.call("net.fetch", [request]))).resolves.toBeUndefined(); + } + await expect(message(target.call("host/ws:test.run", args))).resolves.toBeUndefined(); + await expect(message(target.call("host/ws:test.run", args))).resolves.toContain( + "exceeds 1 capability calls", + ); + }); + + it("queues fetches past six open at once instead of refusing them", async () => { + let open = 0; + let peak = 0; + const target = new WorkspaceRuntimeBridge({} as WorkspaceRuntimeCapability, { + maxConcurrentCalls: 2, + fetch: async () => { + open += 1; + peak = Math.max(peak, open); + await new Promise((resolve) => setTimeout(resolve, 5)); + open -= 1; + return new Response("ok"); + }, + }); + const responses = await Promise.all( + Array.from({ length: 10 }, () => message(target.call("net.fetch", [request]))), + ); + expect(responses).toEqual(Array.from({ length: 10 }, () => undefined)); + expect(peak).toBe(6); + }); + + const request = { + url: "https://example.test/a", + method: "POST", + redirect: "manual", + headers: [["x-a", "1"]], + body: encoder.encode("hi"), + }; + + it("makes the request through the host and buffers the response", async () => { + const seen: Request[] = []; + const target = new WorkspaceRuntimeBridge({} as WorkspaceRuntimeCapability, { + fetch: async (incoming) => { + seen.push(incoming); + return new Response("pong", { status: 201, headers: { "x-b": "2" } }); + }, + }); + const response = await target.call("net.fetch", [request]); + expect(response).toMatchObject({ + result: { status: 201, headers: expect.arrayContaining([["x-b", "2"]]) }, + }); + const result = "result" in response ? (response.result as { body: Uint8Array }) : undefined; + expect(new TextDecoder().decode(result?.body)).toBe("pong"); + expect(seen[0]?.method).toBe("POST"); + expect(seen[0]?.redirect).toBe("manual"); + expect(seen[0]?.headers.get("x-a")).toBe("1"); + expect(await seen[0]?.text()).toBe("hi"); + }); + + it("refuses when the isolate has no network access", async () => { + const target = new WorkspaceRuntimeBridge({} as WorkspaceRuntimeCapability, {}); + await expect(message(target.call("net.fetch", [request]))).resolves.toContain( + "no network access", + ); + }); + + it("refuses a response body over the payload limit", async () => { + const target = new WorkspaceRuntimeBridge({} as WorkspaceRuntimeCapability, { + maxPayloadBytes: 256, + fetch: async () => new Response("x".repeat(1024)), + }); + await expect(message(target.call("net.fetch", [request]))).resolves.toContain( + "exceeds 256 bytes", + ); + }); +}); diff --git a/packages/computer/src/runtime/bridge.ts b/packages/computer/src/runtime/bridge.ts index b1f18395..b1777893 100644 --- a/packages/computer/src/runtime/bridge.ts +++ b/packages/computer/src/runtime/bridge.ts @@ -2,6 +2,7 @@ import { RpcTarget } from "cloudflare:workers"; import { utf8ByteLength, utf8Prefix } from "../text-truncation.js"; import { assertRuntimeValue, type WorkspaceRuntimeCapability } from "./capability.js"; +import { isUnsafeToRepeat } from "./replay-safety.js"; import type { WorkspaceModuleCallContext, WorkspaceModuleFunctions, @@ -19,7 +20,13 @@ export class WorkspaceRuntimeBridge extends RpcTarget { readonly #maxTotalResponseBytes: number; readonly #maxResultBytes: number; readonly #onAttachOutput?: (readable: ReadableStream) => Promise; + readonly #journal?: BridgeJournal; + readonly #fetch?: (request: Request) => Promise; + readonly #callIdPrefix: string; readonly #inFlight = new Set>(); + #fetchesInFlight = 0; + #fetchesRunning = 0; + readonly #fetchQueue: Array<() => void> = []; readonly #abortControllers = new Set(); #cancelled = false; #callTimedOut = false; @@ -39,6 +46,12 @@ export class WorkspaceRuntimeBridge extends RpcTarget { maxTotalResponseBytes?: number; maxResultBytes?: number; onAttachOutput?: (readable: ReadableStream) => Promise; + journal?: BridgeJournal; + // Makes network requests for the isolate's journaled fetch. Absent + // when the isolate has no network access. + fetch?: (request: Request) => Promise; + // Prefix of every call's callId, such as the execution id. + callIdPrefix?: string; } = {}, ) { super(); @@ -52,6 +65,9 @@ export class WorkspaceRuntimeBridge extends RpcTarget { this.#maxTotalResponseBytes = integrations.maxTotalResponseBytes ?? 8 * 1024 * 1024; this.#maxResultBytes = integrations.maxResultBytes ?? 1024 * 1024; this.#onAttachOutput = integrations.onAttachOutput; + this.#journal = integrations.journal; + this.#fetch = integrations.fetch; + this.#callIdPrefix = integrations.callIdPrefix ?? crypto.randomUUID(); } // Drain the runner's framed output stream. The isolate passes the @@ -80,20 +96,43 @@ export class WorkspaceRuntimeBridge extends RpcTarget { // as real values through Workers RPC; this method is the proxy that // keeps the limits on them, so untrusted code cannot overload the // Durable Object with calls, concurrency, time, or bytes. - call(name: string, args: unknown[]): Promise { + call(name: string, args: unknown[], log?: BridgeLogChunk): Promise { const reject = (message: string) => Promise.resolve(boundedError(new Error(message), this.#maxPayloadBytes)); if (this.#cancelled) return reject("Workspace execution is being cancelled."); if (typeof name !== "string" || !Array.isArray(args)) { return reject("Workspace capability calls take a name and an argument list."); } + // A durable run's events since its last call, saved before the call + // that may depend on them. + if (log !== undefined && log !== null && this.#journal !== undefined) { + try { + this.#journal.record(log.from, log.events); + } catch (error) { + return Promise.resolve(boundedError(error, this.#maxPayloadBytes)); + } + } let requestBytes: number; try { requestBytes = measureValue(args, this.#maxPayloadBytes, "request"); } catch (error) { return Promise.resolve(boundedError(error, this.#maxPayloadBytes)); } - if (this.#inFlight.size >= this.#maxConcurrentCalls) { + // fetch only comes through here in a replayable run, so it keeps the + // limits it has anywhere else: none of the capability budgets, and a + // queue rather than a refusal when too many are open at once. + const isFetch = name === "net.fetch"; + if (!isFetch) return this.#capabilityCall(name, args, requestBytes, reject); + return this.#dispatch(name, args, reject, true); + } + + #capabilityCall( + name: string, + args: unknown[], + requestBytes: number, + reject: (message: string) => Promise, + ): Promise { + if (this.#inFlight.size - this.#fetchesInFlight >= this.#maxConcurrentCalls) { return reject( `Workspace execution exceeds ${this.#maxConcurrentCalls} concurrent capability calls.`, ); @@ -108,6 +147,37 @@ export class WorkspaceRuntimeBridge extends RpcTarget { } this.#calls += 1; this.#requestBytes += requestBytes; + return this.#dispatch(name, args, reject, false); + } + + #dispatch( + name: string, + args: unknown[], + reject: (message: string) => Promise, + isFetch: boolean, + ): Promise { + // A durable execution answers a call it already made from its + // journal, so the operation doesn't run twice. + const step = this.#journal?.begin(name, args); + if (step?.kind === "replay") { + // The first attempt counted this response, and the replay repeats + // that attempt, so it counts again. The budget error wasn't counted. + if (!isFetch && !this.#isBudgetError(step.response)) { + this.#responseBytes += replayedBytes(step.response, this.#maxPayloadBytes); + } + return Promise.resolve({ ...step.response, at: step.at, ordinal: step.ordinal }); + } + if (step?.kind === "diverge") { + this.#journal?.diverged(step.message); + return reject(step.message); + } + if (step?.kind === "live" && step.rerun && isUnsafeToRepeat(this.#hostFunction(name))) { + const message = `The execution was interrupted while ${name.slice("host/".length)} was running, which isn't safe to repeat.`; + this.#journal?.diverged(message); + return reject(message); + } + // Stable across attempts for a journaled call, unique otherwise. + const callId = `${this.#callIdPrefix}:${step?.kind === "live" ? step.seq : crypto.randomUUID()}`; const abort = new AbortController(); this.#abortControllers.add(abort); const deadline = Date.now() + this.#maxCallDurationMs; @@ -116,11 +186,13 @@ export class WorkspaceRuntimeBridge extends RpcTarget { return this.#callHostModule(name, args, { signal: abort.signal, deadline, + callId, access: this.#capability.access, resolvePath: (path, options) => this.#capability.resolveConfined(path, options?.allowMissing ?? false), }); } + if (name === "net.fetch") return this.#fetchFor(args[0], abort.signal); const operation = name.startsWith("fs.") ? name.slice(3) : name; switch (operation) { case "readFile": @@ -193,26 +265,58 @@ export class WorkspaceRuntimeBridge extends RpcTarget { // the accepted host operation itself. This prevents a late mutation // after an exit event when the underlying API cannot be aborted. this.#inFlight.add(operation); + if (isFetch) this.#fetchesInFlight += 1; void operation.finally(() => { this.#inFlight.delete(operation); + if (isFetch) this.#fetchesInFlight -= 1; this.#abortControllers.delete(abort); }); // 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.then((response) => { - const bytes = "result" in response ? response.bytes : errorBytes(response); - if (this.#responseBytes + bytes > this.#maxTotalResponseBytes) { - return boundedError( - new Error( - `Workspace execution capability responses exceed ${this.#maxTotalResponseBytes} bytes.`, - ), - this.#maxPayloadBytes, - ); - } - this.#responseBytes += bytes; - return "result" in response ? { result: response.result } : response; - }); + return call + .then((response) => { + if (isFetch) return "result" in response ? { result: response.result } : response; + const bytes = "result" in response ? response.bytes : errorBytes(response); + if (this.#responseBytes + bytes > this.#maxTotalResponseBytes) { + return this.#budgetError(); + } + this.#responseBytes += bytes; + return "result" in response ? { result: response.result } : response; + }) + .then((response) => { + // 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 + // its entry commit in the same implicit transaction. + if (step?.kind !== "live" || this.#journal === undefined) return response; + return { ...response, ...this.#journal.commit(step.seq, response) }; + }); + } + + #budgetError() { + return boundedError( + new Error( + `Workspace execution capability responses exceed ${this.#maxTotalResponseBytes} bytes.`, + ), + this.#maxPayloadBytes, + ); + } + + #isBudgetError(response: BridgeResponse): boolean { + return "error" in response && response.error.message === this.#budgetError().error.message; + } + + // The isolate is waiting for a journaled call its replay never made. + reportStall(): void { + this.#journal?.stalled(); + } + + // Part of a durable run's event log, sent on its own when it grows long + // between calls and output. + recordLog(log: BridgeLogChunk): void { + if (this.#cancelled) return; + this.#journal?.record(log?.from, log?.events); } async cancelAndDrain(): Promise { @@ -231,22 +335,114 @@ export class WorkspaceRuntimeBridge extends RpcTarget { // from a specifier such as `ws:a.b`. Own-property checks keep // isolate code from reaching `toString` or other inherited members. async #callHostModule(name: string, args: unknown[], context: WorkspaceModuleCallContext) { + const fn = this.#hostFunction(name); + if (fn === undefined) { + throw new Error(`Unknown Workspace host module call ${JSON.stringify(name)}.`); + } + // Called on its module, so a method that uses `this` still works. + const result = await fn.call(this.#hostModuleOf(name), hostArguments(args), context); + return hostValue(result ?? null); + } + + #hostModuleOf(name: string) { + if (!name.startsWith("host/")) return undefined; const target = name.slice("host/".length); const dot = target.lastIndexOf("."); - const specifier = dot === -1 ? "" : target.slice(0, dot); - const functionName = dot === -1 ? "" : target.slice(dot + 1); - const functions = this.#hostModules.get(specifier); + return dot === -1 ? undefined : this.#hostModules.get(target.slice(0, dot)); + } + + #hostFunction(name: string) { + const functions = this.#hostModuleOf(name); + const functionName = name.slice(name.lastIndexOf(".") + 1); const fn = functions !== undefined && Object.hasOwn(functions, functionName) ? functions[functionName] : undefined; - if (typeof fn !== "function") { - throw new Error(`Unknown Workspace host module call ${JSON.stringify(name)}.`); + return typeof fn === "function" ? fn : undefined; + } + + // A fetch the isolate journals goes through here: the request crosses + // as plain data, the host makes it through the backend's egress, and + // the response comes back buffered. + async #fetchFor(spec: unknown, signal: AbortSignal) { + if (this.#fetch === undefined) throw new Error("Workspace code has no network access."); + const fetch = this.#fetch; + const request = parseFetchRequest(spec, signal); + // A Worker keeps at most six connections open and queues the rest, + // which also bounds how many buffered bodies this object holds. + while (this.#fetchesRunning >= MAX_OPEN_FETCHES) { + await new Promise((resolve) => this.#fetchQueue.push(resolve)); } - // Called on its module, so a method that uses `this` still works. - const result = await fn.call(functions, hostArguments(args), context); - return hostValue(result ?? null); + this.#fetchesRunning += 1; + try { + signal.throwIfAborted(); + const response = await fetch(request); + return { + status: response.status, + statusText: response.statusText, + headers: [...response.headers], + url: response.url, + redirected: response.redirected, + body: await readLimited(response, this.#maxPayloadBytes), + }; + } finally { + this.#fetchesRunning -= 1; + this.#fetchQueue.shift()?.(); + } + } +} + +const MAX_OPEN_FETCHES = 6; + +function parseFetchRequest(spec: unknown, signal: AbortSignal): Request { + if (spec === null || typeof spec !== "object") throw new Error("fetch takes a request object."); + const { url, method, headers, body, redirect } = spec as Record; + if (typeof url !== "string" || typeof method !== "string") { + throw new Error("fetch takes a request with a url and a method."); + } + if ( + !Array.isArray(headers) || + !headers.every((pair) => Array.isArray(pair) && pair.length === 2) + ) { + throw new Error("fetch request headers must be name and value pairs."); + } + if (body !== null && body !== undefined && !(body instanceof Uint8Array)) { + throw new Error("fetch request bodies must be bytes."); + } + return new Request(url, { + method, + headers: headers.map(([name, value]): [string, string] => [String(name), String(value)]), + body: (body ?? null) as BodyInit | null, + // The Request constructor rejects a value it doesn't know. + ...(typeof redirect === "string" ? { redirect: redirect as RequestRedirect } : {}), + signal, + }); +} + +// Read a response body, refusing one larger than the call's payload limit +// before reading all of it. +async function readLimited(response: Response, maxBytes: number): Promise { + if (response.body === null) return new Uint8Array(0); + const reader = response.body.getReader(); + const chunks: Uint8Array[] = []; + let total = 0; + for (;;) { + const { value, done } = await reader.read(); + if (done) break; + total += value.byteLength; + if (total > maxBytes) { + await reader.cancel().catch(() => undefined); + throw new Error(`fetch response body exceeds ${maxBytes} bytes.`); + } + chunks.push(value); + } + const bytes = new Uint8Array(total); + let offset = 0; + for (const chunk of chunks) { + bytes.set(chunk, offset); + offset += chunk.byteLength; } + return bytes; } function hostArguments(args: unknown[]): WorkspaceRuntimeValue[] { @@ -303,12 +499,60 @@ function decodeBytes(value: unknown): string | Uint8Array { return value instanceof Uint8Array ? value : String(value); } +function replayedBytes(response: BridgeResponse, maxPayloadBytes: number): number { + if (!("result" in response)) return errorBytes(response); + try { + return measureValue(response.result, maxPayloadBytes, "response"); + } catch { + return maxPayloadBytes; + } +} + /** What the bridge sends back for one call: a value, or a bounded error. */ -export type BridgeResponse = +export type BridgeResponse = ( | { readonly result: unknown } | { readonly error: { readonly message: string; readonly code?: string; readonly path?: string }; - }; + } +) & { + /** + * When a journaled call's response was delivered. A durable execution's + * clock reads this, so it reads the same on replay. + */ + readonly at?: number; + /** Numbers a journaled response in the order responses were committed. */ + readonly ordinal?: number; +}; + +/** Part of a durable run's event log, starting at position `from`. */ +export interface BridgeLogChunk { + readonly from: number; + readonly events: readonly number[]; +} + +/** Records a durable execution's calls and answers them on replay. */ +export interface BridgeJournal { + begin( + name: string, + args: readonly unknown[], + ): + | { + readonly kind: "replay"; + readonly response: BridgeResponse; + readonly at: number; + readonly ordinal: number; + } + | { readonly kind: "live"; readonly seq: number; readonly rerun: boolean } + | { 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 }; + /** 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. */ + diverged(message: string): void; + /** Called when the isolate waits for a journaled call it never makes. */ + stalled(): void; +} // A response before the per-execution budget check, carrying its size. type MeasuredResponse = diff --git a/packages/computer/src/runtime/egress.ts b/packages/computer/src/runtime/egress.ts index 440be335..1c063632 100644 --- a/packages/computer/src/runtime/egress.ts +++ b/packages/computer/src/runtime/egress.ts @@ -18,3 +18,20 @@ export function dynamicWorkerEgress(policy: WorkspaceEgressPolicy): { return { globalOutbound: policy.gateway }; } } + +/** + * How the host makes a request on an isolate's behalf, under the same + * policy the isolate's own `fetch` would follow. Absent for `none`. + */ +export function hostEgressFetch(policy: WorkspaceEgressPolicy): { + fetch?: (request: Request) => Promise; +} { + switch (policy.mode) { + case "none": + return {}; + case "direct": + return { fetch: (request) => fetch(request) }; + case "http-gateway": + return { fetch: (request) => policy.gateway.fetch(request) }; + } +} diff --git a/packages/computer/src/runtime/replay-safety.ts b/packages/computer/src/runtime/replay-safety.ts new file mode 100644 index 00000000..dc8768bb --- /dev/null +++ b/packages/computer/src/runtime/replay-safety.ts @@ -0,0 +1,31 @@ +import type { WorkspaceModuleFunction } from "./types.js"; + +const UNSAFE_TO_REPEAT = Symbol.for("cloudflare.workspace.module.unsafeToRepeat"); + +/** + * Mark a host module function as unsafe to run twice. + * + * A replayable run interrupted while one of its calls was in flight runs + * that call again when it resumes, since its response never reached the + * code. That's fine for reads and idempotent writes. A function that + * charges a card or sends a message once per call should be marked, so + * an interruption during it fails the run instead. A function that can + * deduplicate with `context.callId` doesn't need marking. + * + * ```ts + * modules: { + * "ws:billing": { charge: unsafeToRepeat(async ([order, cents]) => ...) }, + * } + * ``` + */ +export function unsafeToRepeat(fn: F): F { + Object.defineProperty(fn, UNSAFE_TO_REPEAT, { value: true }); + return fn; +} + +export function isUnsafeToRepeat(fn: unknown): boolean { + return ( + typeof fn === "function" && + (fn as unknown as Record)[UNSAFE_TO_REPEAT] === true + ); +} diff --git a/packages/computer/src/runtime/runtime.ts b/packages/computer/src/runtime/runtime.ts index fe47aa87..03f709d3 100644 --- a/packages/computer/src/runtime/runtime.ts +++ b/packages/computer/src/runtime/runtime.ts @@ -109,6 +109,7 @@ export class WorkspaceRuntime { stdin: options.stdin, timeoutMs: options.timeoutMs, sync: options.sync, + onInterrupt: options.onInterrupt, }); return wrapModuleHandle( runtime, @@ -157,6 +158,20 @@ export class WorkspaceRuntime { ); } + /** + * Resume or settle the runs a Durable Object restart interrupted. + * Backends do this when they first connect, which the next runtime + * call or `ready()` triggers. Call this from an alarm to resume runs + * without waiting for one. + */ + async recover(): Promise { + for (const [id, backend] of this.#options.backends) { + if (!isModuleBackend(backend)) continue; + const handle = await this.#options.backendHandle(id); + await handle.recover?.(); + } + } + 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 c1e7d76e..050873fc 100644 --- a/packages/computer/src/runtime/types.ts +++ b/packages/computer/src/runtime/types.ts @@ -22,6 +22,12 @@ export interface WorkspaceModuleCallContext { * @returns The confined absolute path. */ resolvePath(path: string, options?: { readonly allowMissing?: boolean }): Promise; + /** + * Identifies this call. When a replayable run is interrupted while the + * call is in flight and runs it again on resume, the second call gets + * the same id, so a function can use it as an idempotency key. + */ + readonly callId: string; } /** @@ -206,6 +212,12 @@ export interface WorkspaceRuntimeExecOptions stdin?: Uint8Array | string; timeoutMs?: number; sync?: "wait" | "defer"; + /** + * What happens to this run if the Durable Object restarts while it + * runs, as on a deploy or crash. Overrides the backend's + * `onInterrupt`. Backends that can't resume a run ignore it. + */ + onInterrupt?: WorkspaceRuntimeInterruptPolicy; /** * How much output `result()` keeps per stream. Output past it is cut * to its last lines and saved in full to a Workspace file. Overrides @@ -259,8 +271,18 @@ export interface ModuleExecutionInput { stdin?: Uint8Array | string; timeoutMs?: number; sync?: "wait" | "defer"; + onInterrupt?: WorkspaceRuntimeInterruptPolicy; } +/** + * What a backend does with a run its Durable Object restarted under. + * `fail` settles it as failed. `restart` runs it again from the start. + * `replay` runs it again and answers the calls it already made from a + * journal, so their effects aren't repeated. `restart` and `replay` are + * experimental and may change in any release. + */ +export type WorkspaceRuntimeInterruptPolicy = "fail" | "restart" | "replay"; + export interface ModuleExecutionEnvelope { id: string; // Identity of the process-local backend runtime that owns this @@ -288,6 +310,9 @@ export interface WorkspaceModuleBackendHandle { }): Promise; killExec(input: { id: string; signal?: KillSignal; runtimeId?: string }): Promise; disposeExec(input: { id: string; runtimeId?: string }): Promise; + // Settle the runs a restart interrupted. A backend that recovers + // when it connects needs nothing more here. + recover?(): 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/tools/ai-sdk/bindings.test.ts b/packages/computer/src/tools/ai-sdk/bindings.test.ts index c75a2ba4..233fe7c6 100644 --- a/packages/computer/src/tools/ai-sdk/bindings.test.ts +++ b/packages/computer/src/tools/ai-sdk/bindings.test.ts @@ -14,6 +14,7 @@ function callContext(): WorkspaceModuleCallContext { signal: new AbortController().signal, deadline: Date.now() + 60_000, access: "read-write", + callId: "test-call", 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 6b09ec6e..3894c1d3 100644 --- a/packages/computer/src/tools/pi-ai/bindings.test.ts +++ b/packages/computer/src/tools/pi-ai/bindings.test.ts @@ -17,6 +17,7 @@ function callContext( signal: new AbortController().signal, deadline: Date.now() + 60_000, access, + callId: "test-call", resolvePath: async (path) => path, }; } diff --git a/packages/computer/tests/script-runner-worker.ts b/packages/computer/tests/script-runner-worker.ts index b2c0a148..7cc3dc37 100644 --- a/packages/computer/tests/script-runner-worker.ts +++ b/packages/computer/tests/script-runner-worker.ts @@ -7,7 +7,7 @@ import { } from "../src/backends/worker-javascript/index.js"; import { createGitClient } from "../src/git/index.js"; import type { WorkspaceRuntimeValue, WorkspaceStub } from "../src/index.js"; -import { Workspace } 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"; import { createGitModule } from "../src/modules/git.js"; @@ -57,20 +57,38 @@ function fakeContainerBackend(): WorkspaceBackend { }; } +// Durable execution tests restart the object with ctx.abort(). Module +// 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(); +let gateOpen = false; +// Named gates a run can wait on after the first, for a second restart. +const opened = new Set(); + +function countCall(name: string): number { + const count = (durableCalls.get(name) ?? 0) + 1; + durableCalls.set(name, count); + return count; +} + // The Worker Loader resolves imports differently under the legacy and // the new module registry, so the same backend runs on both. const NEW_REGISTRY_BACKEND = "worker-javascript-new-registry"; +const NETWORK_BACKEND = "worker-javascript-network"; function javascriptBackend( env: Env, overrides: Partial< - Pick + Pick< + WorkerJavaScriptBackendOptions, + "id" | "compatibilityFlags" | "loader" | "egress" | "maxStdioBytes" + > > = {}, ) { return new WorkerJavaScriptBackend({ loader: env.LOADER, ...overrides, - maxStdioBytes: 64, + maxStdioBytes: overrides.maxStdioBytes ?? 64, maxCapabilityBytes: 1024, maxConcurrentCapabilityCalls: 2, modules: { @@ -93,6 +111,28 @@ function javascriptBackend( async delete(args) { return { deleted: args[0] ?? null }; }, + async tick(args) { + return countCall(String(args[0] ?? "tick")); + }, + // Never settles before the object restarts. + async gate() { + countCall("gate"); + if (!gateOpen) await new Promise(() => undefined); + return "open"; + }, + // Never settles before a restart opens it by name. + async hold(args) { + const name = String(args[0]); + countCall(name); + if (!opened.has(name)) await new Promise(() => undefined); + return "open"; + }, + // The gate, for a call that mustn't run twice. + unsafeGate: unsafeToRepeat(async () => { + countCall("gate"); + if (!gateOpen) await new Promise(() => undefined); + return "open"; + }), async invalidResult() { // SAFETY: The test hands the bridge a non-JSON value on purpose to check that it rejects it. return new Date() as never; @@ -104,6 +144,11 @@ function javascriptBackend( await new Promise((resolve) => setTimeout(resolve, 20)); return null; }, + // Settles after args[0] milliseconds and counts the call as args[1]. + async wait(args) { + await new Promise((resolve) => setTimeout(resolve, Number(args[0]))); + return countCall(String(args[1] ?? "wait")); + }, async marker() { return { __workspace_codec__: { version: 1, type: "bytes", data: [1] }, @@ -135,6 +180,11 @@ export class HostDO extends DurableObject { env.LOADER.load({ ...code, allowExperimental: true }), }, }), + javascriptBackend(env, { + id: NETWORK_BACKEND, + egress: { mode: "http-gateway", gateway: ctx.exports.TestGateway }, + maxStdioBytes: 4096, + }), fakeContainerBackend(), ], }); @@ -192,10 +242,10 @@ export class HostDO extends DurableObject { return handle.id; } - async getRuntime(id: string, resume?: "tail") { + async getRuntime(id: string, resume?: "tail", network?: boolean) { try { const handle = await this.#workspace.runtime.getExec(id, { - backend: "worker-javascript", + backend: network === true ? NETWORK_BACKEND : "worker-javascript", encoding: "utf8", resume, }); @@ -208,6 +258,44 @@ export class HostDO extends DurableObject { } } + async startDurable(input: { + source: string; + id: string; + onInterrupt?: "fail" | "restart" | "replay"; + network?: boolean; + }) { + gateOpen = false; + opened.clear(); + durableCalls.clear(); + await this.#workspace.fs.mkdir("/workspace", { recursive: true }); + const handle = await this.#workspace.runtime.exec(input.source, { + backend: input.network === true ? NETWORK_BACKEND : "worker-javascript", + id: input.id, + onInterrupt: input.onInterrupt, + }); + void handle.result().catch(() => undefined); + // Let the run reach the gate. + for (let attempt = 0; attempt < 500 && !durableCalls.has("gate"); attempt += 1) { + await new Promise((resolve) => setTimeout(resolve, 10)); + } + return handle.id; + } + + // Restart the object, as a deploy or crash would. + restart(open?: string) { + gateOpen = true; + if (open !== undefined) opened.add(open); + this.ctx.abort("test restart"); + } + + async recover() { + await this.#workspace.runtime.recover(); + } + + durableCalls() { + return Object.fromEntries(durableCalls); + } + killRuntime(id: string) { return this.#workspace.runtime.killExec(id, { backend: "worker-javascript" }); } @@ -256,6 +344,20 @@ class StdioProbeBridge extends RpcTarget { } } +// The network a run with egress reaches. It counts each request, so a +// test can tell whether a replay sent it again. +export class TestGateway extends WorkerEntrypoint { + override async fetch(request: Request) { + const count = countCall("fetch"); + return new Response( + `${request.method} ${new URL(request.url).pathname} ${await request.text()} #${count}`, + { + headers: { "x-count": String(count) }, + }, + ); + } +} + export default class extends WorkerEntrypoint { override async fetch(request: Request) { const url = new URL(request.url); @@ -412,12 +514,40 @@ export default class extends WorkerEntrypoint { const response = await stub.getRuntime( url.searchParams.get("id") ?? "missing", url.searchParams.get("resume") === "tail" ? "tail" : undefined, + url.searchParams.has("network"), ); return response.ok ? Response.json(response.result) : Response.json({ error: response.error }, { status: 400 }); } + if (url.pathname === "/durable-start") { + return Response.json({ + id: await stub.startDurable( + (await request.json()) as { + source: string; + id: string; + onInterrupt?: "fail" | "restart" | "replay"; + network?: boolean; + }, + ), + }); + } + + if (url.pathname === "/durable-restart") { + await stub.restart(url.searchParams.get("open") ?? undefined).catch(() => undefined); + return new Response(null, { status: 204 }); + } + + if (url.pathname === "/durable-recover") { + await stub.recover(); + return new Response(null, { status: 204 }); + } + + if (url.pathname === "/durable-calls") { + return Response.json(await stub.durableCalls()); + } + if (url.pathname === "/runtime-kill") { await stub.killRuntime(url.searchParams.get("id") ?? "missing"); return new Response(null, { status: 204 }); diff --git a/packages/computer/tests/script-runner.test.ts b/packages/computer/tests/script-runner.test.ts index 7796e133..56dc8bb5 100644 --- a/packages/computer/tests/script-runner.test.ts +++ b/packages/computer/tests/script-runner.test.ts @@ -332,11 +332,16 @@ describe("WorkspaceRuntime", () => { expect(JSON.parse(text).result.value).toEqual([ "delete", "echo", + "gate", + "hold", "invalidResult", "largeError", "marker", "slow", "sum", + "tick", + "unsafeGate", + "wait", ]); }); @@ -968,3 +973,371 @@ describe.each([ }); }); }); + +describe("durable executions", () => { + const source = ` + import fs from "node:fs/promises"; + import { tick, gate } from "ws:test-host"; + export default async function () { + const roll = Math.random(); + console.log("a"); + await fs.writeFile("/workspace/roll.txt", String(roll)); + const first = await tick("work"); + await gate(); + console.log("b"); + const second = await tick("work"); + return { roll, saved: await fs.readFile("/workspace/roll.txt", "utf8"), first, second }; + } + `; + + async function start( + id: string, + body: { source: string; onInterrupt?: string; network?: boolean }, + ) { + const response = await SELF.fetch("https://example.test/durable-start", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ id, ...body }), + }); + expect(response.status, await response.clone().text()).toBe(200); + await SELF.fetch("https://example.test/durable-restart", { method: "POST" }); + } + + async function result(id: string, options: { network?: boolean } = {}) { + const response = await SELF.fetch( + `https://example.test/runtime-get?id=${encodeURIComponent(id)}${options.network === true ? "&network" : ""}`, + ); + return (await response.json()) as { + status: string; + exitCode: number; + stdout: string; + stderr: string; + value?: unknown; + }; + } + + async function calls() { + return (await (await SELF.fetch("https://example.test/durable-calls")).json()) as Record< + string, + number + >; + } + + it("replays a run after a restart without repeating its calls or output", async () => { + await start("durable-replay", { source, onInterrupt: "replay" }); + const settled = await result("durable-replay"); + expect(settled).toMatchObject({ status: "completed", exitCode: 0, stdout: "a\nb\n" }); + const value = settled.value as { roll: number; saved: string; first: number; second: number }; + expect(value.saved).toBe(String(value.roll)); + expect(value).toMatchObject({ first: 1, second: 2 }); + // The first tick came from the journal. The gate was in flight, so + // it ran again. + expect(await calls()).toEqual({ work: 2, gate: 2 }); + }); + + it("replays the clock a run read before it was interrupted", async () => { + await start("durable-clock", { + source: ` + import fs from "node:fs/promises"; + import { gate, tick } from "ws:test-host"; + export default async function () { + const started = Date.now(); + await fs.writeFile("/workspace/clock.txt", String(started)); + await new Promise((resolve) => setTimeout(resolve, 5)); + const slept = new Date().getTime(); + await tick(slept - started >= 5 ? "slept" : "early"); + await gate(); + return { started, slept, later: Date.now() >= slept }; + } + `, + onInterrupt: "replay", + }); + const settled = await result("durable-clock"); + expect(settled, settled.stderr).toMatchObject({ + status: "completed", + value: { later: true }, + }); + expect(await calls()).toEqual({ slept: 1, gate: 2 }); + }); + + it("doesn't wait out a replayed run's sleeps again", async () => { + await start("durable-sleep", { + source: ` + import { gate, tick } from "ws:test-host"; + export default async function () { + const started = Date.now(); + for (let index = 0; index < 3; index += 1) { + await new Promise((resolve) => setTimeout(resolve, 1000)); + await tick("slept"); + } + await gate(); + return Date.now() - started; + } + `, + onInterrupt: "replay", + }); + const resumed = Date.now(); + const settled = await result("durable-sleep"); + expect(Date.now() - resumed).toBeLessThan(1000); + // The clock still counts the sleeps. + expect(settled).toMatchObject({ status: "completed" }); + expect(settled.value).toBeGreaterThanOrEqual(3000); + expect(await calls()).toEqual({ slept: 3, gate: 2 }); + }); + + it("restarts a run from the beginning", async () => { + await start("durable-restart", { source, onInterrupt: "restart" }); + const settled = await result("durable-restart"); + expect(settled).toMatchObject({ status: "completed", stdout: "a\na\nb\n" }); + expect(settled.stderr).toContain("restarted from the beginning"); + expect(settled.value).toMatchObject({ first: 2, second: 3 }); + expect(await calls()).toEqual({ work: 3, gate: 2 }); + }); + + it("fails a replay that takes a different path", async () => { + await start("durable-diverge", { + source: ` + import fs from "node:fs/promises"; + import { gate, tick } from "ws:test-host"; + export default async function () { + // Generated keys are the one source of randomness a replay + // doesn't repeat. + const key = await crypto.subtle.generateKey({ name: "HMAC", hash: "SHA-256" }, true, ["sign"]); + const raw = new Uint8Array(await crypto.subtle.exportKey("raw", key)); + await fs.writeFile("/workspace/" + raw.slice(0, 8).join("-") + ".txt", "x"); + await gate(); + await tick("after"); + } + `, + onInterrupt: "replay", + }); + const settled = await result("durable-diverge"); + expect(settled).toMatchObject({ status: "failed", exitCode: 1 }); + expect(settled.stderr).toContain("took a different path"); + expect(await calls()).toEqual({ gate: 1 }); + }); + + it("repeats a run's random numbers when it replays", async () => { + await start("durable-random", { + source: ` + import { gate } from "ws:test-host"; + export default async function () { + // Short enough to stay under the test backend's output limit. + const values = [ + Math.random().toFixed(8), + crypto.randomUUID().slice(0, 13), + crypto.getRandomValues(new Uint8Array(4)).join(","), + ]; + console.log(values.join(" ")); + await gate(); + return values; + } + `, + onInterrupt: "replay", + }); + const settled = await result("durable-random"); + // The replay checks the line it prints against the one recorded, and + // would fail on a different one. + expect(settled).toMatchObject({ status: "completed", exitCode: 0 }); + expect(settled.stdout).toBe(`${(settled.value as string[]).join(" ")}\n`); + }); + + it("delivers replayed responses in the order they first arrived", async () => { + await start("durable-order", { + source: ` + import { gate, wait } from "ws:test-host"; + export default async function () { + const slow = wait(80, "slow").then(() => "slow"); + const fast = wait(5, "fast").then(() => "fast"); + const winner = await Promise.race([slow, fast]); + await Promise.all([slow, fast]); + console.log(winner); + await gate(); + return winner; + } + `, + onInterrupt: "replay", + }); + const settled = await result("durable-order"); + expect(settled).toMatchObject({ status: "completed", stdout: "fast\n", value: "fast" }); + expect(await calls()).toEqual({ slow: 1, fast: 1, gate: 2 }); + }); + + it("fires replayed timers in order with replayed responses", async () => { + await start("durable-race", { + source: ` + import { gate, wait } from "ws:test-host"; + // unref() checks the timers hand back the same kind of handle as + // the runtime's own. + const sleep = (ms, value) => + new Promise((resolve) => setTimeout(() => resolve(value), ms).unref()); + export default async function () { + const long = wait(150, "long"); + const first = await Promise.race([sleep(20, "timer"), long.then(() => "call")]); + await long; + const short = wait(5, "short"); + const second = await Promise.race([sleep(150, "timer"), short.then(() => "call")]); + console.log(first, second); + await gate(); + return [first, second]; + } + `, + onInterrupt: "replay", + }); + const settled = await result("durable-race"); + expect(settled).toMatchObject({ status: "completed", stdout: "timer call\n" }); + expect(await calls()).toEqual({ long: 1, short: 1, gate: 2 }); + }); + + it("keeps the live order when the run computes between events", async () => { + await start("durable-compute", { + source: ` + import { gate, wait } from "ws:test-host"; + const sleep = (ms, value) => new Promise((resolve) => setTimeout(resolve, ms)); + export default async function () { + // The clock doesn't move while code computes, so the timer is + // due earlier by the clock than it fires in fact. + let x = 0; + for (let i = 0; i < 1e9; i++) x ^= i; + const call = wait(100, "call").then(() => "call"); + const winner = await Promise.race([sleep(300, "timer"), call]); + await call; + console.log(winner, x > -1); + await gate(); + return winner; + } + `, + onInterrupt: "replay", + }); + const settled = await result("durable-compute"); + expect(settled).toMatchObject({ status: "completed", stdout: "call true\n", value: "call" }); + }); + + it("journals fetch so a replay doesn't send it again", async () => { + await start("durable-fetch", { + source: ` + import { gate } from "ws:test-host"; + export default async function () { + const response = await fetch("https://gateway.test/items", { method: "POST", body: "hi" }); + const text = await response.text(); + console.log(response.status, response.headers.get("x-count"), text); + await gate(); + return text; + } + `, + onInterrupt: "replay", + network: true, + }); + const settled = await result("durable-fetch", { network: true }); + expect(settled).toMatchObject({ + status: "completed", + stdout: "200 1 POST /items hi #1\n", + value: "POST /items hi #1", + }); + expect(await calls()).toEqual({ fetch: 1, gate: 2 }); + }); + + it("doesn't count a replayable run's fetches toward its capability calls", async () => { + await start("durable-many-fetches", { + source: ` + import { gate } from "ws:test-host"; + export default async function () { + // More than the default 256 capability calls, a dozen at a time. + for (let round = 0; round < 25; round++) { + await Promise.all(Array.from({ length: 12 }, () => fetch("https://gateway.test/x"))); + } + await gate(); + return "done"; + } + `, + onInterrupt: "replay", + network: true, + }); + const settled = await result("durable-many-fetches", { network: true }); + expect(settled).toMatchObject({ status: "completed", value: "done" }); + expect(await calls()).toEqual({ fetch: 300, gate: 2 }); + }); + + it("fails a replay that prints different output", async () => { + await start("durable-output", { + source: ` + import { gate } from "ws:test-host"; + export default async function () { + const key = await crypto.subtle.generateKey({ name: "HMAC", hash: "SHA-256" }, true, ["sign"]); + console.log(new Uint8Array(await crypto.subtle.exportKey("raw", key)).slice(0, 8).join(",")); + await gate(); + } + `, + onInterrupt: "replay", + // This backend's output limit leaves the whole message. + network: true, + }); + const settled = await result("durable-output", { network: true }); + expect(settled).toMatchObject({ status: "failed", exitCode: 1 }); + expect(settled.stderr).toContain("it wrote different stdout from the first time"); + }); + + it("fails a replay interrupted during a call that isn't safe to repeat", async () => { + await start("durable-unsafe", { + source: ` + import { unsafeGate } from "ws:test-host"; + export default async function () { + await unsafeGate(); + } + `, + onInterrupt: "replay", + network: true, + }); + const settled = await result("durable-unsafe", { network: true }); + expect(settled).toMatchObject({ status: "failed", exitCode: 1 }); + expect(settled.stderr).toContain( + "interrupted while ws:test-host.unsafeGate was running, which isn't safe to repeat", + ); + expect(await calls()).toEqual({ gate: 1 }); + }); + + it("replays a run interrupted a second time during its replay", async () => { + await start("durable-twice", { + source: ` + import { gate, hold, tick } from "ws:test-host"; + export default async function () { + console.log("one", await tick("a")); + await gate(); + console.log("two", await tick("b")); + await hold("second"); + console.log("three", await tick("c")); + return Math.random(); + } + `, + onInterrupt: "replay", + }); + // Resume the run, and wait for it to reach the second hold. + await SELF.fetch("https://example.test/durable-recover", { method: "POST" }); + for (let attempt = 0; attempt < 100 && (await calls()).second !== 1; attempt += 1) { + await new Promise((resolve) => setTimeout(resolve, 20)); + } + await SELF.fetch("https://example.test/durable-restart?open=second", { method: "POST" }); + const settled = await result("durable-twice"); + expect(settled).toMatchObject({ status: "completed", stdout: "one 1\ntwo 1\nthree 1\n" }); + // Each tick ran once. The gate and the hold were each in flight at an + // interruption, so each ran again once. + expect(await calls()).toEqual({ a: 1, gate: 2, b: 1, second: 2, c: 1 }); + }); + + it("fails an interrupted run by default", async () => { + await start("durable-fail", { source }); + const settled = await result("durable-fail"); + expect(settled).toMatchObject({ status: "failed", exitCode: 1 }); + expect(settled.stderr).toContain("interrupted when its Workspace runtime restarted"); + }); + + it("resumes runs when the host calls recover", async () => { + await start("durable-recover", { source, onInterrupt: "replay" }); + await SELF.fetch("https://example.test/durable-recover", { method: "POST" }); + for (let attempt = 0; attempt < 100 && (await calls()).work !== 2; attempt += 1) { + await new Promise((resolve) => setTimeout(resolve, 20)); + } + expect(await calls()).toEqual({ work: 2, gate: 2 }); + expect(await result("durable-recover")).toMatchObject({ status: "completed" }); + }); +});