diff --git a/.changeset/retry-computerd-restarts.md b/.changeset/retry-computerd-restarts.md new file mode 100644 index 00000000..b6df223a --- /dev/null +++ b/.changeset/retry-computerd-restarts.md @@ -0,0 +1,5 @@ +--- +"@cloudflare/computer": patch +--- + +`container-shell` operations now reconnect after computerd restarts when retrying is safe, and process-local execution handles return `EEXEC_LOST` after container replacement. See [container connection recovery](https://github.com/cloudflare/computer/blob/main/docs/05_runtime_interface.md#command-synchronization). diff --git a/docs/02_sync_protocol.md b/docs/02_sync_protocol.md index c75bc699..bfc6f4ca 100644 --- a/docs/02_sync_protocol.md +++ b/docs/02_sync_protocol.md @@ -267,11 +267,8 @@ edited file) shows up exactly once on the wire. See ## Failure handling -- **Container restart mid-exec.** The DO's connection detects the - closed WebSocket and self-destructs. The next call transparently - rebuilds against the still-running `computerd` (or restarts it if needed). - `pushRev` and the fetch cursor mean the catch-up is incremental, modulo - whatever the container's deployment chose for its DB lifetime. +- **Container restart or transport loss.** The Durable Object detects the closed WebSocket, removes and closes the stale backend handle, and reconnects through the normal computerd health gate. `pushOnce` and `pullOnce` get one reconnect retry inside the original logical operation. `pushRev` and the fetch cursor make those retries idempotent: a torn push replays entries whose watermark did not advance, while a torn pull resumes after its last committed batch. A fresh process-lifetime computerd database resets the matching local watermarks before the replacement handle is exposed, so the next push rebuilds it from a rev-0 baseline. +- **Container restart during command execution.** The pre-exec push must succeed, including its reconnect retry, before `shell.exec` is called. A locally disposed stub proves that a spawn request was never sent and permits one reconnect retry. A generic disconnect after dispatch is ambiguous, so the backend invalidates the handle and reports that the command may have started instead of replaying it. Event-stream failures follow the same no-replay rule. The post-command pull remains safe to retry only against the runtime UUID that ran the command. If reconnect reaches a replacement container, the execution result reports pending sync rather than treating its empty VFS as a successful zero-entry pull. The durable retry intent retains the original runtime UUID. When a scheduled retry confirms that runtime is gone, it reports the sync as lost and clears the unrecoverable intent so later commands on the live runtime can schedule their own pending pulls. - **Container crash mid-apply.** `push` is atomic from the DO's perspective on the receiver: the server wraps the whole batch in a single `db.transactionSync` via the synchronous `applyChangesSync` diff --git a/docs/05_runtime_interface.md b/docs/05_runtime_interface.md index 64a57485..39c02144 100644 --- a/docs/05_runtime_interface.md +++ b/docs/05_runtime_interface.md @@ -98,11 +98,15 @@ push → spawn → events/result → pull A backend with `sync: "none"`, such as `worker-shell`, shares the host store and reports zero push/pull counts. A Container has its own VFS and synchronizes changes before and after command execution. Fully draining either `result()` or the event stream completes the post-command pull before the stream closes. +The pre-command push is a safety gate, not a best-effort optimization. The push and spawn use the same backend handle. If that handle fails before dispatch, the reconnect retry repeats both steps on the replacement container so the command cannot skip its push. If the push still fails, `exec()` rejects before the spawn request is sent; `pushed: 0` means a successful push found no entries, not that synchronization failed. A failed post-command pull does not change the completed command result. The pull is fenced to the runtime UUID that ran the command, so reconnecting to an empty replacement cannot report a clean zero-entry sync. It reports `sync.status: "pending"` and persists that UUID with the retry intent. `retryPendingSync()` resumes against the original runtime when possible; if that runtime was replaced, it clears the unrecoverable intent and reports `status: "lost"` so a later command can schedule sync for the live runtime. + +Container connection failures also get one backend-internal reconnect attempt. Sync calls are safe to repeat. `getExec`, `killExec`, and `disposeExec` are retried only when the new connection reaches the same computerd runtime; a replacement container returns `EEXEC_LOST` instead of applying an old execution id to its new process table. `shell.exec` is different: the backend retries it only when connection setup failed or a locally disposed stub proves that no request was sent. If the transport fails after computerd may have accepted the spawn, the error states that the command may have started and the backend does not replay it. A failure while reading the event stream also invalidates the connection without rerunning the command. + Module backends use host capability calls against the authoritative Workspace and therefore require no push/pull round trip. ## Lifecycle differences -`container-shell` provides computerd's retained process log, replay, signals, and disposal. +`container-shell` provides computerd's retained process log, replay, signals, and disposal. Execution handles are scoped to the UUID of the container process that accepted them, so they cannot target a reused execution id after process replacement. The latest UUID owner for each execution ID is stored in Workspace SQLite, with a bounded in-memory LRU cache, so direct by-ID operations preserve the fence across Durable Object incarnation and cache eviction. `worker-javascript` provides a Workspace-owned execution journal, retained result/events, host cancellation, and explicit disposal. Active Workers cannot be serialized across host restart; orphaned running records are reconciled to failed. diff --git a/docs/07_injected_service.md b/docs/07_injected_service.md index 24b3548d..c323f754 100644 --- a/docs/07_injected_service.md +++ b/docs/07_injected_service.md @@ -144,12 +144,8 @@ Sharp edges actually present in `cloudflare-container.ts`: - `#armUpgrade` must be set up *before* `#postConnect`, because `computerd` can dial back before the `POST /connect` response returns. -- A `#monitoring` flag watches container exit and drops the cached - handle so the next call rebuilds from scratch. -- **No transparent reconnect after a mid-session drop.** If the - WebSocket dies, the caller is expected to reconstruct the - `Workspace` rather than the backend trying to splice a new socket - into the existing session. +- The container host records each monitored generation's exit reason. The dead container closes its WebSocket, and `fetchPort()` also short-circuits later requests with a transport error; either path invalidates the matching Workspace handle. +- **Reconnect replaces the whole session.** If the WebSocket dies, `Workspace` invalidates and closes the matching backend handle, then calls `CloudflareContainerBackend.connect()` again. The replacement runs the complete start, egress-interception, health, `/connect`, and reverse-WebSocket sequence; the backend never splices a new carrier into the dead capnweb session. Replay-safe sync and process lifecycle operations get one retry. Command spawn is retried only when no request was dispatched. ## Environment variables @@ -196,17 +192,9 @@ Today: ## Lifetime -The `computerd` process is long-lived and outlives DO restarts — the -sandbox container is reaped only when its lifetime policy says so, -and a fresh DO incarnation reconnects to the same running daemon over -a new WebSocket (the Cloudflare backend's `#monitoring` flag drops the -cached handle if the container itself exits, forcing a rebuild). - -Caveat: **no on-disk persistence yet** (`packages/computerd/README.md`). -The "same in-memory VFS across DO restarts" picture only holds while -the container process is alive. A container restart loses VFS state; -sync via `UPSTREAM_URL` is what brings state back across container -restarts. +The `computerd` process is long-lived and outlives Durable Object restarts — the sandbox container is reaped only when its lifetime policy says so, and a fresh Durable Object incarnation reconnects to the same running daemon over a new WebSocket. The container monitor and transport error classifier drop stale handles so an operation can reconnect through the readiness gate. + +Caveat: **no on-disk persistence yet** (`packages/computerd/README.md`). The same in-memory VFS across Durable Object restarts only holds while the container process is alive. A container restart loses VFS state. Watermark reconciliation and the next push rebuild the mirror from Durable Object storage, but reconnect cannot recover container-local files that were never pulled before the process died. ## Open questions diff --git a/docs/11_lifecycle.md b/docs/11_lifecycle.md index 6bf7952b..4166a907 100644 --- a/docs/11_lifecycle.md +++ b/docs/11_lifecycle.md @@ -84,6 +84,9 @@ an incarnation boundary. What survives is: container-side cursor the DO has fetched). These are written via the same SQLite transaction as the data they describe, so they cannot drift out of sync with the store. +- The latest container runtime UUID for each execution ID. This lets a + reconstructed Workspace reject stale get, kill, or dispose calls before + they can reach a replacement container that reused the same ID. On every new incarnation `Workspace.ready()` re-runs `#connect()`, which re-enters the backend's bootstrap sequence. If the container is @@ -173,14 +176,11 @@ package WebSocket and require the `computerd` process to be live. 3. **Death.** The WebSocket closes (clean or RST). capnweb errors every pending answer. The session is unrecoverable. -The death case today is **not handled** — the `Workspace` keeps its -`#handle` reference pointing at the dead session, and the next RPC -call throws. The caller is expected to reconstruct the workspace. +Session death is handled at the `Workspace` backend boundary. A close event, capnweb `onRpcBroken` callback, container exit, or classified transport error removes and closes the matching handle. The original operation gets one reconnect retry when replay is safe; the replacement handle is not exposed until computerd passes its health check and `reconcileWatermarks()` has compared the two stores. ### What an in-flight RPC looks like across a transport failure -Because the rev counters drive every operation, a torn RPC is safe -to retry against a fresh session. Specifically: +Because the revision counters drive every sync operation, a torn sync RPC is safe to retry against a fresh session. `Workspace` performs one such reconnect retry automatically. Specifically: - **`pushOnce`.** `pushRev` is written only after `assertAppliedPushCursor` succeeds. A torn push leaves `pushRev` at @@ -192,12 +192,11 @@ to retry against a fresh session. Specifically: past that point, including within the same rev. `applyChanges`'s `alreadyApplied` check drops any duplicates the resume happens to overlap with. -- **`exec.events`.** Each event carries a monotonic `seq` per exec - id. The client reattaches via `getExec({ id, after: seq })`. +- **`exec` dispatch.** A failed connection setup or a local disposed-stub error happens before dispatch and can be retried once. Other transport failures are ambiguous: computerd may have accepted the command before the response was lost. The backend invalidates the handle and reports the failure without replaying the command. +- **`exec.events`.** Each event carries a monotonic `seq` per exec ID and callers can reattach with `getExec({ id, after: seq })`. The current automatic recovery boundary does not reattach a torn event stream; it reports the stream failure and leaves the next explicit operation to reconnect. +- **`getExec`, `killExec`, and `disposeExec`.** These ID-addressed operations get one reconnect retry when the connection still points at the same container runtime UUID. A replacement process has an empty execution registry, so a runtime mismatch returns `EEXEC_LOST` without sending the old execution ID to the replacement. -This is why the sync protocol survives transport failures: every -operation has a persistent cursor, and every receiver is idempotent. -capnweb itself is fragile, but the protocol layered on top isn't. +This is why the sync protocol survives transport failures: every sync operation has a persistent cursor, and every receiver is idempotent. Shell commands require the separate no-replay boundary above because their side effects are not generally idempotent. capnweb itself is fragile, but the protocols layered on top define where recovery is safe. ### Stub disposal contract @@ -394,9 +393,9 @@ items not yet shipped. | DO restart, container alive | New incarnation | Unchanged | Fresh session over fresh socket | | DO hibernate (future) | Isolate evicted, socket survives | Unchanged | *Fresh tables on wake; sync resumes from `_vfs_watermark`, exec resumes from `serializeAttachment` seqs* | | DO OOM | Killed, new incarnation on next event | Unchanged (until backend rebuilds) | Dies, fresh session on next call | -| Container SIGTERM | Unchanged until next call | Restarted; in-memory VFS lost | Dies on container exit; *watermark reconcile on next connect repairs the mismatch* | -| Container OOM/kill | Unchanged until next call | Killed; restarted on next call | Same as SIGTERM | -| WebSocket idle disconnect | Unchanged | Unchanged | Dies on `close`; *reconnect wrapper rebuilds* | +| Container SIGTERM | Invalidates the handle and reconnects on the active or next operation | Restarted; in-memory VFS lost | Dies on container exit; watermark reconciliation rebuilds from Durable Object storage | +| Container OOM/kill | Invalidates the handle and reconnects on the active or next operation | Killed; restarted on reconnect | Same as SIGTERM; container-only unsynced data is lost | +| WebSocket idle disconnect | Invalidates and closes the handle | Unchanged | Dies on `close`; replay-safe operations reconnect once | | Both die (host failure) | New incarnation on next event | New container | Rev-0 baseline from DO store | The recurring theme: **DO storage is the only durable thing in this diff --git a/packages/computer/README.md b/packages/computer/README.md index b0d5fa84..b980d8a7 100644 --- a/packages/computer/README.md +++ b/packages/computer/README.md @@ -476,9 +476,10 @@ fail; the result exposes `sync: { status: "pending", ... }`. Configure a `SyncRetryScheduler` on `Workspace` to persist one coalesced retry per backend, then call `workspace.retryPendingSync(backend)` from your DO's alarm. Retries use bounded exponential backoff and return `"exhausted"` -after the configured maximum. The library does not own your DO's alarm. -See `SyncRetryScheduler`, `SyncRetryIntent`, and `SyncRetryOptions` in -the package exports. +after the configured maximum. A container replacement returns `"lost"` +and clears the unrecoverable intent so new work is not blocked. The library +does not own your DO's alarm. See `SyncRetryScheduler`, `SyncRetryIntent`, +and `SyncRetryOptions` in the package exports. ### Observability diff --git a/packages/computer/src/backend.ts b/packages/computer/src/backend.ts index 87482600..dac310eb 100644 --- a/packages/computer/src/backend.ts +++ b/packages/computer/src/backend.ts @@ -73,6 +73,11 @@ export interface BackendHandle { // The composite WorkspaceRPC stub pointing at the computerd // backend produced. rpc: WorkspaceRPC; + // Durable identity of the runtime process behind this connection. + // Reconnecting to the same process preserves it; a replacement + // process receives a new id. Backends without process-local state + // may omit it. + runtimeId?: string; // Declares whether this backend pairs with an independent // remote store that the Workspace must sync against. // diff --git a/packages/computer/src/backends/container/cloudflare-container.test.ts b/packages/computer/src/backends/container/cloudflare-container.test.ts index b454e106..09989a54 100644 --- a/packages/computer/src/backends/container/cloudflare-container.test.ts +++ b/packages/computer/src/backends/container/cloudflare-container.test.ts @@ -11,6 +11,7 @@ import { describe, expect, test, vi } from "vitest"; +import { WorkspaceTransportError } from "../../transport-failure.js"; import { CloudflareContainerBackend } from "./cloudflare-container.js"; import type { IWorkspaceContainerAPI, WorkspaceRef } from "./container-host.js"; @@ -22,6 +23,8 @@ interface FakeHostOptions { // don't care about transitions. healthSequence?: boolean[]; connectStatus?: number; + start?: () => Promise; + intercept?: () => Promise; restart?: () => Promise; // Pre-set a prior exit reason so connect()'s pre-flight // exitInfo() check observes it. @@ -39,6 +42,7 @@ interface FakeHost { gatewayToken?: string; running: boolean; exit: { exitedAt: number; reason: string } | null; + runtimeId: string | null; simulateExit(reason: string): void; } @@ -51,6 +55,7 @@ function makeFakeHost(opts: FakeHostOptions = {}): FakeHost { calls, running: false, exit: opts.priorExit ?? null, + runtimeId: null, simulateExit(reason: string) { state.exit = { exitedAt: Date.now(), reason }; state.running = false; @@ -69,13 +74,17 @@ function makeFakeHost(opts: FakeHostOptions = {}): FakeHost { calls.push({ name: "start", args: [env, enableInternet] }); state.startEnv = env; state.enableInternet = enableInternet; + await opts.start?.(); + if (!state.running) state.runtimeId = crypto.randomUUID(); state.running = true; // A successful start clears any prior exit, matching // WorkspaceContainerAPI.start. state.exit = null; + return { runtimeId: state.runtimeId ?? "missing-runtime" }; }, async interceptOutboundHttp(host, ref) { calls.push({ name: "interceptOutboundHttp", args: [host, ref] }); + await opts.intercept?.(); state.interceptedHost = host; state.interceptedWorkspace = ref; }, @@ -110,6 +119,8 @@ function makeFakeHost(opts: FakeHostOptions = {}): FakeHost { } state.running = true; state.exit = null; + state.runtimeId = crypto.randomUUID(); + return { runtimeId: state.runtimeId }; }, async status() { calls.push({ name: "status", args: [] }); @@ -126,6 +137,46 @@ function makeFakeHost(opts: FakeHostOptions = {}): FakeHost { const fakeWorkspace: WorkspaceRef = { binding: "TestDO", id: "abc123" }; describe("CloudflareContainerBackend", () => { + test("connect() classifies a container start failure as transport", async () => { + const platformError = new Error( + "There is no container instance that can be provided to this Durable Object, try again later", + ); + const fake = makeFakeHost({ + start: async () => { + throw platformError; + }, + }); + const backend = new CloudflareContainerBackend({ + container: () => ({ getWorkspaceContainer: () => fake.host }), + workspace: fakeWorkspace, + connectTimeoutMs: 300, + }); + + const error = await backend.connect().catch((caught: unknown) => caught); + expect(error).toBeInstanceOf(WorkspaceTransportError); + expect(error).toMatchObject({ cause: platformError }); + expect(String(error)).toMatch(/stage=start/); + }); + + test("connect() classifies egress interception failure as transport", async () => { + const platformError = new Error("container is not ready for egress interception"); + const fake = makeFakeHost({ + intercept: async () => { + throw platformError; + }, + }); + const backend = new CloudflareContainerBackend({ + container: () => ({ getWorkspaceContainer: () => fake.host }), + workspace: fakeWorkspace, + connectTimeoutMs: 300, + }); + + const error = await backend.connect().catch((caught: unknown) => caught); + expect(error).toBeInstanceOf(WorkspaceTransportError); + expect(error).toMatchObject({ cause: platformError }); + expect(String(error)).toMatch(/stage=egress/); + }); + test("connect() throws when the container port never opens", async () => { const fake = makeFakeHost({ healthy: false }); const backend = new CloudflareContainerBackend({ @@ -135,7 +186,10 @@ describe("CloudflareContainerBackend", () => { restartAttempts: 0, }); - await expect(backend.connect()).rejects.toThrow(/stage=health.*port=8080/); + const error = await backend.connect().catch((caught: unknown) => caught); + expect(error).toBeInstanceOf(WorkspaceTransportError); + expect(error).toMatchObject({ cause: expect.any(Error) }); + expect(String(error)).toMatch(/stage=health.*port=8080/); const names = fake.calls.map((c) => c.name); expect(names).toContain("start"); @@ -321,10 +375,12 @@ describe("CloudflareContainerBackend", () => { connectTimeoutMs: 600, }); - await expect(backend.connect()).rejects.toThrow(/POST \/connect returned 502/); + const error = await backend.connect().catch((caught: unknown) => caught); + expect(error).toBeInstanceOf(WorkspaceTransportError); + expect(String(error)).toMatch(/POST \/connect returned 502/); }); - test("connect() throws when the /ws upgrade never arrives", async () => { + test("connect() throws a transport error when the /ws upgrade never arrives", async () => { const fake = makeFakeHost(); const backend = new CloudflareContainerBackend({ container: () => ({ getWorkspaceContainer: () => fake.host }), @@ -332,7 +388,9 @@ describe("CloudflareContainerBackend", () => { connectTimeoutMs: 600, }); - await expect(backend.connect()).rejects.toThrow(/\/ws upgrade did not arrive/); + const error = await backend.connect().catch((caught: unknown) => caught); + expect(error).toBeInstanceOf(WorkspaceTransportError); + expect(String(error)).toMatch(/\/ws upgrade did not arrive/); }); test("handleFetch rejects non-/ws paths", async () => { diff --git a/packages/computer/src/backends/container/cloudflare-container.ts b/packages/computer/src/backends/container/cloudflare-container.ts index 2511d050..5ecd25a8 100644 --- a/packages/computer/src/backends/container/cloudflare-container.ts +++ b/packages/computer/src/backends/container/cloudflare-container.ts @@ -54,6 +54,7 @@ import { WORKSPACE_EGRESS_URL_HEADER, type WorkspaceEgressPolicy, } from "../../runtime/egress.js"; +import { WorkspaceTransportError } from "../../transport-failure.js"; import type { IWorkspaceContainerAPI, WorkspaceRef } from "./container-host.js"; import { probeComputerdHealth } from "./health-probe.js"; @@ -209,10 +210,37 @@ export class CloudflareContainerBackend implements WorkspaceBackend { MOUNT_POINT: "/workspace", ...this.#options.containerEnv, }; - await host.start(env, this.#egress.mode === "direct"); - await host.interceptOutboundHttp(this.#options.egressHost, this.#options.workspace); - if (this.#egress.mode === "http-gateway" && this.#egressToken !== undefined) { - await host.interceptAllOutboundHttp(this.#options.workspace, this.#egressToken); + let runtimeId: string; + try { + ({ runtimeId } = await host.start(env, this.#egress.mode === "direct")); + } catch (error) { + throw new WorkspaceTransportError( + this.#formatStageError("start", { + attempt: 1, + maxAttempts: this.#options.restartAttempts + 1, + restarts: 0, + lastError: error, + priorExit, + }), + { cause: error }, + ); + } + try { + await host.interceptOutboundHttp(this.#options.egressHost, this.#options.workspace); + if (this.#egress.mode === "http-gateway" && this.#egressToken !== undefined) { + await host.interceptAllOutboundHttp(this.#options.workspace, this.#egressToken); + } + } catch (error) { + throw new WorkspaceTransportError( + this.#formatStageError("egress", { + attempt: 1, + maxAttempts: this.#options.restartAttempts + 1, + restarts: 0, + lastError: error, + priorExit, + }), + { cause: error }, + ); } // Arm the upgrade promise before posting /connect — computerd @@ -220,7 +248,7 @@ export class CloudflareContainerBackend implements WorkspaceBackend { // the upgrade can arrive before the POST resolves. this.#armUpgrade(); - await this.#readyWithRestarts(host, env, deadline, priorExit); + runtimeId = await this.#readyWithRestarts(host, env, deadline, priorExit, runtimeId); await this.#postConnect(host, deadline); const ws = await this.#waitForUpgrade(deadline); @@ -273,6 +301,7 @@ export class CloudflareContainerBackend implements WorkspaceBackend { const handle: BackendHandle = { rpc: stub as unknown as WorkspaceRPC, + runtimeId, closed, close: async () => { stopHeartbeat?.(); @@ -377,7 +406,8 @@ export class CloudflareContainerBackend implements WorkspaceBackend { env: Record, deadline: number, priorExit: { exitedAt: number; reason: string } | null, - ): Promise { + initialRuntimeId: string, + ): Promise { const maxAttempts = this.#options.restartAttempts + 1; // Split the remaining time across attempts so a failing // first attempt doesn't starve the restart-retry. Floor at @@ -388,6 +418,7 @@ export class CloudflareContainerBackend implements WorkspaceBackend { let attempt = 0; let restarts = 0; let lastError: unknown; + let runtimeId = initialRuntimeId; while (attempt < maxAttempts) { attempt++; @@ -399,16 +430,16 @@ export class CloudflareContainerBackend implements WorkspaceBackend { return false; }, ); - if (ok) return; + if (ok) return runtimeId; if (attempt < maxAttempts) { try { - await host.restart(env, this.#egress.mode === "direct"); + ({ runtimeId } = await host.restart(env, this.#egress.mode === "direct")); restarts++; } catch (error) { this.#rejectUpgrade?.(error); this.#clearUpgrade(); - throw new Error( + throw new WorkspaceTransportError( this.#formatStageError("restart", { attempt, maxAttempts, @@ -423,7 +454,7 @@ export class CloudflareContainerBackend implements WorkspaceBackend { this.#rejectUpgrade?.(new Error("computerd never became healthy")); this.#clearUpgrade(); - throw new Error( + throw new WorkspaceTransportError( this.#formatStageError("health", { attempt, maxAttempts, @@ -461,7 +492,7 @@ export class CloudflareContainerBackend implements WorkspaceBackend { } #formatStageError( - stage: "start" | "health" | "restart" | "connect" | "ws", + stage: "start" | "egress" | "health" | "restart" | "connect" | "ws", info: { attempt: number; maxAttempts: number; @@ -495,17 +526,19 @@ export class CloudflareContainerBackend implements WorkspaceBackend { } catch (error) { this.#rejectUpgrade?.(error); this.#clearUpgrade(); - throw new Error( + throw new WorkspaceTransportError( `CloudflareContainerBackend(${this.id}) [stage=connect]: POST /connect failed: ${describeError(error)}`, { cause: error }, ); } if (!res.ok) { const body = await res.text().catch(() => ""); - this.#rejectUpgrade?.(new Error(`/connect ${res.status}`)); + const cause = new Error(`/connect ${res.status}`); + this.#rejectUpgrade?.(cause); this.#clearUpgrade(); - throw new Error( + throw new WorkspaceTransportError( `CloudflareContainerBackend(${this.id}) [stage=connect]: POST /connect returned ${res.status}: ${body}`, + { cause }, ); } } @@ -523,7 +556,7 @@ export class CloudflareContainerBackend implements WorkspaceBackend { timer = setTimeout( () => reject( - new Error( + new WorkspaceTransportError( `CloudflareContainerBackend(${this.id}) [stage=ws]: /ws upgrade did not arrive within ${this.#options.connectTimeoutMs}ms`, ), ), diff --git a/packages/computer/src/backends/container/container-host.ts b/packages/computer/src/backends/container/container-host.ts index 0cc2f519..1ba1a8f3 100644 --- a/packages/computer/src/backends/container/container-host.ts +++ b/packages/computer/src/backends/container/container-host.ts @@ -26,6 +26,10 @@ import { destroyContainerExpectingExit, installContainerMonitor, } from "./container-lifecycle.js"; +import { + type ContainerRuntimeIdentity, + CurrentContainerRuntimeIdentity, +} from "./container-runtime-identity.js"; export type { ContainerExitInfo } from "./container-lifecycle.js"; @@ -42,11 +46,15 @@ export interface WorkspaceRef { // Driver surface CloudflareContainerBackend talks to. Implemented // by WorkspaceContainerAPI below; exposed on consumer DOs through // the `ws` accessor that withWorkspaceContainer installs. +export interface ContainerRuntimeInfo { + runtimeId: string; +} + export interface IWorkspaceContainerAPI { - // Idempotent start. Returns once the runtime has accepted the - // start command; readiness is verified by the backend through - // probeComputerdHealth against port(). - start(env: Record, enableInternet: boolean): Promise; + // Idempotent start. Returns the durable identity of the running + // container process once the runtime has accepted the start command; + // readiness is verified by the backend through probeComputerdHealth. + start(env: Record, enableInternet: boolean): Promise; // Wire `host` → workspace inside the container's egress table. // Called once per backend connect(). The implementation @@ -69,7 +77,7 @@ export interface IWorkspaceContainerAPI { // current generation dead. Implementation: destroy() the // container, then start({ env }). Callers bound the number of // restart attempts — this method does no looping of its own. - restart(env: Record, enableInternet: boolean): Promise; + restart(env: Record, enableInternet: boolean): Promise; // Coarse diagnostic state. The `running` flag reports whether // the platform still has a container instance attached; it does @@ -95,6 +103,7 @@ export interface IWorkspaceContainerAPI { export class WorkspaceContainerAPI extends RpcTarget implements IWorkspaceContainerAPI { readonly #container: NonNullable; readonly #ctx: DurableObjectState; + readonly #runtimeIdentity: CurrentContainerRuntimeIdentity; constructor(ctx: DurableObjectState) { super(); @@ -103,6 +112,7 @@ export class WorkspaceContainerAPI extends RpcTarget implements IWorkspaceContai } this.#container = ctx.container; this.#ctx = ctx; + this.#runtimeIdentity = new CurrentContainerRuntimeIdentity(ctx.storage); } async start(env: Record, enableInternet: boolean) { @@ -113,6 +123,7 @@ export class WorkspaceContainerAPI extends RpcTarget implements IWorkspaceContai // destroy resolves, so guarding the start against it would let // a stale-running flag skip the re-launch entirely. const priorExit = containerExitInfo(this.#ctx); + let runtime: ContainerRuntimeIdentity; if (priorExit !== null) { try { await destroyContainerExpectingExit(this.#ctx, this.#container); @@ -120,11 +131,17 @@ export class WorkspaceContainerAPI extends RpcTarget implements IWorkspaceContai // best-effort — the next start() will surface any real // platform-side failure. } - this.#container.start({ enableInternet, env }); + runtime = await this.#launch(env, enableInternet); } else if (!this.#container.running) { - this.#container.start({ enableInternet, env }); + runtime = await this.#launch(env, enableInternet); + } else { + // A Durable Object incarnation can be reconstructed while its + // container stays alive. Reuse the durable runtime id rather + // than treating the new WebSocket as a new process. + runtime = (await this.#runtimeIdentity.get()) ?? (await this.#runtimeIdentity.markStarted()); } - installContainerMonitor(this.#ctx, this.#container); + installContainerMonitor(this.#ctx, this.#container, () => this.#runtimeIdentity.clear(runtime)); + return { runtimeId: runtime.id }; } async restart(env: Record, enableInternet: boolean) { @@ -140,8 +157,20 @@ export class WorkspaceContainerAPI extends RpcTarget implements IWorkspaceContai // succeed against a fresh generation or surface its own // failure. } - this.#container.start({ enableInternet, env }); - installContainerMonitor(this.#ctx, this.#container); + const runtime = await this.#launch(env, enableInternet); + installContainerMonitor(this.#ctx, this.#container, () => this.#runtimeIdentity.clear(runtime)); + return { runtimeId: runtime.id }; + } + + async #launch(env: Record, enableInternet: boolean) { + const runtime = await this.#runtimeIdentity.markStarted(); + try { + this.#container.start({ enableInternet, env }); + return runtime; + } catch (error) { + await this.#runtimeIdentity.clear(runtime); + throw error; + } } async status() { diff --git a/packages/computer/src/backends/container/container-lifecycle.test.ts b/packages/computer/src/backends/container/container-lifecycle.test.ts index ee8f6968..69315621 100644 --- a/packages/computer/src/backends/container/container-lifecycle.test.ts +++ b/packages/computer/src/backends/container/container-lifecycle.test.ts @@ -152,12 +152,13 @@ describe("installContainerMonitor", () => { expect(exit?.exitedAt).toBe(Date.now()); }); - test("records the rejection reason when the monitor rejects", async () => { + test("records the rejection reason and runs current-generation cleanup", async () => { const fake = makeContainer(); const ctx = makeContext(fake.container); + const cleanup = vi.fn(async () => {}); resetContainerLifecycleForTests(ctx); fake.container.start(); - installContainerMonitor(ctx, fake.container); + installContainerMonitor(ctx, fake.container, cleanup); fake.current.reject(new Error("container crashed")); await Promise.resolve(); @@ -165,6 +166,7 @@ describe("installContainerMonitor", () => { const exit = containerExitInfo(ctx); expect(exit?.reason).toBe("container crashed"); + expect(cleanup).toHaveBeenCalledOnce(); }); test("logs at warn level on an unexpected exit", async () => { @@ -226,9 +228,10 @@ describe("installContainerMonitor", () => { const ctx = makeContext(fake.container); resetContainerLifecycleForTests(ctx); + const staleCleanup = vi.fn(async () => {}); fake.container.start(); const firstGeneration = fake.current; - installContainerMonitor(ctx, fake.container); + installContainerMonitor(ctx, fake.container, staleCleanup); // Second generation — a new monitor promise is armed in the // fake's start(); installContainerMonitor bumps the lifecycle's @@ -245,6 +248,7 @@ describe("installContainerMonitor", () => { await Promise.resolve(); await Promise.resolve(); expect(containerExitInfo(ctx)).toBeNull(); + expect(staleCleanup).not.toHaveBeenCalled(); // The current generation's monitor still records normally. secondGeneration.reject(new Error("current generation died")); diff --git a/packages/computer/src/backends/container/container-lifecycle.ts b/packages/computer/src/backends/container/container-lifecycle.ts index f73896af..b6ebd272 100644 --- a/packages/computer/src/backends/container/container-lifecycle.ts +++ b/packages/computer/src/backends/container/container-lifecycle.ts @@ -88,7 +88,11 @@ export function formatExitReason(error: unknown): string { // generation". If a caller arms twice for the same generation, // the worst that happens is two handlers race to write the same // exit state — same value, same generation. -export function installContainerMonitor(ctx: DurableObjectState, container: ContainerHandle): void { +export function installContainerMonitor( + ctx: DurableObjectState, + container: ContainerHandle, + onExit?: () => void | Promise, +): void { const state = getContainerLifecycle(ctx); state.currentGeneration += 1; const generation = state.currentGeneration; @@ -102,8 +106,8 @@ export function installContainerMonitor(ctx: DurableObjectState, container: Cont // continues. Both branches feed into recordExit which never // throws, so the wrapper itself resolves rather than rejecting. state.currentMonitorSettled = monitorPromise.then( - () => recordExit(state, generation, undefined), - (error) => recordExit(state, generation, error), + () => recordExit(state, generation, undefined, onExit), + (error) => recordExit(state, generation, error, onExit), ); } @@ -147,7 +151,12 @@ export function resetContainerLifecycleForTests(ctx: DurableObjectState): void { LIFECYCLE.delete(ctx); } -function recordExit(state: ContainerLifecycleState, generation: number, error: unknown): void { +async function recordExit( + state: ContainerLifecycleState, + generation: number, + error: unknown, + onExit?: () => void | Promise, +): Promise { // Drop late writes from superseded monitors. The state object is // shared across generations; only the current one is allowed to // mutate `exit`. Log nothing on a stale exit — the operator @@ -178,4 +187,13 @@ function recordExit(state: ContainerLifecycleState, generation: number, error: u expected: false, }); } + try { + await onExit?.(); + } catch (cleanupError) { + console.error({ + message: "workspace.container.exit_cleanup_failed", + reason: formatExitReason(cleanupError), + exitedAt: Date.now(), + }); + } } diff --git a/packages/computer/src/backends/container/container-runtime-identity.test.ts b/packages/computer/src/backends/container/container-runtime-identity.test.ts new file mode 100644 index 00000000..15e26954 --- /dev/null +++ b/packages/computer/src/backends/container/container-runtime-identity.test.ts @@ -0,0 +1,55 @@ +import { describe, expect, it, vi } from "vitest"; + +import { CurrentContainerRuntimeIdentity } from "./container-runtime-identity.js"; + +function storage(initial = new Map()) { + return { + get: vi.fn(async (key: string) => initial.get(key)), + put: vi.fn(async (key: string, value: unknown) => { + initial.set(key, value); + }), + delete: vi.fn(async (key: string) => initial.delete(key)), + }; +} + +describe("CurrentContainerRuntimeIdentity", () => { + it("persists a UUID for a newly started container runtime", async () => { + const backing = new Map(); + const current = new CurrentContainerRuntimeIdentity(storage(backing)); + + const runtime = await current.markStarted(); + + expect(runtime.id).toMatch(/^[0-9a-f-]{36}$/); + await expect(current.get()).resolves.toEqual(runtime); + }); + + it("returns the same stored identity across helper instances", async () => { + const backing = new Map(); + const first = new CurrentContainerRuntimeIdentity(storage(backing)); + const started = await first.markStarted(); + const recreated = new CurrentContainerRuntimeIdentity(storage(backing)); + + await expect(recreated.get()).resolves.toEqual(started); + }); + + it("creates a new UUID for each container runtime", async () => { + const current = new CurrentContainerRuntimeIdentity(storage()); + + const first = await current.markStarted(); + const second = await current.markStarted(); + + expect(second.id).not.toBe(first.id); + }); + + it("clears only the runtime identity that stopped", async () => { + const current = new CurrentContainerRuntimeIdentity(storage()); + const stale = await current.markStarted(); + const active = await current.markStarted(); + + await current.clear(stale); + await expect(current.get()).resolves.toEqual(active); + + await current.clear(active); + await expect(current.get()).resolves.toBeNull(); + }); +}); diff --git a/packages/computer/src/backends/container/container-runtime-identity.ts b/packages/computer/src/backends/container/container-runtime-identity.ts new file mode 100644 index 00000000..d948696b --- /dev/null +++ b/packages/computer/src/backends/container/container-runtime-identity.ts @@ -0,0 +1,36 @@ +// Durable identity for the currently-running container process. +// +// A WebSocket reconnect keeps this id. Starting a replacement process +// writes a new UUID. Execution-scoped operations use it to distinguish +// reconnecting to the same computerd from reaching an empty replacement. + +export interface ContainerRuntimeIdentity { + id: string; +} + +interface RuntimeIdentityStorage { + get(key: string): Promise; + put(key: string, value: unknown): Promise; + delete(key: string): Promise; +} + +const STORAGE_KEY = "computer:container-runtime-identity"; + +export class CurrentContainerRuntimeIdentity { + constructor(private readonly storage: RuntimeIdentityStorage) {} + + async get(): Promise { + return (await this.storage.get(STORAGE_KEY)) ?? null; + } + + async markStarted(): Promise { + const runtime = { id: crypto.randomUUID() }; + await this.storage.put(STORAGE_KEY, runtime); + return runtime; + } + + async clear(runtime: ContainerRuntimeIdentity): Promise { + const current = await this.get(); + if (current?.id === runtime.id) await this.storage.delete(STORAGE_KEY); + } +} diff --git a/packages/computer/src/execution-runtime-tracker.test.ts b/packages/computer/src/execution-runtime-tracker.test.ts new file mode 100644 index 00000000..d4757694 --- /dev/null +++ b/packages/computer/src/execution-runtime-tracker.test.ts @@ -0,0 +1,75 @@ +import { describe, expect, it } from "vitest"; + +import { + type ExecutionRuntimeStore, + ExecutionRuntimeTracker, +} from "./execution-runtime-tracker.js"; + +function durableStore(backing = new Map()): ExecutionRuntimeStore { + return { + get: (key) => backing.get(key), + remember: (key, runtimeId) => backing.set(key, runtimeId), + delete: (key, expectedRuntimeId) => { + if (expectedRuntimeId === undefined || backing.get(key) === expectedRuntimeId) { + backing.delete(key); + } + }, + }; +} + +describe("ExecutionRuntimeTracker", () => { + it("evicts the least recently used execution at its limit", () => { + const tracker = new ExecutionRuntimeTracker(2); + tracker.remember("a", "runtime-a"); + tracker.remember("b", "runtime-a"); + + expect(tracker.get("a")).toBe("runtime-a"); + tracker.remember("c", "runtime-b"); + + expect(tracker.get("a")).toBe("runtime-a"); + expect(tracker.get("b")).toBeUndefined(); + expect(tracker.get("c")).toBe("runtime-b"); + }); + + it("updates an existing execution without growing the cache", () => { + const tracker = new ExecutionRuntimeTracker(2); + tracker.remember("a", "runtime-a"); + tracker.remember("b", "runtime-a"); + tracker.remember("a", "runtime-b"); + tracker.remember("c", "runtime-c"); + + expect(tracker.get("a")).toBe("runtime-b"); + expect(tracker.get("b")).toBeUndefined(); + expect(tracker.get("c")).toBe("runtime-c"); + }); + + it("falls back to durable state after cache eviction", () => { + const store = durableStore(); + const tracker = new ExecutionRuntimeTracker(2, store); + tracker.remember("a", "runtime-a"); + tracker.remember("b", "runtime-b"); + tracker.remember("c", "runtime-c"); + + expect(tracker.get("a")).toBe("runtime-a"); + }); + + it("restores runtime ownership in a new Workspace incarnation", () => { + const store = durableStore(); + const first = new ExecutionRuntimeTracker(2, store); + first.remember("execution", "runtime-a"); + + const recreated = new ExecutionRuntimeTracker(2, store); + expect(recreated.get("execution")).toBe("runtime-a"); + }); + + it("deletes only the expected runtime owner", () => { + const tracker = new ExecutionRuntimeTracker(2); + tracker.remember("a", "runtime-b"); + + tracker.delete("a", "runtime-a"); + expect(tracker.get("a")).toBe("runtime-b"); + + tracker.delete("a", "runtime-b"); + expect(tracker.get("a")).toBeUndefined(); + }); +}); diff --git a/packages/computer/src/execution-runtime-tracker.ts b/packages/computer/src/execution-runtime-tracker.ts new file mode 100644 index 00000000..ee1c8581 --- /dev/null +++ b/packages/computer/src/execution-runtime-tracker.ts @@ -0,0 +1,100 @@ +// Bounded runtime ownership cache for process-local execution ids. +// Returned execution handles carry their runtime id directly; durable +// fallback keeps public by-id lifecycle methods fenced across Durable +// Object incarnations and cache eviction. + +import type { Database } from "@cloudflare/dofs"; + +const DEFAULT_MAX_ENTRIES = 1_024; + +export interface ExecutionRuntimeStore { + get(key: string): string | undefined; + remember(key: string, runtimeId: string): void; + delete(key: string, expectedRuntimeId?: string): void; +} + +export class SqlExecutionRuntimeStore implements ExecutionRuntimeStore { + constructor(private readonly db: Database) { + this.db.run(` + CREATE TABLE IF NOT EXISTS computer_execution_runtime ( + execution_key TEXT PRIMARY KEY, + runtime_id TEXT NOT NULL + ) + `); + } + + get(key: string): string | undefined { + return this.db.scalar( + "SELECT runtime_id FROM computer_execution_runtime WHERE execution_key = ?", + key, + ); + } + + remember(key: string, runtimeId: string): void { + this.db.run( + `INSERT INTO computer_execution_runtime (execution_key, runtime_id) + VALUES (?, ?) + ON CONFLICT(execution_key) DO UPDATE SET runtime_id = excluded.runtime_id`, + key, + runtimeId, + ); + } + + delete(key: string, expectedRuntimeId?: string): void { + if (expectedRuntimeId === undefined) { + this.db.run("DELETE FROM computer_execution_runtime WHERE execution_key = ?", key); + return; + } + this.db.run( + "DELETE FROM computer_execution_runtime WHERE execution_key = ? AND runtime_id = ?", + key, + expectedRuntimeId, + ); + } +} + +export class ExecutionRuntimeTracker { + readonly #entries = new Map(); + readonly #maxEntries: number; + readonly #store: ExecutionRuntimeStore | undefined; + + constructor(maxEntries = DEFAULT_MAX_ENTRIES, store?: ExecutionRuntimeStore) { + if (!Number.isSafeInteger(maxEntries) || maxEntries <= 0) { + throw new Error("ExecutionRuntimeTracker maxEntries must be a positive integer"); + } + this.#maxEntries = maxEntries; + this.#store = store; + } + + get(key: string): string | undefined { + const cached = this.#entries.get(key); + if (cached !== undefined) { + this.#touch(key, cached); + return cached; + } + const stored = this.#store?.get(key); + if (stored !== undefined) this.#touch(key, stored); + return stored; + } + + remember(key: string, runtimeId: string): void { + this.#store?.remember(key, runtimeId); + this.#touch(key, runtimeId); + } + + delete(key: string, expectedRuntimeId?: string): void { + this.#store?.delete(key, expectedRuntimeId); + if (expectedRuntimeId !== undefined && this.#entries.get(key) !== expectedRuntimeId) return; + this.#entries.delete(key); + } + + #touch(key: string, runtimeId: string): void { + // Map iteration order is insertion order. Reinsert reads so the + // first key remains the least recently used one. + this.#entries.delete(key); + this.#entries.set(key, runtimeId); + if (this.#entries.size <= this.#maxEntries) return; + const oldest = this.#entries.keys().next().value; + if (oldest !== undefined) this.#entries.delete(oldest); + } +} diff --git a/packages/computer/src/retry.test.ts b/packages/computer/src/retry.test.ts index f34a9734..26575272 100644 --- a/packages/computer/src/retry.test.ts +++ b/packages/computer/src/retry.test.ts @@ -3,6 +3,7 @@ import { SQLiteTestStorage } from "@cloudflare/dofs/testing"; import { describe, expect, it } from "vitest"; import type { BackendHandle, WorkspaceBackend } from "./backend.js"; +import { WorkspaceTransportError } from "./transport-failure.js"; import type { SyncRetryIntent, SyncRetryScheduler, @@ -95,6 +96,119 @@ async function runCommand(ws: Workspace): Promise { + it("clears a lost runtime intent so the live runtime can schedule a new one", async () => { + const scheduler = new MemoryRetryScheduler(); + let connects = 0; + let replacementPulls = 0; + let failLivePull = false; + const backend: WorkspaceBackend = { + id: "sandbox", + type: "fake", + async connect(): Promise { + const connection = connects++; + const runtimeId = connection === 0 ? "runtime-a" : "runtime-b"; + const sync = retryBackend({ + onExec() {}, + async fetchChanges() { + if (connection === 0) { + throw new WorkspaceTransportError("container exited during post-exec pull"); + } + if (failLivePull) throw new Error("live runtime pull failed"); + replacementPulls++; + return { + currentCursor: { rev: 0, path: null }, + appliedPushCursor: { rev: 0, path: null }, + stream: new ReadableStream({ + start(controller) { + controller.close(); + }, + }), + }; + }, + }); + const handle = await sync.connect({} as never); + return { ...handle, runtimeId }; + }, + }; + const ws = new Workspace({ + storage: new SQLiteTestStorage(), + backends: [backend], + retryScheduler: scheduler, + retry: { initialDelayMs: 100, maxDelayMs: 1_000, maxAttempts: 3 }, + now: () => 5_000, + }); + + const handle = await ws.runtime.exec("build", { encoding: "utf8" }); + const result = await handle.result(); + + expect(result.sync).toMatchObject({ + status: "pending", + error: expect.stringContaining("container exited during post-exec pull"), + }); + expect(replacementPulls).toBe(0); + expect(scheduler.intents.get("sandbox")).toEqual({ + backend: "sandbox", + runtimeId: "runtime-a", + attempt: 1, + notBefore: 5_100, + }); + + await expect(ws.retryPendingSync("sandbox")).resolves.toMatchObject({ + status: "lost", + runtimeId: "runtime-a", + }); + expect(replacementPulls).toBe(0); + expect(scheduler.intents.has("sandbox")).toBe(false); + expect(scheduler.cleared).toEqual(["sandbox"]); + + failLivePull = true; + await runCommand(ws); + expect(scheduler.intents.get("sandbox")).toEqual({ + backend: "sandbox", + runtimeId: "runtime-b", + attempt: 1, + notBefore: 5_100, + }); + }); + + it("replaces a stale runtime intent when a live runtime pull becomes pending", async () => { + const scheduler = new MemoryRetryScheduler(); + scheduler.intents.set("sandbox", { + backend: "sandbox", + runtimeId: "runtime-a", + attempt: 3, + notBefore: 0, + }); + const base = retryBackend({ + onExec() {}, + async fetchChanges() { + throw new Error("runtime-b pull failed"); + }, + }); + const backend: WorkspaceBackend = { + ...base, + async connect(host): Promise { + return { ...(await base.connect(host)), runtimeId: "runtime-b" }; + }, + }; + const ws = new Workspace({ + storage: new SQLiteTestStorage(), + backends: [backend], + retryScheduler: scheduler, + retry: { initialDelayMs: 100, maxDelayMs: 1_000, maxAttempts: 3 }, + now: () => 5_000, + }); + + await runCommand(ws); + + expect(scheduler.intents.get("sandbox")).toEqual({ + backend: "sandbox", + runtimeId: "runtime-b", + attempt: 1, + notBefore: 5_100, + }); + }); + it("schedules the exact durable retry intent after a post-command pull failure", async () => { const scheduler = new MemoryRetryScheduler(); let execs = 0; diff --git a/packages/computer/src/runtime/runtime.ts b/packages/computer/src/runtime/runtime.ts index 3a7a61f8..ccae0577 100644 --- a/packages/computer/src/runtime/runtime.ts +++ b/packages/computer/src/runtime/runtime.ts @@ -78,6 +78,7 @@ export class WorkspaceRuntime { options.encoding, true, envelope.sync, + envelope.runtimeId, ); } @@ -106,6 +107,7 @@ export class WorkspaceRuntime { options.encoding, options.resume === undefined || options.resume === "full", envelope.sync, + envelope.runtimeId, ); } @@ -140,6 +142,7 @@ function wrapModuleHandle( encoding: E | undefined, resultMayUseSource = true, sync?: ModuleExecutionEnvelope["sync"], + runtimeId?: string, ): WorkspaceRuntimeExecHandle { let claimed: "result" | "stream" | undefined; let sourceCancelled = false; @@ -201,14 +204,15 @@ function wrapModuleHandle( return drainModuleResult(source, encoding, setReader, sync); } if (!sourceCancelled) await source.cancel("result() requested a full replay"); - const replay = await runtime.getExec({ id }); + const replay = await runtime.getExec({ id, runtimeId }); return drainModuleResult(replay.events, encoding, setReader, replay.sync); })(); return resultPromise; }, }, kill: { - value: (signal?: WorkspaceRuntimeKillOptions["signal"]) => runtime.killExec({ id, signal }), + value: (signal?: WorkspaceRuntimeKillOptions["signal"]) => + runtime.killExec({ id, signal, runtimeId }), }, [Symbol.dispose]: { value: () => { diff --git a/packages/computer/src/runtime/types.ts b/packages/computer/src/runtime/types.ts index cf3d9931..a909f494 100644 --- a/packages/computer/src/runtime/types.ts +++ b/packages/computer/src/runtime/types.ts @@ -148,6 +148,10 @@ export interface ModuleExecutionInput { export interface ModuleExecutionEnvelope { id: string; + // Identity of the process-local backend runtime that owns this + // execution. Omitted by backends whose execution state is durable or + // shared with the host. + runtimeId?: string; events: ReadableStream; // Sync bracket stats for a backend that pairs with a remote store. // The pre-exec push count is known when the envelope is created; @@ -162,9 +166,13 @@ export interface ModuleExecutionEnvelope { export interface WorkspaceModuleBackendHandle { exec(input: ModuleExecutionInput): Promise; - getExec(input: { id: string; after?: number | "tail" }): Promise; - killExec(input: { id: string; signal?: KillSignal }): Promise; - disposeExec(input: { id: string }): Promise; + getExec(input: { + id: string; + after?: number | "tail"; + runtimeId?: string; + }): Promise; + killExec(input: { id: string; signal?: KillSignal; runtimeId?: string }): Promise; + disposeExec(input: { id: string; runtimeId?: string }): 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/shell.test.ts b/packages/computer/src/shell.test.ts index bcbd8fdb..07928c19 100644 --- a/packages/computer/src/shell.test.ts +++ b/packages/computer/src/shell.test.ts @@ -365,7 +365,7 @@ describe("CommandExecutor.exec — push/pull bracket", () => { expect(order).toEqual(["push", "pull"]); // pull fired after drain }); - it("falls back to pushed = 0 when sync.push() throws", async () => { + it("fails before spawn when the pre-exec push throws", async () => { const f = fakeRpc({ events: [exit(1, 0)] }); const sync: Sync = { async push() { @@ -375,11 +375,11 @@ describe("CommandExecutor.exec — push/pull bracket", () => { return applied(3); }, }; - const execution = await new CommandExecutor(f.rpc.shell, sync).exec("noop"); - expect(execution.sync.pushed).toBe(0); - // pull still fires — docs/05 says one failed half doesn't abort the other - const { outcome } = await drain(execution); - expect((outcome as { applied: number }).applied).toBe(3); + + await expect(new CommandExecutor(f.rpc.shell, sync).exec("noop")).rejects.toThrow( + "push offline", + ); + expect(f.calls.exec).toHaveLength(0); }); it("reports a pending sync after a Durable Object storage reset", async () => { diff --git a/packages/computer/src/shell.ts b/packages/computer/src/shell.ts index eb2a267c..1747862e 100644 --- a/packages/computer/src/shell.ts +++ b/packages/computer/src/shell.ts @@ -49,6 +49,7 @@ export type KillSignal = "SIGTERM" | "SIGKILL" | "SIGINT" | "SIGHUP"; // reaches its end, carrying the post-drain pull result. export interface CommandExecution { id: string; + runtimeId?: string; events: ReadableStream; sync: { pushed: number; outcome: Promise }; } @@ -79,6 +80,9 @@ export interface GetExecOptions { // number resumes from that seq+1. Omit to receive every // event from the start of the run (replays the whole log). resume?: "tail" | "full" | number; + // Expected process identity for process-local replay. Internal to + // the Workspace command adapter; direct ShellRPC callers omit it. + runtimeId?: string; } // Push/pull bracket plumbing. CommandExecutor doesn't know about @@ -90,65 +94,82 @@ export interface GetExecOptions { // skipped read-only entries on the sync outcome. export interface Sync { push(): Promise; - pull(): Promise; - onPullPending?(error: unknown): Promise; + pull(runtimeId?: string): Promise; + onPullPending?(error: unknown, runtimeId?: string): Promise; } +type ShellExecInput = Parameters[0]; +type ShellGetInput = Parameters[0] & { runtimeId?: string }; +type ShellExecEnvelope = Awaited> & { runtimeId?: string }; + +// Optional host-owned dispatch boundary. The container Workspace uses +// this to keep its pre-exec push and shell spawn on one backend handle; +// simpler executors use Sync.push followed by the supplied ShellRPC. +export interface CommandDispatchResult { + pushed: number; + envelope: ShellExecEnvelope; +} + +export type CommandDispatch = (input: ShellExecInput) => Promise; +export type CommandGetDispatch = (input: ShellGetInput) => Promise; + export class CommandExecutor { readonly #shell: ShellRPC; readonly #sync: Sync; readonly #observer: WorkspaceObserver; + readonly #dispatch: CommandDispatch | undefined; + readonly #getDispatch: CommandGetDispatch | undefined; - constructor(shell: ShellRPC, sync: Sync, observer: WorkspaceObserver = noopObserver) { + constructor( + shell: ShellRPC, + sync: Sync, + observer: WorkspaceObserver = noopObserver, + dispatch?: CommandDispatch, + getDispatch?: CommandGetDispatch, + ) { this.#shell = shell; this.#sync = sync; this.#observer = observer; + this.#dispatch = dispatch; + this.#getDispatch = getDispatch; } // Spawn a command. Pushes host-side writes first so the command // sees them, then returns the raw event stream and the sync - // bracket stats. The push failure is non-fatal per docs/05 — the - // command still runs and pushed reports 0. + // bracket stats. A failed push aborts before shell.exec: running + // with stale or incomplete workspace contents is not safe. async exec(source: string, options: ExecOptions = {}): Promise { assertNotTemplate(source); - let pushed = 0; - try { + const input: ShellExecInput = { + source, + id: options.id, + cwd: options.cwd, + timeoutMs: options.timeoutMs, + env: options.env, + stdin: + typeof options.stdin === "string" ? new TextEncoder().encode(options.stdin) : options.stdin, + }; + let pushed: number; + let envelope: ShellExecEnvelope; + if (this.#dispatch !== undefined) { + ({ pushed, envelope } = await this.#dispatch(input)); + } else { pushed = await this.#sync.push(); - } catch { - // pushed stays 0 + envelope = await spawnShell(this.#shell, input, this.#observer); } - const envelope = await withSpan( - this.#observer, - "workspace.runtime.exec.spawn", - { - "workspace.runtime.cwd": options.cwd, - "workspace.runtime.timeout_ms": options.timeoutMs, - "workspace.runtime.id": options.id, - }, - () => - this.#shell.exec({ - source, - id: options.id, - cwd: options.cwd, - timeoutMs: options.timeoutMs, - env: options.env, - stdin: - typeof options.stdin === "string" - ? new TextEncoder().encode(options.stdin) - : options.stdin, - }), - (span, outcome) => { - if (outcome.ok) span.setAttribute("workspace.runtime.id", outcome.value.id); - }, - ); // Dispose the RPC envelope when the event stream finishes // draining. Without this, capnweb's exports table holds onto // the envelope for the life of the session — one entry per exec // call — because the inner stream is handed off to the caller // and the envelope can't be bound with `using` here. const drained = disposeOnDone(envelope.events, () => maybeDispose(envelope)); - const { stream, outcome } = withPostPull(drained, this.#sync); - return { id: envelope.id, events: stream, sync: { pushed, outcome } }; + const { stream, outcome } = withPostPull(drained, this.#sync, envelope.runtimeId); + return { + id: envelope.id, + runtimeId: envelope.runtimeId, + events: stream, + sync: { pushed, outcome }, + }; } // Reattach to an in-flight or recently-completed exec. Reattach @@ -157,10 +178,13 @@ export class CommandExecutor { // reattach and drain. async get(id: string, options: GetExecOptions = {}): Promise { const after = resumeToAfter(options.resume); - const envelope = await this.#shell.getExec({ id, after }); + const envelope: ShellExecEnvelope = + this.#getDispatch === undefined + ? await this.#shell.getExec({ id, after }) + : await this.#getDispatch({ id, after, runtimeId: options.runtimeId }); const drained = disposeOnDone(envelope.events, () => maybeDispose(envelope)); - const { stream, outcome } = withPostPull(drained, this.#sync); - return { id, events: stream, sync: { pushed: 0, outcome } }; + const { stream, outcome } = withPostPull(drained, this.#sync, envelope.runtimeId); + return { id, runtimeId: envelope.runtimeId, events: stream, sync: { pushed: 0, outcome } }; } kill(id: string, signal?: KillSignal): Promise { @@ -172,6 +196,26 @@ export class CommandExecutor { } } +export function spawnShell( + shell: ShellRPC, + input: ShellExecInput, + observer: WorkspaceObserver = noopObserver, +): Promise { + return withSpan( + observer, + "workspace.runtime.exec.spawn", + { + "workspace.runtime.cwd": input.cwd, + "workspace.runtime.timeout_ms": input.timeoutMs, + "workspace.runtime.id": input.id, + }, + () => shell.exec(input), + (span, outcome) => { + if (outcome.ok) span.setAttribute("workspace.runtime.id", outcome.value.id); + }, + ); +} + function resumeToAfter(resume: "tail" | "full" | number | undefined): number | "tail" | undefined { if (resume === undefined || resume === "full") return undefined; if (resume === "tail") return "tail"; @@ -187,6 +231,7 @@ export interface PostPullOutcome { export function withPostPull( source: ReadableStream, sync: Sync, + runtimeId?: string, ): { stream: ReadableStream; outcome: Promise } { const reader = source.getReader(); let resolveOutcome!: (outcome: PostPullOutcome) => void; @@ -203,7 +248,7 @@ export function withPostPull( return; } reader.releaseLock(); - const pulled = await runPostPull(sync); + const pulled = await runPostPull(sync, runtimeId); resolveOutcome(pulled); controller.close(); } catch (error) { @@ -236,9 +281,9 @@ export function withPostPull( return { stream, outcome }; } -async function runPostPull(sync: Sync): Promise { +async function runPostPull(sync: Sync, runtimeId?: string): Promise { try { - const result = await sync.pull(); + const result = await sync.pull(runtimeId); return { applied: result.applied, skipped: result.skipped, @@ -246,7 +291,7 @@ async function runPostPull(sync: Sync): Promise { }; } catch (error) { try { - await sync.onPullPending?.(error); + await sync.onPullPending?.(error, runtimeId); } catch {} return { applied: 0, diff --git a/packages/computer/src/stub.test.ts b/packages/computer/src/stub.test.ts index e3448ae2..2c0aeca3 100644 --- a/packages/computer/src/stub.test.ts +++ b/packages/computer/src/stub.test.ts @@ -47,14 +47,16 @@ function composite( } function fakeSync(): import("@cloudflare/computer-rpc").SyncRPC { + let appliedPushRev = 0; return { - async push() { - return { rev: 0, appliedPushCursor: { rev: 0, path: null } }; + async push(input) { + appliedPushRev = input.senderRev; + return { rev: 0, appliedPushCursor: { rev: appliedPushRev, path: null } }; }, async fetchChanges() { return { currentCursor: { rev: 0, path: null }, - appliedPushCursor: { rev: 0, path: null }, + appliedPushCursor: { rev: appliedPushRev, path: null }, stream: new ReadableStream({ start(c) { c.close(); @@ -66,7 +68,11 @@ function fakeSync(): import("@cloudflare/computer-rpc").SyncRPC { return null; }, async watermarks() { - return { currentRev: 0, pushRev: 0, fetchCursor: { rev: 0, path: null } }; + return { + currentRev: 0, + pushRev: 0, + fetchCursor: { rev: appliedPushRev, path: null }, + }; }, async hasObjects() { return []; diff --git a/packages/computer/src/transport-failure.test.ts b/packages/computer/src/transport-failure.test.ts index aa175a5b..c6354f9f 100644 --- a/packages/computer/src/transport-failure.test.ts +++ b/packages/computer/src/transport-failure.test.ts @@ -1,6 +1,11 @@ import { describe, expect, it } from "vitest"; -import { isWorkspaceTransportFailure, WorkspaceTransportError } from "./transport-failure.js"; +import { + isWorkspacePreDispatchTransportFailure, + isWorkspaceTransportFailure, + WorkspacePreDispatchTransportError, + WorkspaceTransportError, +} from "./transport-failure.js"; describe("isWorkspaceTransportFailure", () => { it("recognises WorkspaceTransportError instances", () => { @@ -20,6 +25,42 @@ describe("isWorkspaceTransportFailure", () => { ).toBe(true); }); + it("only treats a locally disposed stub as proof that dispatch did not start", () => { + expect( + isWorkspacePreDispatchTransportFailure( + new Error("Attempted to use RPC stub after it has been disposed."), + ), + ).toBe(true); + expect( + isWorkspacePreDispatchTransportFailure( + new Error("RPC was canceled because RPC session was shut down"), + ), + ).toBe(false); + expect(isWorkspacePreDispatchTransportFailure(new Error("WebSocket closed unexpectedly"))).toBe( + false, + ); + }); + + it("recognises an explicitly pre-dispatch transport failure", () => { + const error = new WorkspacePreDispatchTransportError("pre-exec push failed"); + expect(isWorkspaceTransportFailure(error)).toBe(true); + expect(isWorkspacePreDispatchTransportFailure(error)).toBe(true); + + const cloned = new Error(error.message); + cloned.name = error.name; + expect(isWorkspaceTransportFailure(cloned)).toBe(true); + expect(isWorkspacePreDispatchTransportFailure(cloned)).toBe(true); + }); + + it("walks causes when classifying a pre-dispatch failure", () => { + const inner = new Error("Attempted to use RPC stub after it has been disposed."); + expect( + isWorkspacePreDispatchTransportFailure( + new WorkspaceTransportError("shell unavailable", { cause: inner }), + ), + ).toBe(true); + }); + it("recognises WebSocket transport failures", () => { expect(isWorkspaceTransportFailure(new Error("WebSocket is not open"))).toBe(true); expect(isWorkspaceTransportFailure(new Error("WebSocket closed unexpectedly"))).toBe(true); diff --git a/packages/computer/src/transport-failure.ts b/packages/computer/src/transport-failure.ts index 5dbe8cba..77168c0d 100644 --- a/packages/computer/src/transport-failure.ts +++ b/packages/computer/src/transport-failure.ts @@ -31,7 +31,7 @@ // inspection, and lets callers wrap an underlying cause for // observability without losing the classification. export class WorkspaceTransportError extends Error { - override readonly name = "WorkspaceTransportError"; + override readonly name: string = "WorkspaceTransportError"; constructor(message: string, options?: { cause?: unknown }) { super(message); @@ -41,9 +41,23 @@ export class WorkspaceTransportError extends Error { } } +// A transport failure that happened before a side-effecting request +// was dispatched. The reconnect loop may replay the whole operation, +// including any completed preflight needed by the replacement handle. +export class WorkspacePreDispatchTransportError extends WorkspaceTransportError { + override readonly name = "WorkspacePreDispatchTransportError"; +} + // Phrases that consistently indicate a dead transport. Conservative // list — only phrases the workspace stack reliably produces or // that capnweb / Cloudflare Workers ws emit on a closed session. +const PRE_DISPATCH_PATTERNS: RegExp[] = [ + // This is thrown locally when code calls a stub whose session is + // already disposed. No frame can have been sent, so even a + // side-effecting operation such as shell.exec is safe to retry. + /rpc stub after it has been disposed/i, +]; + const TRANSPORT_PATTERNS: RegExp[] = [ // capnweb session shutdown / cancellation. Phrasing checked // against node_modules/capnweb/dist/index.js: the @@ -51,7 +65,7 @@ const TRANSPORT_PATTERNS: RegExp[] = [ // has been disposed." and pre-shutdown cancellation reads as // "RPC was canceled because RPC session was shut down...". /rpc session was shut down/i, - /rpc stub after it has been disposed/i, + ...PRE_DISPATCH_PATTERNS, /rpc was canceled/i, // WebSocket transport failures. /websocket is not open/i, @@ -82,7 +96,12 @@ export function isWorkspaceTransportFailure(error: unknown): boolean { // .name survives a Workers RPC structured-clone hop even // when the subclass identity does not, so cross-DO callers // still classify a WorkspaceTransportError correctly. - if (current.name === "WorkspaceTransportError") return true; + if ( + current.name === "WorkspaceTransportError" || + current.name === "WorkspacePreDispatchTransportError" + ) { + return true; + } for (const pattern of TRANSPORT_PATTERNS) { if (pattern.test(current.message)) return true; } @@ -95,3 +114,28 @@ export function isWorkspaceTransportFailure(error: unknown): boolean { } return false; } + +// A transport failure does not normally reveal whether the peer +// accepted an operation before the connection died. Keep a narrower +// classifier for failures that prove dispatch never happened. Shell +// exec uses this to avoid replaying a command with unknown side +// effects; idempotent sync operations use the broader classifier. +export function isWorkspacePreDispatchTransportFailure(error: unknown): boolean { + let current: unknown = error; + for (let depth = 0; depth < 8 && current !== undefined && current !== null; depth++) { + if (!(current instanceof Error)) return false; + if ( + current instanceof WorkspacePreDispatchTransportError || + current.name === "WorkspacePreDispatchTransportError" + ) { + return true; + } + for (const pattern of PRE_DISPATCH_PATTERNS) { + if (pattern.test(current.message)) return true; + } + const next = (current as Error & { cause?: unknown }).cause; + if (next === current) return false; + current = next; + } + return false; +} diff --git a/packages/computer/src/workspace.test.ts b/packages/computer/src/workspace.test.ts index 95d3dc05..12a3d5b5 100644 --- a/packages/computer/src/workspace.test.ts +++ b/packages/computer/src/workspace.test.ts @@ -1165,101 +1165,187 @@ describe("Workspace mutation serialization", () => { }); describe("Workspace transport-failure invalidation", () => { - // A backend whose RPC throws a transport-like error while its - // `closed` promise never resolves used to leave the Workspace - // holding a dead handle. The cache must drop the handle on the - // way out so the next operation reconnects. - function transportFailingBackend( - id: string, - onConnect: () => void, - ): { backend: WorkspaceBackend; failNext: () => void } { - let shouldFail = false; - const sync: import("@cloudflare/computer-rpc").SyncRPC = { + it("push() reconnects and succeeds within the same call", async () => { + let connects = 0; + let closes = 0; + const replacement = fakeRpc(); + const stale: import("@cloudflare/computer-rpc").SyncRPC = { ...fakeRpc(), - async push(input) { - if (shouldFail) throw new WorkspaceTransportError("WebSocket closed"); - const reader = input.changes.getReader(); - try { - while (true) { - const { done } = await reader.read(); - if (done) break; - } - } finally { - reader.releaseLock(); - } - return { rev: 0, appliedPushCursor: { rev: input.senderRev, path: null } }; + async push() { + throw new WorkspaceTransportError("WebSocket closed"); }, - async fetchChanges() { - if (shouldFail) throw new WorkspaceTransportError("WebSocket closed"); + }; + const backend: WorkspaceBackend = { + id: "only", + type: "fake", + async connect(): Promise { + const sync = connects++ === 0 ? stale : replacement; return { - currentCursor: { rev: 0, path: null }, - appliedPushCursor: { rev: 0, path: null }, - stream: new ReadableStream({ - start(c) { - c.close(); - }, - }), + rpc: composite(sync), + closed: new Promise(() => {}), + close: async () => { + closes++; + }, }; }, - async watermarks() { - return { currentRev: 0, pushRev: 0, fetchCursor: { rev: 0, path: null } }; + }; + const ws = new Workspace({ storage: makeStorage(), backends: [backend] }); + await ws.fs.writeFile("/a.txt", "hi"); + + await expect(ws.push()).resolves.toBeGreaterThan(0); + expect(connects).toBe(2); + expect(closes).toBe(1); + await expect(replacement.readEntry("/a.txt")).resolves.toMatchObject({ + kind: "file", + path: "/a.txt", + }); + }); + + it("waits for stale handle close before sharing the reconnect", async () => { + let connects = 0; + let releaseClose: (() => void) | undefined; + let closeStarted: (() => void) | undefined; + const closeBegan = new Promise((resolve) => { + closeStarted = resolve; + }); + const stale: import("@cloudflare/computer-rpc").SyncRPC = { + ...fakeRpc(), + async push() { + throw new WorkspaceTransportError("WebSocket closed"); }, }; const backend: WorkspaceBackend = { - id, + id: "only", type: "fake", async connect(): Promise { - onConnect(); - // closed promise stays pending forever — simulating a - // wedged transport that never produces a clean signal. + const sync = connects++ === 0 ? stale : fakeRpc(); + return { + rpc: composite(sync), + close: async () => { + closeStarted?.(); + await new Promise((resolve) => { + releaseClose = resolve; + }); + }, + }; + }, + }; + const ws = new Workspace({ storage: makeStorage(), backends: [backend] }); + await ws.fs.writeFile("/a.txt", "hi"); + + const push = ws.push(); + await closeBegan; + const ready = ws.ready("only"); + await Promise.resolve(); + expect(connects).toBe(1); + + releaseClose?.(); + await Promise.all([push, ready]); + expect(connects).toBe(2); + }); + + it("pull() reconnects and succeeds within the same call", async () => { + let connects = 0; + let closes = 0; + const stale: import("@cloudflare/computer-rpc").SyncRPC = { + ...fakeRpc(), + async fetchChanges() { + throw new WorkspaceTransportError("RPC session was shut down"); + }, + }; + const backend: WorkspaceBackend = { + id: "only", + type: "fake", + async connect(): Promise { + const sync = connects++ === 0 ? stale : fakeRpc(); return { rpc: composite(sync), closed: new Promise(() => {}), - close: async () => {}, + close: async () => { + closes++; + }, }; }, }; - return { - backend, - failNext: () => { - shouldFail = true; + const ws = new Workspace({ storage: makeStorage(), backends: [backend] }); + + await expect(ws.pull()).resolves.toEqual({ applied: 0, skipped: [] }); + expect(connects).toBe(2); + expect(closes).toBe(1); + }); + + it("retries a readiness failure before dispatching the sync operation", async () => { + let connects = 0; + const backend: WorkspaceBackend = { + id: "only", + type: "fake", + async connect(): Promise { + connects++; + if (connects === 1) { + throw new WorkspaceTransportError( + "CloudflareContainerBackend(only): connect failed at stage=health", + ); + } + return { rpc: composite(fakeRpc()), close: async () => {} }; }, }; - } + const ws = new Workspace({ storage: makeStorage(), backends: [backend] }); + await ws.fs.writeFile("/ready.txt", "ready"); - it("push() invalidates the cached handle on a transport error", async () => { + await expect(ws.push()).resolves.toBeGreaterThan(0); + expect(connects).toBe(2); + }); + + it("closes an unpublished handle when watermark reconciliation fails", async () => { let connects = 0; - const { backend, failNext } = transportFailingBackend("only", () => { - connects++; - }); + let closes = 0; + const broken: import("@cloudflare/computer-rpc").SyncRPC = { + ...fakeRpc(), + async watermarks() { + throw new WorkspaceTransportError("WebSocket closed during reconcile"); + }, + }; + const backend: WorkspaceBackend = { + id: "only", + type: "fake", + async connect(): Promise { + const sync = connects++ === 0 ? broken : fakeRpc(); + return { + rpc: composite(sync), + close: async () => { + closes++; + }, + }; + }, + }; const ws = new Workspace({ storage: makeStorage(), backends: [backend] }); - await ws.ready("only"); - expect(connects).toBe(1); - await ws.fs.writeFile("/a.txt", "hi"); - failNext(); - await expect(ws.push()).rejects.toThrow(/WebSocket closed/); - - // Next operation must reconnect — the bad handle is gone. - // Reset the fail flag so the second connect's RPC works. - // (failNext flipped a flag on the closure; re-binding is fine - // because connect() returns a fresh handle that reads it.) - // For this assertion we just need a fresh connect attempt. - await ws.push().catch(() => undefined); + await expect(ws.pull()).resolves.toEqual({ applied: 0, skipped: [] }); expect(connects).toBe(2); + expect(closes).toBe(1); }); - it("pull() invalidates the cached handle on a transport error", async () => { + it("preserves the terminal cause when reconnect retry is exhausted", async () => { let connects = 0; - const { backend, failNext } = transportFailingBackend("only", () => { - connects++; - }); + const errors = [ + new WorkspaceTransportError("first session closed"), + new WorkspaceTransportError("replacement was not ready"), + ]; + const backend: WorkspaceBackend = { + id: "only", + type: "fake", + async connect(): Promise { + throw errors[connects++]; + }, + }; const ws = new Workspace({ storage: makeStorage(), backends: [backend] }); - await ws.ready("only"); - expect(connects).toBe(1); - failNext(); - await expect(ws.pull()).rejects.toThrow(/WebSocket closed/); - await ws.pull().catch(() => undefined); + + const error = await ws.pull().catch((caught: unknown) => caught); + expect(error).toBeInstanceOf(WorkspaceTransportError); + expect(error).toMatchObject({ cause: errors[1] }); + expect(String(error)).toMatch(/pull failed after 1 reconnect retry/); + expect(String(error)).toMatch(/first session closed/); + expect(String(error)).toMatch(/replacement was not ready/); expect(connects).toBe(2); }); @@ -1293,11 +1379,82 @@ describe("Workspace transport-failure invalidation", () => { expect(connects).toBe(1); }); - it("shell.exec invalidates the cached handle on a transport error", async () => { + it("uses the replacement shell after a pre-exec push reconnects", async () => { let connects = 0; + let staleShellCalls = 0; + let replacementShellCalls = 0; + const staleSync: import("@cloudflare/computer-rpc").SyncRPC = { + ...fakeRpc(), + async push() { + throw new WorkspaceTransportError("WebSocket closed during pre-exec push"); + }, + }; + const replacementBase = fakeRpc(); + const replacementSync: import("@cloudflare/computer-rpc").SyncRPC = { + ...replacementBase, + async fetchChanges(input) { + const result = await replacementBase.fetchChanges(input); + return { + ...result, + appliedPushCursor: { rev: Number.MAX_SAFE_INTEGER, path: null }, + }; + }, + }; + const backend: WorkspaceBackend = { + id: "only", + type: "fake", + async connect(): Promise { + const generation = connects++; + const shell: import("@cloudflare/computer-rpc").ShellRPC = { + async exec() { + if (generation === 0) { + staleShellCalls++; + throw new Error("stale shell used after reconnect"); + } + replacementShellCalls++; + return { + id: "replacement-run", + events: new ReadableStream({ + start(controller) { + controller.enqueue({ id: "replacement-run", seq: 1, name: "exit", code: 0 }); + controller.close(); + }, + }), + }; + }, + getExec: () => Promise.reject(new Error("not used")), + killExec: () => Promise.reject(new Error("not used")), + disposeExec: () => Promise.reject(new Error("not used")), + }; + return { + rpc: { sync: generation === 0 ? staleSync : replacementSync, shell }, + close: async () => {}, + }; + }, + }; + const ws = new Workspace({ storage: makeStorage(), backends: [backend] }); + await ws.fs.writeFile("/input.txt", "ready"); + + const result = await (await ws.runtime.exec("true")).result(); + expect(result.exitCode).toBe(0); + expect(connects).toBe(2); + expect(staleShellCalls).toBe(0); + expect(replacementShellCalls).toBe(1); + }); + + it("does not spawn when pre-exec push exhausts its reconnect retry", async () => { + let connects = 0; + let shellCalls = 0; + const sync: import("@cloudflare/computer-rpc").SyncRPC = { + ...fakeRpc(), + async push() { + throw new WorkspaceTransportError("container unavailable"); + }, + }; const shell: import("@cloudflare/computer-rpc").ShellRPC = { async exec() { - throw new WorkspaceTransportError("RPC session was shut down"); + shellCalls++; + throw new Error("must not run"); }, getExec: () => Promise.reject(new Error("not used")), killExec: () => Promise.reject(new Error("not used")), @@ -1308,19 +1465,526 @@ describe("Workspace transport-failure invalidation", () => { type: "fake", async connect(): Promise { connects++; + return { + rpc: { sync, shell }, + close: async () => {}, + }; + }, + }; + const ws = new Workspace({ storage: makeStorage(), backends: [backend] }); + await ws.fs.writeFile("/input.txt", "required"); + + await expect(ws.runtime.exec("side-effect")).rejects.toThrow( + /shell\.exec failed after 1 reconnect retry.*pre-exec push failed/, + ); + expect(connects).toBe(2); + expect(shellCalls).toBe(0); + }); + + it("pushes again when the connection closes between push and exec", async () => { + let connects = 0; + let closeFirst: (() => void) | undefined; + const firstClosed = new Promise((resolve) => { + closeFirst = resolve; + }); + const backend: WorkspaceBackend = { + id: "only", + type: "fake", + async connect(): Promise { + const generation = connects++; + const baseSync = fakeRpc(); + const sync: import("@cloudflare/computer-rpc").SyncRPC = + generation === 0 + ? { + ...baseSync, + async push(input) { + const result = await baseSync.push(input); + closeFirst?.(); + return result; + }, + } + : baseSync; + const shell: import("@cloudflare/computer-rpc").ShellRPC = { + async exec() { + if (generation === 0) { + throw new Error("Attempted to use RPC stub after it has been disposed."); + } + if ((await sync.readEntry("/required.txt")) === null) { + throw new Error("new connection did not receive the pre-exec push"); + } + return { + id: "replacement-run", + events: new ReadableStream({ + start(controller) { + controller.enqueue({ id: "replacement-run", seq: 1, name: "exit", code: 0 }); + controller.close(); + }, + }), + }; + }, + getExec: () => Promise.reject(new Error("not used")), + killExec: () => Promise.reject(new Error("not used")), + disposeExec: () => Promise.reject(new Error("not used")), + }; + return { + rpc: { sync, shell }, + closed: generation === 0 ? firstClosed : undefined, + close: async () => {}, + }; + }, + }; + const ws = new Workspace({ storage: makeStorage(), backends: [backend] }); + await ws.fs.writeFile("/required.txt", "present"); + + const result = await (await ws.runtime.exec("true")).result(); + expect(result.exitCode).toBe(0); + expect(connects).toBe(2); + }); + + it("retries shell.exec when a disposed stub proves dispatch did not start", async () => { + let connects = 0; + let closes = 0; + let starts = 0; + let disposals = 0; + const backend: WorkspaceBackend = { + id: "only", + type: "fake", + async connect(): Promise { + const generation = connects++; + const sync = fakeRpc(); + const shell: import("@cloudflare/computer-rpc").ShellRPC = { + async exec() { + if (generation === 0) { + throw new Error("Attempted to use RPC stub after it has been disposed."); + } + if ((await sync.readEntry("/required.txt")) === null) { + throw new Error("replacement container was not pushed before exec"); + } + starts++; + return Object.assign( + { + id: "fresh-run", + events: new ReadableStream({ + start(controller) { + controller.enqueue({ id: "fresh-run", seq: 1, name: "exit", code: 0 }); + controller.close(); + }, + }), + }, + { + [Symbol.dispose]() { + disposals++; + }, + }, + ); + }, + getExec: () => Promise.reject(new Error("not used")), + killExec: () => Promise.reject(new Error("not used")), + disposeExec: () => Promise.reject(new Error("not used")), + }; + return { + rpc: { sync, shell }, + close: async () => { + closes++; + }, + }; + }, + }; + const ws = new Workspace({ storage: makeStorage(), backends: [backend] }); + await ws.fs.writeFile("/required.txt", "present"); + + const result = await (await ws.runtime.exec("true")).result(); + expect(result.exitCode).toBe(0); + expect(connects).toBe(2); + expect(closes).toBe(1); + expect(starts).toBe(1); + expect(disposals).toBe(1); + }); + + it("does not replay shell.exec after an ambiguous transport failure", async () => { + let connects = 0; + let closes = 0; + let dispatches = 0; + const backend: WorkspaceBackend = { + id: "only", + type: "fake", + async connect(): Promise { + const generation = connects++; + const shell: import("@cloudflare/computer-rpc").ShellRPC = { + async exec() { + dispatches++; + if (generation === 0) { + throw new WorkspaceTransportError("RPC session was shut down"); + } + return { + id: "second-request", + events: new ReadableStream({ + start(controller) { + controller.enqueue({ id: "second-request", seq: 1, name: "exit", code: 0 }); + controller.close(); + }, + }), + }; + }, + getExec: () => Promise.reject(new Error("not used")), + killExec: () => Promise.reject(new Error("not used")), + disposeExec: () => Promise.reject(new Error("not used")), + }; return { rpc: { sync: fakeRpc(), shell }, sync: "none", - closed: new Promise(() => {}), - close: async () => {}, + close: async () => { + closes++; + }, }; }, }; const ws = new Workspace({ storage: makeStorage(), backends: [backend] }); - await expect(ws.runtime.exec("true")).rejects.toThrow(/RPC session was shut down/); + + const error = await ws.runtime.exec("side-effect").catch((caught: unknown) => caught); + expect(String(error)).toMatch(/may have started/); + expect(String(error)).toMatch(/was not replayed/); expect(connects).toBe(1); - await ws.runtime.exec("true").catch(() => undefined); + expect(dispatches).toBe(1); + expect(closes).toBe(1); + + const result = await (await ws.runtime.exec("second request")).result(); + expect(result.exitCode).toBe(0); + expect(connects).toBe(2); + expect(dispatches).toBe(2); + }); + + for (const operation of ["getExec", "killExec", "disposeExec"] as const) { + it(`${operation} reconnects and retries a transport failure`, async () => { + let connects = 0; + let closes = 0; + let calls = 0; + const backend: WorkspaceBackend = { + id: "only", + type: "fake", + async connect(): Promise { + const generation = connects++; + const failFirst = () => { + calls++; + if (generation === 0) throw new WorkspaceTransportError("WebSocket closed"); + }; + const shell: import("@cloudflare/computer-rpc").ShellRPC = { + exec: () => Promise.reject(new Error("not used")), + async getExec() { + failFirst(); + return { + id: "run", + events: new ReadableStream({ + start(controller) { + controller.enqueue({ id: "run", seq: 1, name: "exit", code: 0 }); + controller.close(); + }, + }), + }; + }, + async killExec() { + failFirst(); + }, + async disposeExec() { + failFirst(); + }, + }; + return { + rpc: { sync: fakeRpc(), shell }, + sync: "none", + close: async () => { + closes++; + }, + }; + }, + }; + const ws = new Workspace({ storage: makeStorage(), backends: [backend] }); + + if (operation === "getExec") { + await (await ws.runtime.getExec("run")).result(); + } else if (operation === "killExec") { + await ws.runtime.killExec("run"); + } else { + await ws.runtime.disposeExec("run"); + } + + expect(connects).toBe(2); + expect(closes).toBe(1); + expect(calls).toBe(2); + }); + } + + it("retries lifecycle operations when reconnect reaches the same runtime", async () => { + let connects = 0; + let killCalls = 0; + const backend: WorkspaceBackend = { + id: "only", + type: "fake", + async connect(): Promise { + const connection = connects++; + const shell: import("@cloudflare/computer-rpc").ShellRPC = { + async exec(input) { + const id = input.id ?? "run"; + return { + id, + events: new ReadableStream({ + start(controller) { + controller.enqueue({ id, seq: 1, name: "exit", code: 0 }); + controller.close(); + }, + }), + }; + }, + getExec: () => Promise.reject(new Error("not used")), + async killExec() { + killCalls++; + if (connection === 0) throw new WorkspaceTransportError("WebSocket closed"); + }, + disposeExec: () => Promise.reject(new Error("not used")), + }; + return { + rpc: { sync: fakeRpc(), shell }, + sync: "none", + runtimeId: "runtime-a", + close: async () => {}, + }; + }, + }; + const ws = new Workspace({ storage: makeStorage(), backends: [backend] }); + const execution = await ws.runtime.exec("true", { id: "run" }); + await execution.result(); + + await expect(execution.kill()).resolves.toBeUndefined(); expect(connects).toBe(2); + expect(killCalls).toBe(2); + }); + + for (const operation of ["getExec", "killExec", "disposeExec"] as const) { + it(`${operation} rejects when reconnect reaches a replacement runtime`, async () => { + let connects = 0; + let replacementCalls = 0; + const backend: WorkspaceBackend = { + id: "only", + type: "fake", + async connect(): Promise { + const connection = connects++; + const runtimeId = connection === 0 ? "runtime-a" : "runtime-b"; + const failOrCountReplacement = () => { + if (connection === 0) throw new WorkspaceTransportError("WebSocket closed"); + replacementCalls++; + }; + const shell: import("@cloudflare/computer-rpc").ShellRPC = { + async exec(input) { + const id = input.id ?? "run"; + return { + id, + events: new ReadableStream({ + start(controller) { + controller.enqueue({ id, seq: 1, name: "exit", code: 0 }); + controller.close(); + }, + }), + }; + }, + async getExec() { + failOrCountReplacement(); + return { + id: "run", + events: new ReadableStream({ + start(controller) { + controller.enqueue({ id: "run", seq: 1, name: "exit", code: 0 }); + controller.close(); + }, + }), + }; + }, + async killExec() { + failOrCountReplacement(); + }, + async disposeExec() { + failOrCountReplacement(); + }, + }; + return { + rpc: { sync: fakeRpc(), shell }, + sync: "none", + runtimeId, + close: async () => {}, + }; + }, + }; + const ws = new Workspace({ storage: makeStorage(), backends: [backend] }); + const execution = await ws.runtime.exec("true", { id: "run" }); + await execution.result(); + + let result: Promise; + if (operation === "getExec") { + result = ws.runtime.getExec("run"); + } else if (operation === "killExec") { + result = execution.kill(); + } else { + result = ws.runtime.disposeExec("run"); + } + + await expect(result).rejects.toMatchObject({ code: "EEXEC_LOST" }); + expect(connects).toBe(2); + expect(replacementCalls).toBe(0); + }); + } + + it("an old execution handle cannot target a reused id in a replacement runtime", async () => { + let connects = 0; + let closeFirst: (() => void) | undefined; + let replacementKills = 0; + const firstClosed = new Promise((resolve) => { + closeFirst = resolve; + }); + const backend: WorkspaceBackend = { + id: "only", + type: "fake", + async connect(): Promise { + const connection = connects++; + const runtimeId = connection === 0 ? "runtime-a" : "runtime-b"; + const shell: import("@cloudflare/computer-rpc").ShellRPC = { + async exec(input) { + const id = input.id ?? "reused"; + return { + id, + events: new ReadableStream({ + start(controller) { + controller.enqueue({ id, seq: 1, name: "exit", code: 0 }); + controller.close(); + }, + }), + }; + }, + getExec: () => Promise.reject(new Error("not used")), + async killExec() { + replacementKills++; + }, + async disposeExec() {}, + }; + return { + rpc: { sync: fakeRpc(), shell }, + sync: "none", + runtimeId, + closed: connection === 0 ? firstClosed : undefined, + close: async () => {}, + }; + }, + }; + const ws = new Workspace({ storage: makeStorage(), backends: [backend] }); + const oldExecution = await ws.runtime.exec("old", { id: "reused" }); + await oldExecution.result(); + + closeFirst?.(); + await Promise.resolve(); + const replacement = await ws.runtime.exec("new", { id: "reused" }); + await replacement.result(); + + await expect(oldExecution.kill()).rejects.toMatchObject({ code: "EEXEC_LOST" }); + expect(replacementKills).toBe(0); + }); + + it("persists execution runtime fencing across Workspace incarnations", async () => { + const storage = makeStorage(); + const firstBackend: WorkspaceBackend = { + id: "only", + type: "fake", + async connect(): Promise { + return { + rpc: { + sync: fakeRpc(), + shell: { + async exec(input) { + const id = input.id ?? "persisted"; + return { + id, + events: new ReadableStream({ + start(controller) { + controller.enqueue({ id, seq: 1, name: "exit", code: 0 }); + controller.close(); + }, + }), + }; + }, + getExec: () => Promise.reject(new Error("not used")), + killExec: async () => {}, + disposeExec: async () => {}, + }, + }, + sync: "none", + runtimeId: "runtime-a", + close: async () => {}, + }; + }, + }; + const first = new Workspace({ storage, backends: [firstBackend] }); + const execution = await first.runtime.exec("true", { id: "persisted" }); + await execution.result(); + await first.close(); + + let replacementKills = 0; + const replacementBackend: WorkspaceBackend = { + id: "only", + type: "fake", + async connect(): Promise { + return { + rpc: { + sync: fakeRpc(), + shell: { + exec: () => Promise.reject(new Error("not used")), + getExec: () => Promise.reject(new Error("not used")), + async killExec() { + replacementKills++; + }, + disposeExec: async () => {}, + }, + }, + sync: "none", + runtimeId: "runtime-b", + close: async () => {}, + }; + }, + }; + const recreated = new Workspace({ storage, backends: [replacementBackend] }); + + await expect(recreated.runtime.killExec("persisted")).rejects.toMatchObject({ + code: "EEXEC_LOST", + }); + expect(replacementKills).toBe(0); + }); + + it("does not reconnect shell.exec for a non-transport RPC error", async () => { + let connects = 0; + let calls = 0; + const shell: import("@cloudflare/computer-rpc").ShellRPC = { + async exec() { + calls++; + throw new Error("EEXEC_BUSY: id is already running"); + }, + getExec: () => Promise.reject(new Error("not used")), + killExec: () => Promise.reject(new Error("not used")), + disposeExec: () => Promise.reject(new Error("not used")), + }; + const backend: WorkspaceBackend = { + id: "only", + type: "fake", + async connect(): Promise { + connects++; + return { + rpc: { sync: fakeRpc(), shell }, + sync: "none", + close: async () => {}, + }; + }, + }; + const ws = new Workspace({ storage: makeStorage(), backends: [backend] }); + + await expect(ws.runtime.exec("true")).rejects.toThrow(/EEXEC_BUSY/); + await expect(ws.runtime.exec("true")).rejects.toThrow(/EEXEC_BUSY/); + expect(connects).toBe(1); + expect(calls).toBe(2); }); it("shell.exec invalidates the cached handle on a mid-stream transport error", async () => { diff --git a/packages/computer/src/workspace.ts b/packages/computer/src/workspace.ts index ac4b9e90..46dd9187 100644 --- a/packages/computer/src/workspace.ts +++ b/packages/computer/src/workspace.ts @@ -9,6 +9,7 @@ // Command-backend pre-exec push / post-exec pull brackets are // routed through Workspace.runtime.exec. +import type { ShellRPC } from "@cloudflare/computer-rpc"; import { pullOnce, pushOnce, reconcileWatermarks } from "@cloudflare/computer-rpc/driver"; import { type ApplyResult, @@ -27,6 +28,7 @@ import { } from "./artifacts/index.js"; import type { AssetsClient } from "./assets/index.js"; import type { BackendHandle, WorkspaceBackend } from "./backend.js"; +import { ExecutionRuntimeTracker, SqlExecutionRuntimeStore } from "./execution-runtime-tracker.js"; import type { GitClient, GitClientFactory, GitIdentity } from "./git/index.js"; import { MountIndex } from "./mounts/index.js"; import { buildMountRegistry, type MountValue } from "./mounts/registry.js"; @@ -40,12 +42,20 @@ import { type WorkspaceRegisteredBackend, type WorkspaceRuntimeEvent, } from "./runtime/types.js"; -import { CommandExecutor } from "./shell.js"; +import { CommandExecutor, maybeDispose, spawnShell } from "./shell.js"; import { WorkspaceStub } from "./stub.js"; -import { isWorkspaceTransportFailure } from "./transport-failure.js"; +import { + isWorkspacePreDispatchTransportFailure, + isWorkspaceTransportFailure, + WorkspacePreDispatchTransportError, + WorkspaceTransportError, +} from "./transport-failure.js"; export interface SyncRetryIntent { backend: string; + // Container process whose post-command changes are pending. Durable + // retries must not report success against an empty replacement. + runtimeId?: string; attempt: number; notBefore: number; } @@ -71,13 +81,33 @@ export interface SyncRetryOptions { export type WorkspaceRetryPendingSyncResult = | { status: "idle"; backend: string } | { status: "complete"; backend: string; applied: number; skipped: ApplyResult["skipped"] } - | { status: "pending"; backend: string; attempt: number; notBefore: number; error: string } - | { status: "exhausted"; backend: string; attempt: number; error: string }; + | { + status: "pending"; + backend: string; + runtimeId?: string; + attempt: number; + notBefore: number; + error: string; + } + | { + status: "exhausted"; + backend: string; + runtimeId?: string; + attempt: number; + error: string; + } + | { status: "lost"; backend: string; runtimeId: string; error: string }; const DEFAULT_RETRY_INITIAL_DELAY_MS = 1_000; const DEFAULT_RETRY_MAX_DELAY_MS = 60_000; const DEFAULT_RETRY_MAX_ATTEMPTS = 5; +// When a backend RPC fails with a transport error, how much replay +// the operation tolerates. "always" suits idempotent calls; a +// "pre-dispatch" operation is replayed only when the failure proves +// no frame reached the peer. +type BackendRetryPolicy = "always" | "pre-dispatch"; + export interface WorkspaceOptions { // Local store backing this Workspace. In a Durable Object, pass // `ctx.storage`; in tests, pass a SQLiteTestStorage from @@ -237,6 +267,10 @@ export class Workspace { // In-flight connect promises keyed by backend id, so concurrent // callers for the same backend share one connect pass. readonly #connecting = new Map>(); + // A transport-failed handle must finish closing before connect() + // runs again. This prevents a concurrent caller from reaching a + // backend's own cache while it still points at the stale handle. + readonly #disconnecting = new Map>(); // Per-backend CommandExecutor facades. Constructed alongside each // handle; reused for the life of the handle. readonly #shells = new Map(); @@ -244,6 +278,10 @@ export class Workspace { // unified backend handle. Cleared alongside #shells so an adapter // never outlives the shell it wraps. readonly #commandHandles = new Map(); + // Last known container runtime for recent backend/execution ids. + // Returned handles carry their own id; the bounded LRU supports + // direct by-id lifecycle calls without growing for the DO lifetime. + readonly #executionRuntimes: ExecutionRuntimeTracker; readonly #moduleHandles = new Map(); readonly #connectingModuleHandles = new Map>(); #connectionGeneration = 0; @@ -297,6 +335,10 @@ export class Workspace { : createDisabledArtifactsClient(); this.#db = new Database(options.storage); initializeSchema(this.#db, this.#now); + this.#executionRuntimes = new ExecutionRuntimeTracker( + 1_024, + new SqlExecutionRuntimeStore(this.#db), + ); this.#fs = new WorkspaceFilesystem(this.#db, { now: this.#now }); const registered = (options.backends ?? []).slice(); this.#backends = registered.filter( @@ -550,16 +592,15 @@ export class Workspace { { "workspace.sync.backend": resolvedId }, async () => { if (resolvedId === undefined || this.#moduleBackendsById.has(resolvedId)) return 0; - const handle = await this.#handleFor(resolvedId); - // A backend that reuses the host store as its sole - // source of truth has nothing to ship and no remote to - // ship to. Short-circuit so the shell exec bracket can - // keep calling push() unconditionally without paying - // for it. - if (handle.sync === "none") return 0; - return this.#runWithInvalidation(resolvedId, handle, () => - pushOnce(this.#db, handle.rpc.sync, resolvedId), - ); + return this.#runWithReconnect(resolvedId, "push", async (handle) => { + // A backend that reuses the host store as its sole + // source of truth has nothing to ship and no remote to + // ship to. Short-circuit so the shell exec bracket can + // keep calling push() unconditionally without paying + // for it. + if (handle.sync === "none") return 0; + return pushOnce(this.#db, handle.rpc.sync, resolvedId); + }); }, (span, outcome) => { if (outcome.ok) span.setAttribute("workspace.sync.pushed", outcome.value); @@ -595,12 +636,13 @@ export class Workspace { return { status: "exhausted", backend: resolvedId, + ...(intent.runtimeId === undefined ? {} : { runtimeId: intent.runtimeId }), attempt: intent.attempt, error: "pending sync retry attempts exhausted", }; } try { - const result = await this.#pullResolved(resolvedId); + const result = await this.#pullResolved(resolvedId, intent.runtimeId); await scheduler.clear(resolvedId); return { status: "complete", @@ -610,22 +652,35 @@ export class Workspace { }; } catch (error) { const message = safeErrorMessage(error); + if ( + intent.runtimeId !== undefined && + (error as { code?: unknown } | null)?.code === "EEXEC_LOST" + ) { + await scheduler.clear(resolvedId); + return { + status: "lost", + backend: resolvedId, + runtimeId: intent.runtimeId, + error: message, + }; + } if (intent.attempt >= this.#retryMaxAttempts) { return { status: "exhausted", backend: resolvedId, + ...(intent.runtimeId === undefined ? {} : { runtimeId: intent.runtimeId }), attempt: intent.attempt, error: message, }; } - const next = this.#retryIntent(resolvedId, intent.attempt + 1); + const next = this.#retryIntent(resolvedId, intent.attempt + 1, intent.runtimeId); await scheduler.schedule(next); return { status: "pending", ...next, error: message }; } }); } - #pullResolved(resolvedId: string | undefined): Promise { + #pullResolved(resolvedId: string | undefined, expectedRuntimeId?: string): Promise { return withSpan( this.#observer, "workspace.sync.pull", @@ -634,11 +689,13 @@ export class Workspace { if (resolvedId === undefined || this.#moduleBackendsById.has(resolvedId)) { return { applied: 0, skipped: [] }; } - const handle = await this.#handleFor(resolvedId); - if (handle.sync === "none") return { applied: 0, skipped: [] }; - return this.#runWithInvalidation(resolvedId, handle, () => - pullOnce(this.#db, handle.rpc.sync, resolvedId), - ); + return this.#runWithReconnect(resolvedId, "pull", async (handle) => { + if (expectedRuntimeId !== undefined) { + assertExecutionRuntime("post-command sync", expectedRuntimeId, handle.runtimeId); + } + if (handle.sync === "none") return { applied: 0, skipped: [] }; + return pullOnce(this.#db, handle.rpc.sync, resolvedId); + }); }, (span, outcome) => { if (!outcome.ok) return; @@ -648,52 +705,112 @@ export class Workspace { ); } - async #schedulePendingSync(id: string): Promise { + async #schedulePendingSync(id: string, runtimeId?: string): Promise { const scheduler = this.#retryScheduler; if (scheduler === undefined) return; await this.#serialize(id, async (resolvedId) => { - if (resolvedId === undefined || (await scheduler.get(resolvedId)) !== undefined) return; - await scheduler.schedule(this.#retryIntent(resolvedId, 1)); + if (resolvedId === undefined) return; + const existing = await scheduler.get(resolvedId); + if (existing !== undefined && (runtimeId === undefined || existing.runtimeId === runtimeId)) { + return; + } + await scheduler.schedule(this.#retryIntent(resolvedId, 1, runtimeId)); }); } - #retryIntent(backend: string, attempt: number): SyncRetryIntent { + #retryIntent(backend: string, attempt: number, runtimeId?: string): SyncRetryIntent { const delay = Math.min( this.#retryMaxDelayMs, this.#retryInitialDelayMs * 2 ** Math.max(0, attempt - 1), ); - return { backend, attempt, notBefore: this.#now() + delay }; + return { + backend, + ...(runtimeId === undefined ? {} : { runtimeId }), + attempt, + notBefore: this.#now() + delay, + }; } - // Drop a cached handle when an operation fails with a known - // transport-level error. Matches by identity — a concurrent - // close() / `closed` watcher that already swapped the entry must - // not be clobbered. Returns true if the cached entry was the one - // we removed. - #invalidateHandle(id: string, handle: BackendHandle): boolean { + // Drop and close a cached handle after a transport failure. + // Matches by identity so a late error from an old operation cannot + // tear down a replacement that another caller already installed. + // Cache deletion is synchronous; close() is awaited before retrying + // so backends with their own handle cache cannot return the same + // broken session from connect(). + async #invalidateHandle(id: string, handle: BackendHandle): Promise { if (this.#handles.get(id) !== handle) return false; this.#handles.delete(id); this.#shells.delete(id); this.#commandHandles.delete(id); + const closing = handle.close().catch(() => undefined); + this.#disconnecting.set(id, closing); + try { + await closing; + } finally { + if (this.#disconnecting.get(id) === closing) this.#disconnecting.delete(id); + } return true; } - // Wrap an RPC-backed operation so a transport failure invalidates - // the cached handle before the error rethrows. Non-transport - // errors pass through untouched. - async #runWithInvalidation( + // Run a backend operation with one reconnect retry. Handle + // acquisition is inside the loop so a readiness failure and an RPC + // failure follow the same bounded policy. The default "always" + // policy assumes an idempotent operation: pushOnce and pullOnce are + // safe replay boundaries because their durable watermarks advance + // only after committed work and their apply paths absorb duplicates. + async #runWithReconnect( id: string, - handle: BackendHandle, - op: () => Promise, + operation: string, + op: (handle: BackendHandle) => Promise, + policy: BackendRetryPolicy = "always", ): Promise { - try { - return await op(); - } catch (error) { - if (isWorkspaceTransportFailure(error)) { - this.#invalidateHandle(id, handle); + let firstError: unknown; + for (let attempt = 0; attempt < 2; attempt++) { + let handle: BackendHandle | undefined; + try { + handle = await this.#handleFor(id); + return await op(handle); + } catch (error) { + if (!isWorkspaceTransportFailure(error)) { + if (firstError !== undefined && (error as { code?: unknown })?.code === "EEXEC_LOST") { + throw Object.assign( + new Error( + `${safeErrorMessage(error)} Original transport failure: ${safeErrorMessage(firstError)}`, + { cause: firstError }, + ), + { name: "WorkspaceExecutionLostError", code: "EEXEC_LOST" }, + ); + } + throw error; + } + if (handle !== undefined) await this.#invalidateHandle(id, handle); + + // A failed connection is always before dispatch. Once an + // operation reached a handle, only an "always" policy or a + // local disposed-stub failure makes replay safe. + const canRetry = + handle === undefined || + policy === "always" || + isWorkspacePreDispatchTransportFailure(error); + if (!canRetry) { + throw new WorkspaceTransportError( + `Workspace backend ${JSON.stringify(id)}: ${operation} transport failed; the command may have started and was not replayed: ${safeErrorMessage(error).slice(0, 240)}`, + { cause: error }, + ); + } + if (attempt === 0) { + firstError = error; + continue; + } + const initial = safeErrorMessage(firstError).slice(0, 160); + const terminal = safeErrorMessage(error).slice(0, 240); + throw new WorkspaceTransportError( + `Workspace backend ${JSON.stringify(id)}: ${operation} failed after 1 reconnect retry: initial=${initial}; last=${terminal}`, + { cause: error }, + ); } - throw error; } + throw new Error("unreachable reconnect state"); } // Per-backend mutation FIFO. Public push() / pull() calls and each @@ -756,11 +873,13 @@ export class Workspace { this.#shells.clear(); this.#commandHandles.clear(); this.#connecting.clear(); + const disconnecting = [...this.#disconnecting.values()]; this.#moduleHandles.clear(); this.#connectingModuleHandles.clear(); this.#readyPromise = undefined; - await Promise.all( - [...handles, ...moduleHandles].map(async (h) => { + await Promise.all([ + ...disconnecting, + ...[...handles, ...moduleHandles].map(async (h) => { try { await h.close?.(); } catch { @@ -768,7 +887,7 @@ export class Workspace { // gone shouldn't take the workspace down with it. } }), - ); + ]); } // Unified backend handle used by the runtime. Module backends @@ -780,72 +899,64 @@ export class Workspace { return this.#commandHandleFor(id); } - // Command adapters are cached per backend so the module and - // command paths are symmetric. The cache is cleared alongside - // #shells whenever a handle is invalidated, so an adapter never - // outlives the shell it closed over. + // Command adapters are cached per backend, but the CommandExecutor + // they contain resolves the current ShellRPC for every operation. + // A pre-exec push can replace the connection, so binding the adapter + // to the handle that existed before that push would dispatch the + // command on a stale session. async #commandHandleFor(id: string): Promise { const cached = this.#commandHandles.get(id); if (cached) return cached; - const { shell, handle } = await this.#shellFor(id); - const onError = (error: unknown) => this.#onShellError(id, handle, error); + const shell = this.#shellFor(id); const adapter: WorkspaceModuleBackendHandle = { exec: async (input) => { - let envelope: Awaited>; - try { - envelope = await shell.exec(input.source, { - id: input.id, - cwd: input.cwd, - timeoutMs: input.timeoutMs, - env: input.env, - stdin: input.stdin, - }); - } catch (error) { - onError(error); - throw error; - } + const envelope = await shell.exec(input.source, { + id: input.id, + cwd: input.cwd, + timeoutMs: input.timeoutMs, + env: input.env, + stdin: input.stdin, + }); + this.#rememberExecutionRuntime(id, envelope.id, envelope.runtimeId); return { id: envelope.id, - events: watchStreamForTransportError( - envelope.events, - onError, - ) as ReadableStream, + runtimeId: envelope.runtimeId, + events: envelope.events as ReadableStream, sync: envelope.sync, }; }, - getExec: async ({ id: execId, after }) => { + getExec: async ({ id: execId, after, runtimeId }) => { const resume = after === undefined ? "full" : after; - let envelope: Awaited>; - try { - envelope = await shell.get(execId, { resume }); - } catch (error) { - onError(error); - throw error; - } + const expectedRuntimeId = + runtimeId ?? this.#executionRuntimes.get(this.#executionRuntimeKey(id, execId)); + const envelope = await shell.get(execId, { resume, runtimeId: expectedRuntimeId }); + this.#rememberExecutionRuntime(id, envelope.id, envelope.runtimeId); return { id: envelope.id, - events: watchStreamForTransportError( - envelope.events, - onError, - ) as ReadableStream, + runtimeId: envelope.runtimeId, + events: envelope.events as ReadableStream, sync: envelope.sync, }; }, - killExec: async ({ id: execId, signal }) => { - try { - await shell.kill(execId, signal); - } catch (error) { - onError(error); - throw error; - } - }, - disposeExec: async ({ id: execId }) => { - try { - await shell.dispose(execId); - } catch (error) { - onError(error); - throw error; - } + killExec: ({ id: execId, signal, runtimeId }) => + this.#runExecutionOperation( + id, + "shell.killExec", + execId, + runtimeId ?? this.#executionRuntimes.get(this.#executionRuntimeKey(id, execId)), + (shellRpc) => shellRpc.killExec({ id: execId, signal }), + ), + disposeExec: async ({ id: execId, runtimeId }) => { + const key = this.#executionRuntimeKey(id, execId); + const expectedRuntimeId = runtimeId ?? this.#executionRuntimes.get(key); + await this.#runExecutionOperation( + id, + "shell.disposeExec", + execId, + expectedRuntimeId, + (shellRpc) => shellRpc.disposeExec({ id: execId }), + ); + this.#executionRuntimes.delete(key, expectedRuntimeId); }, }; this.#commandHandles.set(id, adapter); @@ -900,6 +1011,10 @@ export class Workspace { #handleFor(id: string): Promise { const cached = this.#handles.get(id); if (cached !== undefined) return Promise.resolve(cached); + const disconnecting = this.#disconnecting.get(id); + if (disconnecting !== undefined) { + return disconnecting.then(() => this.#handleFor(id)); + } const inflight = this.#connecting.get(id); if (inflight !== undefined) return inflight; const backend = this.#backendsById.get(id); @@ -909,56 +1024,65 @@ export class Workspace { const generation = this.#connectionGeneration; let promise!: Promise; promise = (async () => { - const handle = await withSpan( - this.#observer, - "workspace.connect", - { "workspace.backend.id": id, "workspace.backend.type": backend.type }, - () => - backend.connect({ - db: this.#db, - fs: this.#fs, - git: this.#gitFactory ? this.git : DISABLED_GIT_CLIENT, - artifacts: this.#artifacts, - }), - ); - if (generation !== this.#connectionGeneration) { - await handle.close().catch(() => undefined); - throw new Error(`Workspace closed while backend ${JSON.stringify(id)} was connecting.`); - } - // Reconcile watermarks before publishing the handle. If the - // remote restarted between our pushes / fetches it has lost - // state we thought it had; reset the local cursors so the - // next tick rebaselines. - // - // A backend that declares sync: "none" has no remote store - // to reconcile against; skip the pass entirely. - if (handle.sync !== "none") { - await reconcileWatermarks(this.#db, handle.rpc.sync, id); - } - if (generation !== this.#connectionGeneration) { - await handle.close().catch(() => undefined); - throw new Error(`Workspace closed while backend ${JSON.stringify(id)} was connecting.`); - } - this.#handles.set(id, handle); - // Watch the transport for mid-session loss. Backends without - // a `closed` promise (in-process fakes) opt out by omitting - // it; we only react when it's wired. - if (handle.closed) { - handle.closed - .catch(() => {}) - .then(() => { - // Only clear if this handle is still the current one - // for this id. A close() that already ran will have - // dropped the entry; a subsequent #handleFor may have - // installed a new one. - if (this.#handles.get(id) === handle) { - this.#handles.delete(id); - this.#shells.delete(id); - this.#commandHandles.delete(id); - } - }); + let handle: BackendHandle | undefined; + try { + handle = await withSpan( + this.#observer, + "workspace.connect", + { "workspace.backend.id": id, "workspace.backend.type": backend.type }, + () => + backend.connect({ + db: this.#db, + fs: this.#fs, + git: this.#gitFactory ? this.git : DISABLED_GIT_CLIENT, + artifacts: this.#artifacts, + }), + ); + if (generation !== this.#connectionGeneration) { + throw new Error(`Workspace closed while backend ${JSON.stringify(id)} was connecting.`); + } + // Reconcile watermarks before publishing the handle. If the + // remote restarted between our pushes / fetches it has lost + // state we thought it had; reset the local cursors so the + // next tick rebaselines. + // + // A backend that declares sync: "none" has no remote store + // to reconcile against; skip the pass entirely. + if (handle.sync !== "none") { + await reconcileWatermarks(this.#db, handle.rpc.sync, id); + } + if (generation !== this.#connectionGeneration) { + throw new Error(`Workspace closed while backend ${JSON.stringify(id)} was connecting.`); + } + this.#handles.set(id, handle); + // Watch the transport for mid-session loss. Backends without + // a `closed` promise (in-process fakes) opt out by omitting + // it; we only react when it's wired. + if (handle.closed) { + handle.closed + .catch(() => {}) + .then(() => { + // Only clear if this handle is still the current one + // for this id. A close() that already ran will have + // dropped the entry; a subsequent #handleFor may have + // installed a new one. + if (this.#handles.get(id) === handle) { + this.#handles.delete(id); + this.#shells.delete(id); + this.#commandHandles.delete(id); + } + }); + } + return handle; + } catch (error) { + // A handle that fails setup was never published, but the + // backend may still cache it internally. Close it so the next + // connect attempt cannot return the same broken session. + if (handle !== undefined && this.#handles.get(id) !== handle) { + await handle.close().catch(() => undefined); + } + throw error; } - return handle; })().finally(() => { // Always drop this in-flight entry so a failed connect can be // retried, without deleting a newer connection started after close(). @@ -968,28 +1092,150 @@ export class Workspace { return promise; } - // Per-backend CommandExecutor, constructed on demand and cached - // for the life of the handle. Returns both the shell and the - // BackendHandle it was built against so the caller can hold the - // handle reference for a later identity check; #invalidateHandle - // clears both caches together, so a shell pulled from #shells is - // always paired with the live handle for that id at the moment - // of the lookup. - async #shellFor(id: string): Promise<{ shell: CommandExecutor; handle: BackendHandle }> { - const handle = await this.#handleFor(id); + // Per-backend CommandExecutor, constructed on demand. Exec keeps + // its pre-command push and spawn on one BackendHandle. If that + // handle fails before dispatch, the reconnect retry repeats both + // steps on the replacement so the command cannot skip its push. + // Other shell operations resolve a handle at call time and bind + // stream failures to the handle that produced the envelope. + #shellFor(id: string): CommandExecutor { const cached = this.#shells.get(id); - if (cached !== undefined) return { shell: cached, handle }; + if (cached !== undefined) return cached; + const dispatch = (input: Parameters[0]) => + this.#runWithReconnect( + id, + "shell.exec", + async (handle) => { + let pushed: number; + try { + pushed = await this.#pushForExec(id, handle); + } catch (error) { + if (!isWorkspaceTransportFailure(error)) throw error; + throw new WorkspacePreDispatchTransportError( + `pre-exec push failed: ${safeErrorMessage(error)}`, + { cause: error }, + ); + } + const envelope = await spawnShell(handle.rpc.shell, input, this.#observer); + return { + pushed, + envelope: wrapShellEnvelope( + envelope, + (error) => this.#onShellError(id, handle, error), + handle.runtimeId, + ), + }; + }, + "pre-dispatch", + ); + const getDispatch = (input: Parameters[0] & { runtimeId?: string }) => + this.#runShellEnvelope( + id, + "shell.getExec", + (shell) => shell.getExec({ id: input.id, after: input.after }), + "always", + { executionId: input.id, runtimeId: input.runtimeId }, + ); + const rpc: ShellRPC = { + exec: async (input) => (await dispatch(input)).envelope, + getExec: (input) => getDispatch(input), + killExec: (input) => + this.#runWithReconnect(id, "shell.killExec", (handle) => handle.rpc.shell.killExec(input)), + disposeExec: (input) => + this.#runWithReconnect(id, "shell.disposeExec", (handle) => + handle.rpc.shell.disposeExec(input), + ), + }; const shell = new CommandExecutor( - handle.rpc.shell, + rpc, { push: () => this.push(id), - pull: () => this.pull(id), - onPullPending: () => this.#schedulePendingSync(id), + pull: (runtimeId) => this.#pullForExec(id, runtimeId), + onPullPending: (_error, runtimeId) => this.#schedulePendingSync(id, runtimeId), }, this.#observer, + dispatch, + getDispatch, ); this.#shells.set(id, shell); - return { shell, handle }; + return shell; + } + + #pullForExec(id: string, runtimeId?: string): Promise { + return this.#serialize(id, (resolvedId) => this.#pullResolved(resolvedId, runtimeId)); + } + + #pushForExec(id: string, handle: BackendHandle): Promise { + return this.#serialize(id, (resolvedId) => + withSpan( + this.#observer, + "workspace.sync.push", + { "workspace.sync.backend": resolvedId }, + async () => { + if (resolvedId === undefined || this.#moduleBackendsById.has(resolvedId)) return 0; + if (handle.sync === "none") return 0; + return pushOnce(this.#db, handle.rpc.sync, resolvedId); + }, + (span, outcome) => { + if (outcome.ok) span.setAttribute("workspace.sync.pushed", outcome.value); + }, + ), + ); + } + + #runShellEnvelope( + id: string, + operation: string, + op: (shell: ShellRPC) => ReturnType, + policy: BackendRetryPolicy = "always", + expected?: { executionId: string; runtimeId?: string }, + ): ReturnType { + return this.#runWithReconnect( + id, + operation, + async (handle) => { + if (expected !== undefined) { + assertExecutionRuntime(expected.executionId, expected.runtimeId, handle.runtimeId); + } + const envelope = await op(handle.rpc.shell); + return wrapShellEnvelope( + envelope, + (error) => this.#onShellError(id, handle, error), + handle.runtimeId, + ); + }, + policy, + ); + } + + #runExecutionOperation( + id: string, + operation: string, + executionId: string, + runtimeId: string | undefined, + op: (shell: ShellRPC) => Promise, + ): Promise { + return this.#runWithReconnect(id, operation, async (handle) => { + assertExecutionRuntime(executionId, runtimeId, handle.runtimeId); + await op(handle.rpc.shell); + }); + } + + #executionRuntimeKey(backendId: string, executionId: string): string { + return JSON.stringify([backendId, executionId]); + } + + #rememberExecutionRuntime( + backendId: string, + executionId: string, + runtimeId: string | undefined, + ): void { + if (runtimeId !== undefined) { + this.#executionRuntimes.remember( + this.#executionRuntimeKey(backendId, executionId), + runtimeId, + ); + } } // Invalidate the cached handle for `id` when a shell-routed RPC @@ -999,10 +1245,48 @@ export class Workspace { // a concurrent reconnect already installed. #onShellError(id: string, handle: BackendHandle, error: unknown): void { if (!isWorkspaceTransportFailure(error)) return; - this.#invalidateHandle(id, handle); + void this.#invalidateHandle(id, handle); } } +function assertExecutionRuntime( + executionId: string, + expectedRuntimeId: string | undefined, + currentRuntimeId: string | undefined, +): void { + if (expectedRuntimeId === undefined || currentRuntimeId === expectedRuntimeId) return; + throw Object.assign( + new Error( + `Execution ${JSON.stringify(executionId)} was lost when its container runtime was replaced.`, + ), + { name: "WorkspaceExecutionLostError", code: "EEXEC_LOST" }, + ); +} + +// Keep ownership of the original capnweb result envelope while +// replacing its event stream with a transport-aware wrapper. The +// CommandExecutor disposes this local envelope when the stream ends; +// forwarding that disposal releases the real remote envelope exactly +// once. +function wrapShellEnvelope( + envelope: Awaited>, + onError: (error: unknown) => void, + runtimeId?: string, +): Awaited> & { runtimeId?: string } { + let disposed = false; + const wrapped = { + id: envelope.id, + runtimeId, + events: watchStreamForTransportError(envelope.events, onError), + [Symbol.dispose]() { + if (disposed) return; + disposed = true; + maybeDispose(envelope); + }, + }; + return wrapped; +} + // Pass an execution event stream through unchanged, but classify any // error that tears it down. A transport-classified mid-stream failure // invalidates the cached backend handle so the next call reconnects, diff --git a/packages/computer/tests/stub-soak-worker.ts b/packages/computer/tests/stub-soak-worker.ts index 941c9dba..f438475c 100644 --- a/packages/computer/tests/stub-soak-worker.ts +++ b/packages/computer/tests/stub-soak-worker.ts @@ -36,14 +36,16 @@ export interface Env { // methods return empty/no-op; exec returns a single exit event so // shell.exec resolves without a real subprocess. function fakeBackend(): WorkspaceBackend { + let appliedPushRev = 0; const sync: SyncRPC = { - async push() { - return { rev: 0, appliedPushCursor: { rev: 0, path: null } }; + async push(input) { + appliedPushRev = input.senderRev; + return { rev: 0, appliedPushCursor: { rev: appliedPushRev, path: null } }; }, async fetchChanges() { return { currentCursor: { rev: 0, path: null }, - appliedPushCursor: { rev: 0, path: null }, + appliedPushCursor: { rev: appliedPushRev, path: null }, stream: new ReadableStream({ start(c) { c.close(); @@ -55,7 +57,11 @@ function fakeBackend(): WorkspaceBackend { return null; }, async watermarks() { - return { currentRev: 0, pushRev: 0, fetchCursor: { rev: 0, path: null } }; + return { + currentRev: 0, + pushRev: 0, + fetchCursor: { rev: appliedPushRev, path: null }, + }; }, async hasObjects() { return [];