computer: Restart or replay interrupted isolate JavaScript runs

.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.

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
 

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,
   };
 }

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 |

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<unknown>,
+  ): void;
+  request(): void;
+  failed(): void;
+  deliver(payload: Payload): Promise<Payload> | 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<unknown> = 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 = <T>(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<number[]>("[...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<string>("[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<unknown>>(async () => ({
+      status: 302,
+      statusText: "Found",
+      headers: [],
+      body: null,
+    }));
+    const { run } = load([], undefined, call);
+    const status = await run<Promise<number>>(
+      "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);
+  });
+});