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.

docs/17_isolate_javascript.md

@@ -143,7 +143,66 @@ 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. See [Interrupted executions](#interrupted-executions) for what happens next.
+
+## Interrupted executions
+
+A deploy or a crash restarts the Durable Object and ends every run in flight. `onInterrupt` decides what happens to those runs. Set a default on the backend and override it per run:
+
+```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. Calls it already made run again, and its output repeats after a note on standard error. |
+| `"replay"` | The run starts again from the beginning, but every call it already made is answered from a journal instead of running again, and its output isn't repeated. Once it catches up, it carries on live. |
+
+The code being run doesn't need to change for either. `"restart"` and `"replay"` are [experimental](../packages/computer/README.md#status): their behavior and stored data may change in any release, and a run interrupted across such an upgrade fails instead of resuming.
+
+### When a run resumes
+
+Interrupted runs resume when the backend next connects, which the next runtime call such as `exec()` or `getExec()` triggers, as does `workspace.ready({ all: true })`. A caller polling a run or reconnecting to its output does this on its own. 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. To resume runs without waiting for traffic, call `workspace.runtime.recover()` from the Durable Object's alarm. The Workspace doesn't set an alarm itself, since a Durable Object only has one.
+
+### How replay works
+
+A replayable run saves its code, input and limits when it starts. Every filesystem call, host module call and `fetch` it makes is written to a journal in SQLite before the run sees the response. On replay, the run starts from the top and gets those responses back from the journal, in the order it first saw them, so a `Promise.race` or a timer racing a call goes the same way.
+
+A replayable run also gets versions of a few APIs that would otherwise differ between attempts:
+
+- `Date.now()`, `new Date()` and `performance.now()` read a clock that gives the same times on replay.
+- `setTimeout`, `setInterval`, `scheduler.wait()` and `AbortSignal.timeout()` fire in the same order on replay, without waiting out delays they already waited.
+- `Math.random()`, `crypto.getRandomValues()` and `crypto.randomUUID()` return the same values, from a random seed saved with the run.
+- `fetch` goes through the host, under the backend's egress policy, so a replay doesn't send a request again.
+
+A replay has to take the same path as the first attempt: the same calls, with the same arguments, printing the same output. If it doesn't, the run fails with a message saying where it went differently. Everything saved for a run is deleted when it settles.
+
+### Calls that mustn't run twice
+
+A call still running when the object restarted never returned, so the replay runs it again. Each host module call gets a `context.callId` that stays the same when that happens, so a function can pass it on as an idempotency key. A function that can't, such as one that charges a card, can be wrapped in `unsafeToRepeat()`, and an interruption during it 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 }),
+  },
+}
+```
+
+### Limits
+
+- A run gets three attempts, counted again each time it journals a call, and has to finish within its original timeout, counted from when it first started.
+- A run only resumes on the same runtime. A new compatibility date or flags, or a new `maxStdioBytes` or `maxCapabilityBytes`, fails the runs it interrupted, and so does an upgrade of this package that changes how runs replay.
+- Interrupted runs restart four at a time, since a Durable Object can only run ten Dynamic Workers at once.
+- A replayable run's `fetch` is buffered rather than streamed, and each body has to fit in `maxCapabilityBytes` (1 MiB by default). Requests don't count toward `maxCapabilityCalls`, and at most six are open at once, with the rest waiting. A request in flight at the restart is sent again. Aborting a request stops the code waiting for it, but the host's request runs until it finishes.
+- Keys from `crypto.subtle.generateKey()`, `node:crypto` randomness, `node:timers` and `cloudflare:sockets` aren't replayed. A run that prints their results or passes them to a call fails on replay.
+- The journal uses Durable Object storage until the run settles: roughly the size of every response, plus the run's code.
 
 ## 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,287 @@
+import { getEventListeners } from "node:events";
+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("detaches abort listeners from waits and requests that finish", async () => {
+    const call = vi.fn<(...args: unknown[]) => Promise<unknown>>(async () => ({
+      status: 200,
+      statusText: "OK",
+      headers: [],
+      body: null,
+    }));
+    const { run } = load([], undefined, call);
+    const signals = run<Record<string, AbortSignal>>(`
+      const signals = {
+        wait: new AbortController().signal,
+        fetch: new AbortController().signal,
+        request: new AbortController().signal,
+      };
+      for (let wait = 0; wait < 3; wait += 1) scheduler.wait(0, { signal: signals.wait });
+      fetch("https://example.test/a", { signal: signals.fetch });
+      // The Request a fetch builds follows its signal, as it would anyway.
+      new Request("https://example.test/a", { signal: signals.request });
+      signals;
+    `);
+    await settle();
+    expect(getEventListeners(signals.wait, "abort")).toHaveLength(0);
+    expect(getEventListeners(signals.fetch, "abort")).toHaveLength(
+      getEventListeners(signals.request, "abort").length,
+    );
+  });
+
+  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);
+  });
+});