diff --git a/.agents/skills/capnweb/SKILL.md b/.agents/skills/capnweb/SKILL.md index e650304b..bc4e8610 100644 --- a/.agents/skills/capnweb/SKILL.md +++ b/.agents/skills/capnweb/SKILL.md @@ -14,7 +14,9 @@ description: | between the Durable Object and `computerd`. It's an object-capability RPC system with promise pipelining, structured-clone-style transfer of stubs, and bidirectional calls. The wire format used here is text -JSON over a long-lived WebSocket, with an HTTP batch alternative. +JSON over a long-lived WebSocket. That is the only carrier: capnweb's +HTTP batch transport cannot deliver a stream returned from a call, +which is what every read on this interface is. ## Where things live @@ -48,10 +50,10 @@ you don't explicitly dispose stubs, you leak resources on the other side of the connection. This matters more here than in many capnweb deployments because the -connection is **long-lived**. HTTP batch sessions auto-dispose -everything when the batch ends, but our WebSocket between the -Durable Object and `computerd` stays up for the lifetime of the workspace. -Every undisposed stub stays alive until that connection drops. +connection is **long-lived**. A short-lived session disposes everything +when it ends; our WebSocket between the Durable Object and `computerd` +stays up for the lifetime of the workspace, so every undisposed stub +stays alive until that connection drops. ## The caller-disposes rule diff --git a/.agents/skills/debugging-computerd-fuse/SKILL.md b/.agents/skills/debugging-computerd-fuse/SKILL.md index c609979d..fca758fe 100644 --- a/.agents/skills/debugging-computerd-fuse/SKILL.md +++ b/.agents/skills/debugging-computerd-fuse/SKILL.md @@ -1,6 +1,6 @@ --- name: debugging-computerd-fuse -description: Debug computerd in real-FUSE mode end-to-end without workerd, vitest-pool-workers, or wrangler in the loop. Boot the linux-x64 binary in a privileged docker container, drive its capnweb /ws endpoint from Node, simulate DO-side sync from a SQLiteTestStorage, and isolate FUSE-related deadlocks. Load when a real-FUSE bug reproduces locally but unit tests pass, when the harness vitest tests hang against a real container, or when you need to attribute a wedge to FUSE vs sync vs exec. +description: Debug computerd in real-FUSE mode end-to-end without workerd, vitest-pool-workers, or wrangler in the loop. Boot the linux-x64 binary in a privileged docker container, drive its capnweb /api endpoint from Node, simulate DO-side sync from a SQLiteTestStorage, and isolate FUSE-related deadlocks. Load when a real-FUSE bug reproduces locally but unit tests pass, when the harness vitest tests hang against a real container, or when you need to attribute a wedge to FUSE vs sync vs exec. --- # Debugging computerd against real FUSE @@ -87,10 +87,12 @@ with a missing `/dev/fuse` would have failed startup outright. ## Drive computerd from a Node script -computerd serves a composite `WorkspaceRPC` over `/ws` (capnweb WebSocket) -and `/api` (capnweb HTTP batch). The `@cloudflare/computer-rpc/client` -package wraps the WS form and the `/driver` subpath exposes -`pushOnce`/`pullOnce` against a Node-side `Database`. +computerd serves a composite `WorkspaceRPC` over `/api`, a capnweb +WebSocket. That is the only RPC carrier: capnweb's HTTP batch transport +cannot deliver a returned stream, which is what every read on this +interface is. The `@cloudflare/computer-rpc/client` package wraps the +socket and the `/driver` subpath exposes `pushOnce`/`pullOnce` against a +Node-side `Database`. Set up a probe project once: @@ -123,7 +125,7 @@ import { pullOnce, pushOnce } from "@cloudflare/computer-rpc/driver"; import { WebSocket } from "ws"; const url = process.env.COMPUTERD_URL; // e.g. http://127.0.0.1:18080 -const wsUrl = `${url.replace(/^http(s?):\/\//, "ws$1://")}/ws`; +const wsUrl = `${url.replace(/^http(s?):\/\//, "ws$1://")}/api`; const storage = new SQLiteTestStorage(); const db = new Database(storage); diff --git a/.changeset/connect-names-host-paths.md b/.changeset/connect-names-host-paths.md new file mode 100644 index 00000000..b1cad5d0 --- /dev/null +++ b/.changeset/connect-names-host-paths.md @@ -0,0 +1,5 @@ +--- +"@cloudflare/computer": minor +--- + +The /connect caller now provides both API & healthcheck endpoints. The container no longer builds either path itself, so a host is free to serve them wherever it likes. diff --git a/.changeset/remove-upstream-url.md b/.changeset/remove-upstream-url.md new file mode 100644 index 00000000..a1ffec68 --- /dev/null +++ b/.changeset/remove-upstream-url.md @@ -0,0 +1,5 @@ +--- +"@cloudflare/computer": minor +--- + +The UPSTREAM_URL environment variable has been removed along with the container's own sync loop. Syncing is driven by whichever peer holds the other end of the Cap'n Web session. diff --git a/.changeset/rename-ws-endpoint-to-api.md b/.changeset/rename-ws-endpoint-to-api.md new file mode 100644 index 00000000..d21fef3f --- /dev/null +++ b/.changeset/rename-ws-endpoint-to-api.md @@ -0,0 +1,5 @@ +--- +"@cloudflare/computer": minor +--- + +The Cap'n Web /ws endpoint has been renamed to /api at both ends of the connection. A durable object that routes the container's outbound upgrade must match /api in its own fetch handler. Support for the Cap'n Web HTTP batch transport has been removed, so /api carries a websocket only. diff --git a/AGENTS.md b/AGENTS.md index c886cc99..7469bf19 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -133,8 +133,9 @@ loop. Reach for these when you're chasing a behavior the unit tests don't cover. - `shell` boots a debian-slim container with the linux `computerd` binary mounted under `/usr/local/bin`. The starting point for anything that needs a real FUSE mount. -- `computerd-soak.mjs` boots two `computerd` containers wired peer-to-peer and - soaks the sync loop. Use it to chase convergence or churn bugs. +- `computerd-soak.mjs` boots one `computerd` container, takes the host's + part of the sync loop from a local store, and reports convergence lag + beside container memory. Use it to chase convergence or churn bugs. - `computerd-stub-soak.mjs` soaks the long-lived WebSocket session and reads `session.getStats()` to detect stub-disposal drift. Run it for changes around the capnweb lifecycle. @@ -147,6 +148,10 @@ loop. Reach for these when you're chasing a behavior the unit tests don't cover. tasks against the mount with a tmpfs baseline for comparison. - `exec-tests` boots `computerd` in docker with FUSE disabled and exercises a few `shell.exec` scenarios. +- `lib/cloudflare-workers-stub.mjs` is a node `--import` hook that stubs + the `cloudflare:workers` module, so a host-side script can import + `@cloudflare/computer`'s main entry outside workerd. Not runnable on + its own; `exec-tests` uses it. - `npm-bench.sh` / `run-npm-bench.sh` benchmark npm package installs on native disk vs the FUSE mount. `run-npm-bench.sh` is the user-facing entry point that boots a privileged Docker container; diff --git a/docs/07_injected_service.md b/docs/07_injected_service.md index c323f754..e3b8b58f 100644 --- a/docs/07_injected_service.md +++ b/docs/07_injected_service.md @@ -32,9 +32,8 @@ staging the binary into a container image. same paths. The backend is picked by `FUSE_MOUNT` (default `auto`, see the env-var table below). 2. **Dirty tracking.** Writes that flow through FUSE land in the - in-container VFS database; sync (when `UPSTREAM_URL` is set) is - what surfaces those revisions back out. See doc 02 for the sync - protocol. + in-container VFS database; the host pulls those revisions back out + across the capnweb session. See doc 02 for the sync protocol. 3. **Exec.** Runs shell commands and streams stdout/stderr back over capnweb. See [05. Shell Interface](./05_runtime_interface.md). 4. **Apply.** Accepts changes pushed by the DO and writes them into @@ -52,11 +51,15 @@ backend pins it to `8080`) and serves: | --- | --- | --- | | `/health` | `GET`, `HEAD` | Liveness probe; `200 ok\n` as soon as the HTTP server binds. | | `/__computerd/info` | `GET` | Runtime info: FUSE backend, mount point, port. | -| `/api` | `POST` | HTTP-batch capnweb transport. | -| `/ws` | `GET` (upgrade) | WebSocket capnweb transport — the bootstrap stub is `WorkspaceRPC`. | -| `/connect` | `POST` | Tells `computerd` to dial *out* to a caller-supplied URL and serve a `WorkspaceRPC` session over that outbound WebSocket. Used by the Cloudflare backend (see below). | +| `/api` | `GET` (upgrade) | WebSocket capnweb transport — the bootstrap stub is `WorkspaceRPC`. Only the exact path upgrades. A request without an `Upgrade` header gets `400`; an unsupported `Sec-WebSocket-Version` gets `426` and the versions the server speaks. | +| `/api/watermarks` | `GET`, `HEAD` | Sync revisions: `currentRev`, `pushRev`, `fetchCursor`. The same values `sync.watermarks()` returns, for callers that want a few numbers without holding a session. | +| `/connect` | `POST` | Tells `computerd` to dial *out* to a caller-supplied endpoint and serve a `WorkspaceRPC` session over that outbound WebSocket. Used by the Cloudflare backend (see below). | | `/` | `GET` | Banner/info page. | +`/api` is the workspace surface: the session itself, plus anything that +reads through it. `/__computerd` is daemon introspection, which is why +runtime info sits there and revisions do not. + The capnweb bootstrap interface is **`WorkspaceRPC`** (defined in `packages/rpc/`), split into `sync` and `shell` sub-stubs. @@ -108,9 +111,9 @@ Provider-agnostic shape — three steps, in order: path the mount is awaited *before* `listen`, so by the time `/health` answers FUSE is up too. With `FUSE_MOUNT=none` there is no FUSE step at all. -3. **Open the capnweb session.** Either the host upgrades to `/ws` - directly, or it asks `computerd` (via `POST /connect`) to dial *out* to a - URL it controls and serve the session over that outbound socket. +3. **Open the capnweb session.** Either the host upgrades to `/api` + directly, or it asks `computerd` (via `POST /connect`) to dial *out* to an + endpoint it controls and serve the session over that outbound socket. Either way, the bootstrap stub is `WorkspaceRPC`. ### Cloudflare Containers specifics @@ -132,8 +135,10 @@ wires it like this: repeated until it returns `200`. 4. **Invert the WebSocket.** The DO arms an upgrade slot (`#armUpgrade`) and then `POST`s to `/connect` on the container - (`#postConnect`). `computerd` reads that request and dials *out* to the - egress at `${egressHost}/ws`. Because the egress is intercepted, + (`#postConnect`). The request names the egress base and both paths, + so `computerd` polls `base + health` and then dials `base + api`; + the daemon assembles no paths of its own. Because the egress is + intercepted, that outbound dial loops back to the DO's `handleFetch()`, which accepts the upgrade and resolves the in-flight `#pendingUpgrade`. The capnweb session then runs over that socket. **The WebSocket @@ -154,10 +159,9 @@ These are the variables `computerd` actually consumes (see | Variable | Default | Meaning | | --- | --- | --- | -| `PORT` | `45678` | Port the HTTP/WS server listens on. CF backend pins this to `8080`. | +| `PORT` | `45678` | Port the HTTP server listens on. CF backend pins this to `8080`. | | `MOUNT_POINT` | `/workspace` | Absolute path inside the container to mount the FUSE filesystem at. Ignored when `FUSE_MOUNT=none`. | | `FUSE_MOUNT` | `auto` | Backend selector: `auto` probes `/dev/fuse` (linux) or macFUSE (darwin) and falls back to the userspace shim; `fuse` / `macfuse` require the corresponding real backend; `shim` forces the userspace shim; `none` skips the mount entirely. | -| `UPSTREAM_URL` | unset | If set, `computerd` starts a sync client against this URL to push/pull VFS revisions. | | `EXEC_LOG_MAX_BYTES` | runner default | Caps the per-exec stdout/stderr log retained in-memory. | | `LOG_FILE` | unset | If set, every `console.log` / `console.error` line and any `uncaughtException` / `unhandledRejection` is also appended to this file. Stdout/stderr behaviour is unchanged. | diff --git a/docs/08_capnweb_interface.md b/docs/08_capnweb_interface.md index c4ed867a..d708b0d4 100644 --- a/docs/08_capnweb_interface.md +++ b/docs/08_capnweb_interface.md @@ -6,16 +6,14 @@ [capnweb](https://github.com/cloudflare/capnweb) is the RPC framing used between the Durable Object and the in-container `computerd` computerd. -The wire format is text JSON over a single WebSocket (with an HTTP-batch -alternative). The interface served is `WorkspaceRPC`, defined in +The wire format is text JSON over a single WebSocket. The interface +served is `WorkspaceRPC`, defined in `packages/rpc/src/interface.ts` and consumed by both sides. ## Transport -- **Carrier.** One long-lived WebSocket per Workspace. The DO opens it - against the computerd's `/ws` endpoint, with `/api` available - as an HTTP-batch alternative (single POST per call) for callers that - can't hold a socket. Default port is `45678`; it will become a +- **Carrier.** One long-lived WebSocket per Workspace, on the + computerd's `/api` endpoint. Default port is `45678`; it will become a build-time variable so hosts can pin a non-default port. - **Framing.** capnweb text frames. Binary frames are unsupported. **(planned)** the server will fail the session loudly on the first diff --git a/docs/11_lifecycle.md b/docs/11_lifecycle.md index 4166a907..9e72d830 100644 --- a/docs/11_lifecycle.md +++ b/docs/11_lifecycle.md @@ -24,7 +24,7 @@ capnweb WebSocket session. │ │ Workspace │ │ │ │ computerd │ │ │ │ fs: WorkspaceFS │ │ │ │ HTTP server │ │ │ │ shell: ShellRPC │ │ │ │ /health │ │ -│ │ sync: SyncRPC │◀─┼── capnweb WS ───▶│ │ /connect /ws │ │ +│ │ sync: SyncRPC │◀─┼── capnweb WS ───▶│ │ /connect /api │ │ │ │ push() / pull() │ │ │ │ FUSE mount │ │ │ │ ready() │ │ │ │ exec runner │ │ │ └──────────┬──────────┘ │ │ └────────┬─────────┘ │ @@ -50,7 +50,7 @@ The 1:1 mapping is load-bearing for several reasons: doesn't have to multiplex multiple WS peers. The DO is the WebSocket *server* in this pairing, even though `computerd` -exposes its own `/ws` server-side and could be dialed directly. The +exposes its own `/api` server-side and could be dialed directly. The reason for the inversion is documented in [07. Injected Service §Bootstrap sequence](./07_injected_service.md): the egress interceptor needs to be wired before any traffic flows, and @@ -90,7 +90,7 @@ an incarnation boundary. What survives is: On every new incarnation `Workspace.ready()` re-runs `#connect()`, which re-enters the backend's bootstrap sequence. If the container is -still alive, the backend's `POST /connect` + `/ws` handshake produces +still alive, the backend's `POST /connect` + `/api` handshake produces a fresh capnweb session against the same in-memory VFS on the container side. If the container died too (e.g. host OOM took both), the next sync round is a rev-0 baseline rebuild from the DO's store. @@ -129,9 +129,9 @@ that point so the next call rebuilds from scratch (see the container host and ba The critical asymmetry: the **container's VFS is process-lifetime in-memory**, while the **DO's VFS is durable SQLite**. A container -restart loses container-side state. Sync via `UPSTREAM_URL` (which -the Cloudflare backend wires automatically) is what brings state back -on the next push/pull round. +restart loses container-side state. The durable object drives sync +across the capnweb session it opens through `POST /connect`, and that +is what brings state back on the next push/pull round. ## Capnweb lifecycle @@ -373,7 +373,7 @@ between agent turns, and it's exactly where hibernation pays off. Adopting PartySocket for reconnect/backoff would require the DO to be the WebSocket *client* dialing `computerd`'s -`/ws` endpoint. That model is appealing for reconnect, but +`/api` endpoint. That model is appealing for reconnect, but hibernatable WebSockets only work server-side via `ctx.acceptWebSocket()` — there is no hibernation API for outbound client sockets. **Inverting the dial direction permanently forecloses diff --git a/docs/README.md b/docs/README.md index b36ce6ef..0a359ce3 100644 --- a/docs/README.md +++ b/docs/README.md @@ -87,7 +87,7 @@ To build the binary from source instead, run `npm run build:bin `artifacts/computerd/computerd-linux-x64`, then `COPY` that into the image. -`computerd`'s own default port is `45678`; the Cloudflare container backend pins the in-image listener to `8080`, which is what `examples/container/` uses. See [07. Injected Service](./07_injected_service.md) for the env vars (`PORT`, `MOUNT_POINT`, `FUSE_MOUNT`, `UPSTREAM_URL`, `EXEC_LOG_MAX_BYTES`) and the reverse-dial boot sequence. +`computerd`'s own default port is `45678`; the Cloudflare container backend pins the in-image listener to `8080`, which is what `examples/container/` uses. See [07. Injected Service](./07_injected_service.md) for the env vars (`PORT`, `MOUNT_POINT`, `FUSE_MOUNT`, `EXEC_LOG_MAX_BYTES`) and the reverse-dial boot sequence. ## Example diff --git a/examples/container/README.md b/examples/container/README.md index 0e813bc3..9f6f220b 100644 --- a/examples/container/README.md +++ b/examples/container/README.md @@ -21,7 +21,7 @@ client ─► Worker /c//{file,exec} ▼ DO (ContainerExample) ──► Container ──► computerd (:8080) ▲ │ - │ ws://computer.internal/ws │ + │ ws://computer.internal/api │ └────────── capnweb session ◄──────┘ ``` @@ -30,19 +30,20 @@ client ─► Worker /c//{file,exec} `Workspace` instance. That backend owns the entire computerd lifecycle: container start, outbound egress interception, port-readiness polling, POST - `/connect` to computerd, `/ws` upgrade routing, and capnweb session + `/connect` to computerd, `/api` upgrade routing, and capnweb session attach. 2. computerd reaches the Worker through the container's **outbound interception** (`ctx.container.interceptOutboundHttp("computer.internal", …)`, set up by the backend). The DO passes `ctx.exports.WorkspaceProxy({ props: { binding, id } })` as the egress fetcher; that `WorkerEntrypoint` (re-exported from - `@cloudflare/computer`) routes `/ws` upgrades back to the owning DO. + `@cloudflare/computer`) routes `/api` upgrades back to the owning DO. 3. When `Workspace.ready()` is called for the first time, the backend posts `/connect` into computerd with - `{ url: "http://computer.internal" }`. computerd polls - `computer.internal/health`, then dials - `ws://computer.internal/ws`. + `{ base: "http://computer.internal", health: "/health", api: "/api" }`. + computerd polls `computer.internal/health`, then dials + `ws://computer.internal/api`. Naming both paths in the request keeps + the daemon from assembling routes it does not serve. 4. `WorkspaceProxy.fetch` forwards the upgrade to the DO's `fetch()` via the DO binding looked up from its props. The DO's `fetch()` delegates to `backend.handleFetch(req)`, which performs the diff --git a/examples/container/src/index.ts b/examples/container/src/index.ts index 28510bd1..791cd58e 100644 --- a/examples/container/src/index.ts +++ b/examples/container/src/index.ts @@ -4,7 +4,7 @@ // The DO is a thin shell over CloudflareContainerBackend: it picks // the container (this.ctx.container) and the egress fetcher // (ctx.exports.WorkspaceProxy bound to this DO instance), forwards -// container-bound /ws upgrades back through the backend, and +// container-bound /api upgrades back through the backend, and // otherwise just calls into a single Workspace instance. // // Wire shape: @@ -14,7 +14,7 @@ // ▼ // ContainerExample DO ──► Container ──► computerd (:8080) // ▲ │ -// │ ws://computer.internal/ws │ +// │ ws://computer.internal/api │ // └─── capnweb session ◀─────────────┘ import { DurableObject, tracing } from "cloudflare:workers"; @@ -92,7 +92,7 @@ function workspaceOptions(self: InstanceType): WorkspaceOp // hands back round-trip into this DO; the actual SyncRPC + ShellRPC // traffic stays on the computerd ↔ DO capnweb wire. export class ContainerExample extends withWorkspace(ContainerBase, workspaceOptions) { - // ---- WebSocket: computerd's outbound /ws upgrade --------------------- + // ---- WebSocket: computerd's outbound /api upgrade -------------------- override async fetch(request: Request): Promise { return this.backend.handleFetch(request); diff --git a/examples/mcp/src/index.test.ts b/examples/mcp/src/index.test.ts index 8cbe9933..189e062e 100644 --- a/examples/mcp/src/index.test.ts +++ b/examples/mcp/src/index.test.ts @@ -26,7 +26,7 @@ describe("Computer Code Mode MCP", () => { expect(health.status).toBe(200); expect(await health.text()).toBe("ok\n"); - const internal = await SELF.fetch("https://example.test/ws"); + const internal = await SELF.fetch("https://example.test/api"); expect(internal.status).toBe(404); }); diff --git a/examples/mcp/src/index.ts b/examples/mcp/src/index.ts index e8f56f82..d546d4fe 100644 --- a/examples/mcp/src/index.ts +++ b/examples/mcp/src/index.ts @@ -60,7 +60,7 @@ export class ComputerMCP extends withWorkspace(ComputerMCPBase, workspaceOptions const path = new URL(request.url).pathname; // computerd reaches this callback through an internal binding. The public // Worker forwards only /mcp. - if (path === "/ws") return this.containerShell.handleFetch(request); + if (path === "/api") return this.containerShell.handleFetch(request); if (path !== "/mcp") return new Response("not found", { status: 404 }); const unauthorized = authorize(request, this.env.MCP_TOKEN); diff --git a/examples/think-compare-runtimes/worker/think/agents.ts b/examples/think-compare-runtimes/worker/think/agents.ts index 04425f0f..52adf962 100644 --- a/examples/think-compare-runtimes/worker/think/agents.ts +++ b/examples/think-compare-runtimes/worker/think/agents.ts @@ -260,7 +260,7 @@ export class WorkspaceThinkAgent extends RuntimeThinkAgent { override async fetch(request: Request): Promise { const url = new URL(request.url); - if (url.pathname === "/ws" && this.#activeBackend) { + if (url.pathname === "/api" && this.#activeBackend) { return this.#activeBackend.handleFetch(request); } return super.fetch(request); diff --git a/examples/think/src/agent.ts b/examples/think/src/agent.ts index 7a82b725..624af906 100644 --- a/examples/think/src/agent.ts +++ b/examples/think/src/agent.ts @@ -46,7 +46,7 @@ import { createWorkersAI } from "workers-ai-provider"; // WorkerShellBackend reaches WorkspaceServiceProxy through // `ctx.exports.WorkspaceServiceProxy(...)` so the in-isolate shell // can call back into the host workspace. WorkspaceProxy carries the -// container's outbound /ws egress back to this DO. +// container's outbound /api egress back to this DO. export { WorkspaceProxy, WorkspaceServiceProxy }; const MODEL_ID = "@cf/zai-org/glm-5.2"; @@ -73,7 +73,7 @@ export class Assistant extends withWorkspaceContainer(AssistantBase) { * Container backend used when `exec` needs a real Linux userland. * The DO itself owns the container binding through the * withWorkspaceContainer mixin; CloudflareContainerBackend handles - * startup, outbound egress interception, the /ws upgrade, and the + * startup, outbound egress interception, the /api upgrade, and the * capnweb session. */ readonly #containerBackend = new CloudflareContainerBackend({ @@ -103,10 +103,10 @@ export class Assistant extends withWorkspaceContainer(AssistantBase) { useThink: true, }) as Workspace & ThinkWorkspaceCompatibility; - /** Forwarded by WorkspaceProxy for computerd's outbound /ws upgrade. */ + /** Forwarded by WorkspaceProxy for computerd's outbound /api upgrade. */ override async fetch(request: Request): Promise { const url = new URL(request.url); - if (url.pathname === "/ws") { + if (url.pathname === "/api") { return this.#containerBackend.handleFetch(request); } return super.fetch(request); diff --git a/examples/think/src/index.ts b/examples/think/src/index.ts index 1fb31ee2..ac043217 100644 --- a/examples/think/src/index.ts +++ b/examples/think/src/index.ts @@ -11,7 +11,7 @@ * The Assistant, WorkspaceProxy, and WorkspaceServiceProxy classes are * re-exported so the runtime can resolve them by name: Assistant is * the DO binding and container class, WorkspaceProxy carries computerd's - * outbound /ws upgrade back to the DO, and WorkspaceServiceProxy is + * outbound /api upgrade back to the DO, and WorkspaceServiceProxy is * the loopback Fetcher the worker backend hands into its Dynamic * Worker so the in-isolate shell can reach back into the host * workspace. diff --git a/examples/tutorial/README.md b/examples/tutorial/README.md index 47015283..aedf4f17 100644 --- a/examples/tutorial/README.md +++ b/examples/tutorial/README.md @@ -164,7 +164,7 @@ export class RecipeAgent extends withWorkspaceContainer(RecipeBase) { }) as Workspace & ThinkWorkspaceCompatibility; override async fetch(request: Request): Promise { - return new URL(request.url).pathname === "/ws" + return new URL(request.url).pathname === "/api" ? this.#backend.handleFetch(request) : super.fetch(request); } @@ -179,7 +179,7 @@ network access. the durable object can start and stop its own container. The `workspace: { binding, id }` pair is how the container finds its way home: it dials the named binding at that id, which is why `fetch` has to -hand `/ws` to the backend before the base class sees it. +hand `/api` to the backend before the base class sees it. ## 5. Hook the workspace up to the agent diff --git a/examples/tutorial/src/index.ts b/examples/tutorial/src/index.ts index 9a0aae33..c2b077f2 100644 --- a/examples/tutorial/src/index.ts +++ b/examples/tutorial/src/index.ts @@ -80,7 +80,7 @@ export class RecipeAgent extends withWorkspaceContainer(RecipeBase) { } override async fetch(request: Request): Promise { - return new URL(request.url).pathname === "/ws" + return new URL(request.url).pathname === "/api" ? this.#backend.handleFetch(request) : super.fetch(request); } diff --git a/package-lock.json b/package-lock.json index a53cb7a1..17616fe2 100644 --- a/package-lock.json +++ b/package-lock.json @@ -13495,7 +13495,7 @@ "license": "MIT", "dependencies": { "acorn": "^8.17.0", - "capnweb": "^0.8.0", + "capnweb": "^0.10.0", "just-bash": "^3.0.1" }, "devDependencies": { @@ -13540,6 +13540,16 @@ "dev": true, "license": "MIT OR Apache-2.0" }, + "packages/computer/node_modules/capnweb": { + "version": "0.10.0", + "resolved": "https://registry.npmjs.org/capnweb/-/capnweb-0.10.0.tgz", + "integrity": "sha512-lQ2nlzaifkqBeNfs8RLErn7NZ8FNf3vn1W7J3nSH73edWG+0B5nHZGR51S2S9HMZEuZcEypOO469dZRY/BluFA==", + "license": "MIT", + "workspaces": [ + ".", + "packages/*" + ] + }, "packages/computerd": { "name": "@cloudflare/computerd", "version": "0.2.1", @@ -13583,7 +13593,7 @@ "version": "0.2.1", "dependencies": { "@cloudflare/dofs": "*", - "capnweb": "^0.8.0" + "capnweb": "^0.10.0" }, "devDependencies": { "@cloudflare/workers-types": "^4.20260616.1", @@ -13599,6 +13609,16 @@ "integrity": "sha512-mOhf5TUEB1m2vPrxtqoIGfz0fUC9xyxRDx5gWHy5s+OCo6dcV+g7wI1R7gYCMFohhqF/2y2xeKVwMwCJjfn/WA==", "dev": true, "license": "MIT OR Apache-2.0" + }, + "packages/rpc/node_modules/capnweb": { + "version": "0.10.0", + "resolved": "https://registry.npmjs.org/capnweb/-/capnweb-0.10.0.tgz", + "integrity": "sha512-lQ2nlzaifkqBeNfs8RLErn7NZ8FNf3vn1W7J3nSH73edWG+0B5nHZGR51S2S9HMZEuZcEypOO469dZRY/BluFA==", + "license": "MIT", + "workspaces": [ + ".", + "packages/*" + ] } } } diff --git a/packages/computer-computerd-linux-x64/README.md b/packages/computer-computerd-linux-x64/README.md index 1f4254bf..e508f0ce 100644 --- a/packages/computer-computerd-linux-x64/README.md +++ b/packages/computer-computerd-linux-x64/README.md @@ -49,4 +49,3 @@ examples that intentionally track release candidates. | `PORT` | `8080` | HTTP + WebSocket listener port. | | `MOUNT_POINT` | `/workspace` | Path the FUSE filesystem mounts at. | | `FUSE_MOUNT` | `auto` | Backend selector. `auto` probes `/dev/fuse` (linux) or macFUSE (darwin) and falls back to the userspace shim. `fuse` and `macfuse` require their respective real backend. `shim` forces the userspace shim. `none` skips the mount entirely; HTTP / WS still come up. | -| `UPSTREAM_URL` | unset | If set, computerd dials this WebSocket on boot and runs a bidirectional sync loop against it. | diff --git a/packages/computer/package.json b/packages/computer/package.json index f0eefbe1..84da591e 100644 --- a/packages/computer/package.json +++ b/packages/computer/package.json @@ -117,7 +117,7 @@ }, "dependencies": { "acorn": "^8.17.0", - "capnweb": "^0.8.0", + "capnweb": "^0.10.0", "just-bash": "^3.0.1" }, "peerDependencies": { diff --git a/packages/computer/src/backends/container/cloudflare-container.test.ts b/packages/computer/src/backends/container/cloudflare-container.test.ts index 09989a54..51eebe92 100644 --- a/packages/computer/src/backends/container/cloudflare-container.test.ts +++ b/packages/computer/src/backends/container/cloudflare-container.test.ts @@ -4,7 +4,7 @@ // The successful connect() path constructs a WebSocketPair, which // is a workerd global not available under the vitest node runner. // These tests cover the paths that bail before the upgrade (port -// never opens, /connect non-2xx, /ws upgrade timeout), the +// never opens, /connect non-2xx, /api upgrade timeout), the // handleFetch input validation, and the factory + workspace-ref // plumbing. The full happy-path round-trip is covered by the live // example. @@ -34,6 +34,7 @@ interface FakeHostOptions { interface FakeHost { host: IWorkspaceContainerAPI; calls: { name: string; args: unknown[] }[]; + connectBody?: Record; startEnv?: Record; enableInternet?: boolean; interceptedHost?: string; @@ -102,6 +103,10 @@ function makeFakeHost(opts: FakeHostOptions = {}): FakeHost { return new Response(null, { status: 200 }); } if (url.pathname === "/connect") { + state.connectBody = await request + .clone() + .json() + .catch(() => undefined); if (connectStatus !== 200) { return new Response(`/connect ${connectStatus}`, { status: connectStatus }); } @@ -241,7 +246,7 @@ describe("CloudflareContainerBackend", () => { egress: { mode: "http-gateway", gateway }, }); await expect(backend.connect()).rejects.toThrow(); - const request = new Request("https://workspace.internal/ws", { + const request = new Request("https://workspace.internal/api", { method: "POST", body: "payload", headers: { @@ -275,7 +280,7 @@ describe("CloudflareContainerBackend", () => { await expect(backend.connect()).rejects.toThrow(); const response = await backend.handleFetch( - new Request("https://workspace.internal/ws", { + new Request("https://workspace.internal/api", { headers: { "x-workspace-egress-token": fake.gatewayToken ?? "", "x-workspace-egress-url": "ftp://api.example.test/data", @@ -380,7 +385,28 @@ describe("CloudflareContainerBackend", () => { expect(String(error)).toMatch(/POST \/connect returned 502/); }); - test("connect() throws a transport error when the /ws upgrade never arrives", async () => { + test("connect() names the egress base and both paths in the /connect body", async () => { + // The container assembles no host paths of its own, so the + // request has to carry them. A missing field would leave the + // daemon with nothing to dial. + const fake = makeFakeHost(); + const backend = new CloudflareContainerBackend({ + container: () => ({ getWorkspaceContainer: () => fake.host }), + workspace: fakeWorkspace, + connectTimeoutMs: 600, + }); + + await backend.connect().catch(() => {}); + + expect(fake.connectBody).toMatchObject({ + base: "http://computer.internal", + health: "/health", + api: "/api", + }); + expect(typeof fake.connectBody?.healthTimeoutMs).toBe("number"); + }); + + test("connect() throws a transport error when the /api upgrade never arrives", async () => { const fake = makeFakeHost(); const backend = new CloudflareContainerBackend({ container: () => ({ getWorkspaceContainer: () => fake.host }), @@ -390,10 +416,10 @@ describe("CloudflareContainerBackend", () => { const error = await backend.connect().catch((caught: unknown) => caught); expect(error).toBeInstanceOf(WorkspaceTransportError); - expect(String(error)).toMatch(/\/ws upgrade did not arrive/); + expect(String(error)).toMatch(/\/api upgrade did not arrive/); }); - test("handleFetch rejects non-/ws paths", async () => { + test("handleFetch rejects non-/api paths", async () => { const fake = makeFakeHost(); const backend = new CloudflareContainerBackend({ container: () => ({ getWorkspaceContainer: () => fake.host }), @@ -409,8 +435,8 @@ describe("CloudflareContainerBackend", () => { container: () => ({ getWorkspaceContainer: () => fake.host }), workspace: fakeWorkspace, }); - const res = await backend.handleFetch(new Request("http://computer.internal/ws")); - expect(res.status).toBe(426); + const res = await backend.handleFetch(new Request("http://computer.internal/api")); + expect(res.status).toBe(400); }); test("connect() consults host.exitInfo() before host.start()", async () => { @@ -454,9 +480,9 @@ describe("CloudflareContainerBackend", () => { test("connect() restarts the host when initial readiness fails and recovers", async () => { // First attempt drains all probes as failures; restart() runs; // the second attempt's very first probe answers healthy. - // connect() still fails at the /ws upgrade (no WebSocketPair + // connect() still fails at the /api upgrade (no WebSocketPair // under node) — the point is that readiness recovered after - // restart and we reached the /connect POST and /ws upgrade. + // restart and we reached the /connect POST and /api upgrade. const fake = makeFakeHost({ healthSequence: [ // First attempt — enough failures to exhaust the budget. diff --git a/packages/computer/src/backends/container/cloudflare-container.ts b/packages/computer/src/backends/container/cloudflare-container.ts index 5ecd25a8..155a7cb1 100644 --- a/packages/computer/src/backends/container/cloudflare-container.ts +++ b/packages/computer/src/backends/container/cloudflare-container.ts @@ -79,7 +79,7 @@ export interface CloudflareContainerBackendOptions { container: () => ContainerHostHolder | Promise; // Identifies the Workspace-owning DO. Fixed for the lifetime of - // the backend: the backend lives inside this DO and the /ws + // the backend: the backend lives inside this DO and the /api // upgrade always lands here. Plain {binding, id} data so it // survives the Workers RPC hop to a cross-DO container host. workspace: WorkspaceRef; @@ -100,7 +100,7 @@ export interface CloudflareContainerBackendOptions { containerEnv?: Record; // Total time the backend waits for: container port to open, - // /connect POST to return, /ws upgrade to arrive. Default 30s. + // /connect POST to return, /api upgrade to arrive. Default 30s. connectTimeoutMs?: number; // Period for the application-level heartbeat — a watermarks() @@ -138,6 +138,11 @@ export interface CloudflareContainerBackendOptions { } const DEFAULT_EGRESS_HOST = "computer.internal"; +// Paths the egress proxy serves. The container assembles no paths of +// its own, so these travel in the /connect request and both ends stay +// in step from one place. +const EGRESS_HEALTH_PATH = "/health"; +const EGRESS_API_PATH = "/api"; const DEFAULT_CONTAINER_PORT = 8080; const DEFAULT_CONNECT_TIMEOUT_MS = 30_000; const DEFAULT_HEARTBEAT_INTERVAL_MS = 20_000; @@ -160,7 +165,7 @@ export class CloudflareContainerBackend implements WorkspaceBackend { readonly #egress: WorkspaceEgressPolicy; readonly #egressToken: string | undefined; - // State for the in-flight /ws upgrade. handleFetch() resolves + // State for the in-flight /api upgrade. handleFetch() resolves // #pendingUpgrade; connect() awaits it. #pendingUpgrade: Promise | undefined; #resolveUpgrade: ((ws: WebSocket) => void) | undefined; @@ -328,7 +333,7 @@ export class CloudflareContainerBackend implements WorkspaceBackend { return handle; } - // Routes a /ws upgrade Request into the in-flight connect(). + // Routes an /api upgrade Request into the in-flight connect(). // Returns the 101 response that the WorkspaceProxy fetch handler // forwards back to the container. async handleFetch(req: Request): Promise { @@ -354,11 +359,14 @@ export class CloudflareContainerBackend implements WorkspaceBackend { return this.#egress.gateway.fetch(new Request(parsedUrl, sanitized)); } const url = new URL(req.url); - if (url.pathname !== "/ws") { + if (url.pathname !== EGRESS_API_PATH) { return new Response("not found", { status: 404 }); } + // A request with no Upgrade header is a malformed handshake, which + // is a 400. 426 belongs to the narrower case of a version this end + // does not speak, and is what the daemon answers for it. if (req.headers.get("upgrade") !== "websocket") { - return new Response("expected websocket upgrade", { status: 426 }); + return new Response(`${EGRESS_API_PATH} requires a websocket upgrade`, { status: 400 }); } const pair = new WebSocketPair(); @@ -519,7 +527,9 @@ export class CloudflareContainerBackend implements WorkspaceBackend { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ - url: `http://${this.#options.egressHost}`, + base: `http://${this.#options.egressHost}`, + health: EGRESS_HEALTH_PATH, + api: EGRESS_API_PATH, healthTimeoutMs: remaining, }), }); @@ -557,7 +567,7 @@ export class CloudflareContainerBackend implements WorkspaceBackend { () => reject( new WorkspaceTransportError( - `CloudflareContainerBackend(${this.id}) [stage=ws]: /ws upgrade did not arrive within ${this.#options.connectTimeoutMs}ms`, + `CloudflareContainerBackend(${this.id}) [stage=ws]: /api upgrade did not arrive within ${this.#options.connectTimeoutMs}ms`, ), ), remaining, diff --git a/packages/computer/src/backends/container/container-host.ts b/packages/computer/src/backends/container/container-host.ts index 1ba1a8f3..eba11ba9 100644 --- a/packages/computer/src/backends/container/container-host.ts +++ b/packages/computer/src/backends/container/container-host.ts @@ -34,7 +34,7 @@ import { export type { ContainerExitInfo } from "./container-lifecycle.js"; // Identifies the Durable Object that owns the Workspace and answers -// the /ws upgrade. Plain data so it can travel over Workers RPC. +// the /api upgrade. Plain data so it can travel over Workers RPC. export interface WorkspaceRef { // Binding name in the host Worker's env that resolves to the // DurableObjectNamespace for the Workspace-owning DO class. diff --git a/packages/computer/src/backends/test.ts b/packages/computer/src/backends/test.ts index 75ca24c2..66e9e7ff 100644 --- a/packages/computer/src/backends/test.ts +++ b/packages/computer/src/backends/test.ts @@ -41,7 +41,7 @@ export class TestBackend implements WorkspaceBackend { // first RPC. The probe surfaces "harness forgot to start the // container" up front. await probeHealth(this.#url); - const client = createWorkspaceClient({ url: `${wsUrl}/ws` }); + const client = createWorkspaceClient({ url: `${wsUrl}/api` }); return { rpc: client, close: async () => { diff --git a/packages/computer/src/proxy.ts b/packages/computer/src/proxy.ts index 3da08082..6d21d6f3 100644 --- a/packages/computer/src/proxy.ts +++ b/packages/computer/src/proxy.ts @@ -33,10 +33,10 @@ // // override async fetch(req: Request): Promise { // // The DO answers /health (port-readiness poll from the -// // backend) and /ws (capnweb upgrade) on its own fetch(). +// // backend) and /api (capnweb upgrade) on its own fetch(). // const url = new URL(req.url); // if (url.pathname === "/health") return new Response("ok\n"); -// if (url.pathname === "/ws") return this.#backend.handleFetch(req); +// if (url.pathname === "/api") return this.#backend.handleFetch(req); // return new Response("not found", { status: 404 }); // } // } @@ -46,7 +46,7 @@ // proxy looks up `env[binding]` at fetch time and falls back to a // clear error if the name doesn't resolve. The DO class doesn't // need to live in @cloudflare/computer — the proxy works for any -// DO that implements a fetch() handler answering /health and /ws. +// DO that implements a fetch() handler answering /health and /api. import { RpcTarget, WorkerEntrypoint } from "cloudflare:workers"; @@ -71,14 +71,14 @@ export class WorkspaceProxy extends WorkerEntrypoint { override async fetch(request: Request): Promise { const url = new URL(request.url); const egressToken = request.headers.get("x-workspace-egress-token"); - if (url.pathname === "/ws" && egressToken !== null) { + if (url.pathname === "/api" && egressToken !== null) { return Response.json({ callbackUrl: request.url, originalUrl: request.headers.get("x-workspace-egress-url"), @@ -35,7 +35,7 @@ export class TestStorageDO extends DurableObject { body: await request.text(), }); } - if (url.pathname === "/ws") { + if (url.pathname === "/api") { return new Response( url.searchParams.has("token") ? `from-do:${url.searchParams.get("token")}` : "from-do", { status: 200 }, diff --git a/packages/computer/tests/proxy.test.ts b/packages/computer/tests/proxy.test.ts index eb3db15d..95f42e0f 100644 --- a/packages/computer/tests/proxy.test.ts +++ b/packages/computer/tests/proxy.test.ts @@ -3,7 +3,7 @@ // into the DO. Two-and-a-half pieces of logic to pin: // // - /health answers 200 ok\n (the port-readiness probe). -// - /ws looks up env[binding] and forwards the request to the +// - /api looks up env[binding] and forwards the request to the // named DO instance. // - anything else is 404. // @@ -34,27 +34,27 @@ describe("WorkspaceProxy", () => { expect(res.headers.get("content-type")).toMatch(/text\/plain/); }); - it("/ws forwards to the DO at env[binding]", async () => { - const res = await SELF.fetch("http://proxy.test/ws", { + it("/api forwards to the DO at env[binding]", async () => { + const res = await SELF.fetch("http://proxy.test/api", { headers: { "x-test-id": freshId(), "x-test-binding": "COMPUTERD" }, }); expect(res.status).toBe(200); expect(await res.text()).toBe("from-do"); }); - it("accepts tokenized callback health and forwards a normalized tokenized /ws", async () => { + it("accepts tokenized callback health and normalizes the tokenized callback to /api", async () => { const token = "123e4567-e89b-12d3-a456-426614174000"; const health = await SELF.fetch(`http://proxy.test/__workspace_connect/${token}/health`, { headers: { "x-test-id": freshId() }, }); expect(health.status).toBe(200); - const websocket = await SELF.fetch(`http://proxy.test/__workspace_connect/${token}/ws`, { + const websocket = await SELF.fetch(`http://proxy.test/__workspace_connect/${token}/api`, { headers: { "x-test-id": freshId(), "x-test-binding": "COMPUTERD" }, }); expect(await websocket.text()).toBe(`from-do:${token}`); }); - it("routes egress callbacks through /ws while preserving the original request", async () => { + it("routes egress callbacks through /api while preserving the original request", async () => { const res = await SELF.fetch("https://api.example.test/v1/data?format=json", { method: "POST", body: "payload", @@ -66,7 +66,7 @@ describe("WorkspaceProxy", () => { }); expect(await res.json()).toEqual({ - callbackUrl: "https://api.example.test/ws", + callbackUrl: "https://api.example.test/api", originalUrl: "https://api.example.test/v1/data?format=json", egressToken: "secret-token", method: "POST", @@ -74,15 +74,15 @@ describe("WorkspaceProxy", () => { }); }); - it("/ws returns 500 when env[binding] is missing", async () => { - const res = await SELF.fetch("http://proxy.test/ws", { + it("/api returns 500 when env[binding] is missing", async () => { + const res = await SELF.fetch("http://proxy.test/api", { headers: { "x-test-id": freshId(), "x-test-binding": "NOT_A_BINDING" }, }); expect(res.status).toBe(500); expect(await res.text()).toMatch(/NOT_A_BINDING is not a DurableObjectNamespace/); }); - it("returns 404 for paths other than /health and /ws", async () => { + it("returns 404 for paths other than /health and /api", async () => { const res = await SELF.fetch("http://proxy.test/anything-else", { headers: { "x-test-id": freshId() }, }); diff --git a/packages/computerd/README.md b/packages/computerd/README.md index 4d18d696..7ba1088e 100644 --- a/packages/computerd/README.md +++ b/packages/computerd/README.md @@ -28,8 +28,8 @@ Current endpoints: - `GET /__computerd/info` returns JSON with the selected FUSE backend, mount point, and bound port. - `GET /__computerd/stats` returns JSON with DOFS table row counts, total inline and blob byte sizes, the orphan-blob subset, and process resident memory. Useful for watching how the store grows under load. - `GET /` returns `200 OK` with an empty JSON object: `{}`. -- `POST /api` is a capnweb HTTP-batch RPC endpoint backed by `@cloudflare/computer-rpc`. Non-POST methods return `405`. -- `GET /ws` upgrades to a WebSocket carrying the same capnweb RPC surface. This is the container's primary sync carrier. +- `GET /api` upgrades to a WebSocket carrying the capnweb RPC surface backed by `@cloudflare/computer-rpc`. This is the container's only RPC carrier. A request without an `Upgrade` header returns `400`; a handshake naming an unsupported `Sec-WebSocket-Version` returns `426` along with the versions the server speaks. +- `GET /api/watermarks` returns JSON with `currentRev`, `pushRev`, and `fetchCursor`, read through the same `watermarks()` the wire serves. For samplers that want a few numbers without opening a session. It sits under `/api` because it reads the workspace surface; `/__computerd` is for daemon introspection. All other paths and methods return `404`/`405` with a `text/plain` body. @@ -38,9 +38,9 @@ Current filesystem support: - `@platformatic/vfs` in-memory filesystem provided by `@cloudflare/dofs`'s node provider. - FUSE operation adapter covering the full `fuse-native` operation surface. - Unsupported FUSE operations return `ENOSYS` to the kernel; the binding logs a one-shot warning per operation. -- capnweb RPC over `/api` and `/ws` exposes the workspace database and an `exec` runner to clients. -- Optional host/DO synchronization: when `UPSTREAM_URL` is set, `computerd` opens a `SyncClient` from `@cloudflare/computer-rpc/client` against that URL and runs the sync loop in the background. -- No on-disk persistence yet — the in-memory VFS is rebuilt on each start, with sync pulling state back from the upstream when configured. +- capnweb RPC over `/api` exposes the workspace database and an `exec` runner to clients. +- Synchronization is driven by whoever holds the other end of the session. The daemon serves `SyncRPC`; it does not run a sync loop of its own. +- No on-disk persistence yet — the in-memory VFS is rebuilt on each start, and the host pushes state back after a restart. ## FUSE write model @@ -101,14 +101,13 @@ FUSE_MOUNT=auto # default: probe /dev/fuse or macFUSE, fall back to the users FUSE_MOUNT=fuse # require the linux kernel FUSE backend (/dev/fuse) FUSE_MOUNT=macfuse # require macFUSE on darwin FUSE_MOUNT=shim # force the userspace dev shim (no FUSE) -FUSE_MOUNT=none # skip the mount entirely; HTTP + /api + /ws still come up +FUSE_MOUNT=none # skip the mount entirely; HTTP and /api still come up ``` Additional environment variables: ```sh -UPSTREAM_URL=https://example/ws # open a SyncClient against this capnweb endpoint -EXEC_LOG_MAX_BYTES=1048576 # cap the in-memory exec log buffer (bytes) +EXEC_LOG_MAX_BYTES=1048576 # cap the in-memory exec log buffer (bytes) ``` `FUSE_MOUNT=auto` is the friendly default: if `/dev/fuse` (or macFUSE) is available `computerd` mounts a real FUSE filesystem, otherwise it transparently falls back to the userspace shim. Pin the value (`fuse` / `macfuse` / `shim` / `none`) when a test needs to assert a specific code path. @@ -124,7 +123,7 @@ How it works: - A periodic poll (~250 ms) walks `MOUNT_POINT`, diffs it against a content-hash shadow, and pushes any new or changed entries into the VFS. - The shadow doubles as a loop suppressor: after a write in either direction the shadow matches both sides, so the next tick on the opposite side sees no diff. -`exec` runs with `cwd=MOUNT_POINT` exactly as it does under real FUSE, so a child process that writes into the mount point ends up writing through the shim into the VFS — and onward to the DO when `UPSTREAM_URL` is set. +`exec` runs with `cwd=MOUNT_POINT` exactly as it does under real FUSE, so a child process that writes into the mount point ends up writing through the shim into the VFS, and onward to the host on its next pull. Caveats. The shim is dev-only: diff --git a/packages/computerd/src/cli/computerd.test.ts b/packages/computerd/src/cli/computerd.test.ts index f9b38b9d..9c524566 100644 --- a/packages/computerd/src/cli/computerd.test.ts +++ b/packages/computerd/src/cli/computerd.test.ts @@ -105,13 +105,13 @@ test("computerd exposes file IO through real FUSE when FUSE_MOUNT=fuse", async ( expect(await fs.readFile(path.join(mountPoint, "dir", "hello.txt"), "utf8")).toBe("hello fuse"); }); -test("/ws serves a capnweb WorkspaceRPC session", async (_ctx) => { +test("/api serves a capnweb WorkspaceRPC session", async (_ctx) => { const { createWorkspaceClient } = await import("@cloudflare/computer-rpc/client"); const port = await getAvailablePort(); const mountPoint = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-mount-")); await startComputerd({ port, mountPoint, env: { FUSE_MOUNT: "none" } }); - const client = createWorkspaceClient({ url: `ws://127.0.0.1:${port}/ws` }); + const client = createWorkspaceClient({ url: `ws://127.0.0.1:${port}/api` }); try { // hasObjects against a fresh DB returns the empty subset. expect(await client.sync.hasObjects([])).toEqual([]); @@ -133,15 +133,85 @@ test("/ws serves a capnweb WorkspaceRPC session", async (_ctx) => { } }); -test("/api serves a capnweb HTTP-batch WorkspaceRPC session", async (_ctx) => { - const { newHttpBatchRpcSession } = await import("capnweb"); +test("/api refuses anything that is not a websocket handshake", async (_ctx) => { + // /api carries one transport. A caller that arrives over plain HTTP, + // or botches the handshake, should learn that from the status rather + // than from a capnweb error several calls later. const port = await getAvailablePort(); const mountPoint = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-mount-")); await startComputerd({ port, mountPoint, env: { FUSE_MOUNT: "none" } }); + const base = `http://127.0.0.1:${port}/api`; + + // No Upgrade header at all: never reaches the upgrade listener. + const plainGet = await fetch(base); + expect(plainGet.status).toBe(400); + expect(await plainGet.text()).toMatch(/websocket/i); + + const post = await fetch(base, { method: "POST", body: "[]" }); + expect(post.status).toBe(400); + + // An upgrade attempt naming a version we do not speak gets 426 and + // the versions we do speak, per the websocket specification. + const badVersion = await rawRequest(port, [ + "GET /api HTTP/1.1", + `Host: 127.0.0.1:${port}`, + "Upgrade: websocket", + "Connection: Upgrade", + "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==", + "Sec-WebSocket-Version: 7", + ]); + expect(badVersion).toMatch(/^HTTP\/1\.1 426 /); + expect(badVersion).toMatch(/Sec-WebSocket-Version: 13, 8/i); + + // A handshake missing its key is malformed, not a version problem. + const noKey = await rawRequest(port, [ + "GET /api HTTP/1.1", + `Host: 127.0.0.1:${port}`, + "Upgrade: websocket", + "Connection: Upgrade", + "Sec-WebSocket-Version: 13", + ]); + expect(noKey).toMatch(/^HTTP\/1\.1 400 /); + + // A subpath under /api is not the session endpoint: only the exact + // path upgrades. + const subpath = await rawRequest(port, [ + "GET /api/watermarks HTTP/1.1", + `Host: 127.0.0.1:${port}`, + "Upgrade: websocket", + "Connection: Upgrade", + "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==", + "Sec-WebSocket-Version: 13", + ]); + expect(subpath).toMatch(/^HTTP\/1\.1 404 /); + + // An unknown path is still a 404, upgrade header or not. + const unknown = await rawRequest(port, [ + "GET /nope HTTP/1.1", + `Host: 127.0.0.1:${port}`, + "Upgrade: websocket", + "Connection: Upgrade", + "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==", + "Sec-WebSocket-Version: 13", + ]); + expect(unknown).toMatch(/^HTTP\/1\.1 404 /); +}); + +test("/api/watermarks reports sync revisions over plain HTTP", async (_ctx) => { + // Samplers want three numbers on an interval. Opening an RPC session + // per sample is the wrong shape for that. + const port = await getAvailablePort(); + const mountPoint = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-watermarks-")); + await startComputerd({ port, mountPoint, env: { FUSE_MOUNT: "none" } }); - // HTTP batch flushes on first await; each call is a fresh session. - const stub = newHttpBatchRpcSession(`http://127.0.0.1:${port}/api`); - expect(await stub.sync.hasObjects([])).toEqual([]); + const res = await fetch(`http://127.0.0.1:${port}/api/watermarks`); + expect(res.status).toBe(200); + const body = await res.json(); + expect(body).toMatchObject({ + currentRev: expect.any(Number), + pushRev: expect.any(Number), + fetchCursor: { rev: expect.any(Number) }, + }); }); test("/__computerd/stats returns DOFS table sizes and process memory", async (_ctx) => { @@ -222,7 +292,7 @@ test("FUSE_MOUNT=shim materialises an RPC push under the mount point", async (_c const mountPoint = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-shim-push-")); await startComputerd({ port, mountPoint, env: { FUSE_MOUNT: "shim" } }); - const client = createWorkspaceClient({ url: `ws://127.0.0.1:${port}/ws` }); + const client = createWorkspaceClient({ url: `ws://127.0.0.1:${port}/api` }); onTestFinished(() => client.close()); const db = new Database(new SQLiteTestStorage()); @@ -291,8 +361,10 @@ test("/connect re-dial tears down the prior WebSocket session", async (_ctx) => const peerPort = await getAvailablePort(); const opened = []; const peerSockets = new Set(); + // Deliberately non-default paths: the daemon must use what the + // request names, not paths of its own. const peerServer = http.createServer((req, res) => { - if (req.url === "/health") { + if (req.url === "/probe") { res.writeHead(200, { "content-type": "text/plain" }); res.end("ok\n"); return; @@ -305,7 +377,7 @@ test("/connect re-dial tears down the prior WebSocket session", async (_ctx) => }); const wss = new WebSocketServer({ noServer: true }); peerServer.on("upgrade", (req, socket, head) => { - if (req.url !== "/ws") { + if (req.url !== "/socket") { socket.destroy(); return; } @@ -337,7 +409,11 @@ test("/connect re-dial tears down the prior WebSocket session", async (_ctx) => const peerUrl = `http://127.0.0.1:${peerPort}`; const connect = async () => { - const res = await postJson(`http://127.0.0.1:${port}/connect`, { url: peerUrl }); + const res = await postJson(`http://127.0.0.1:${port}/connect`, { + base: peerUrl, + health: "/probe", + api: "/socket", + }); expect(res.statusCode).toBe(200); }; @@ -354,6 +430,49 @@ test("/connect re-dial tears down the prior WebSocket session", async (_ctx) => expect(opened[1].closed).toBe(false, "second peer WS should still be open"); }); +test("/connect rejects a body that does not name every part", async (_ctx) => { + // The daemon assembles no host paths of its own, so a request that + // leaves one out has to be refused rather than defaulted. + const port = await getAvailablePort(); + const mountPoint = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-mount-")); + await startComputerd({ port, mountPoint, env: { FUSE_MOUNT: "none" } }); + const url = `http://127.0.0.1:${port}/connect`; + + const cases = [ + [{}, "empty body"], + [{ health: "/health", api: "/api" }, "no base"], + [{ base: "http://127.0.0.1:1", api: "/api" }, "no health"], + [{ base: "http://127.0.0.1:1", health: "/health" }, "no api"], + [{ base: "ftp://127.0.0.1:1", health: "/health", api: "/api" }, "unsupported scheme"], + [{ base: "http://127.0.0.1:1", health: "health", api: "/api" }, "health lacks a slash"], + [{ base: "http://127.0.0.1:1", health: "/health", api: "api" }, "api lacks a slash"], + [ + { base: "http://127.0.0.1:1", health: "/health", api: "http://elsewhere/api" }, + "api is an absolute address", + ], + [ + { base: "http://127.0.0.1:1", health: "//elsewhere/health", api: "/api" }, + "health is protocol relative", + ], + ]; + + for (const [body, label] of cases) { + const res = await postJson(url, body); + expect(res.statusCode, `${label} should be refused`).toBe(400); + } + + // Control: a well-formed body must clear validation. Port 1 has + // nothing listening, so it gets as far as the health probe and + // fails there instead, which is a different status. + const ok = await postJson(url, { + base: "http://127.0.0.1:1", + health: "/health", + api: "/api", + healthTimeoutMs: 250, + }); + expect(ok.statusCode, "a complete body should reach the health probe").toBe(502); +}); + async function startComputerd({ port, mountPoint, @@ -484,6 +603,34 @@ async function waitFor(predicate, { timeoutMs = 2_000, intervalMs = 10 } = {}) { throw new Error("waitFor: predicate did not become true within the timeout"); } +// Write a request onto a raw socket and return the response head. +// http.request() will not send a malformed websocket handshake, and +// fetch() will not send one at all, so the handshake cases need this. +function rawRequest(port, lines) { + return new Promise((resolve, reject) => { + const socket = net.connect(port, "127.0.0.1", () => { + socket.write(`${lines.join("\r\n")}\r\n\r\n`); + }); + let buf = ""; + socket.setEncoding("utf8"); + socket.on("data", (chunk) => { + buf += chunk; + }); + socket.once("error", reject); + // The server may hold the socket open after refusing the + // handshake, so settle on the end of the response head. + const settle = () => { + socket.destroy(); + resolve(buf); + }; + socket.on("data", () => { + if (buf.includes("\r\n\r\n")) settle(); + }); + socket.once("close", () => resolve(buf)); + setTimeout(settle, 2_000); + }); +} + function postJson(url, body) { const payload = JSON.stringify(body); return new Promise((resolve, reject) => { diff --git a/packages/computerd/src/cli/computerd.ts b/packages/computerd/src/cli/computerd.ts index 2727c6e3..a0e42e95 100644 --- a/packages/computerd/src/cli/computerd.ts +++ b/packages/computerd/src/cli/computerd.ts @@ -5,14 +5,9 @@ import { createServer, type IncomingMessage, type Server, type ServerResponse } import type { Socket } from "node:net"; import { isAbsolute } from "node:path"; import type { ExecEvent as RpcExecEvent } from "@cloudflare/computer-rpc"; -import { createWorkspaceClient, type WorkspaceClient } from "@cloudflare/computer-rpc/client"; import { isStubTrackingEnabled, stubSnapshot } from "@cloudflare/computer-rpc/debug"; import type { RunnerLike } from "@cloudflare/computer-rpc/server"; -import { - acceptWebSocketSession, - createWorkspaceServer, - serveHTTPBatch, -} from "@cloudflare/computer-rpc/server"; +import { acceptWebSocketSession, createWorkspaceServer } from "@cloudflare/computer-rpc/server"; import type { Database } from "@cloudflare/dofs"; import { WebSocket, WebSocketServer } from "ws"; import { Runner } from "../exec/index.js"; @@ -177,34 +172,52 @@ function createHTTPServer( const server = createServer((request, response) => { const path = requestPath(request); - // /api — capnweb HTTP-batch endpoint. Single POST per call; - // request body carries the serialized message, response body - // carries the reply. Useful for environments that can't open - // a WebSocket (curl, fetch from a Worker without ws upgrade). + // /api — the capnweb endpoint. It carries one transport, a + // websocket, so a request that reaches the ordinary handler here + // has no Upgrade header and cannot be served. Say so plainly: + // the alternative is a caller that connects, makes one call that + // appears to work, and then fails on a stream. if (path === "/api") { - if (request.method !== "POST") { - send(response, 405, "method not allowed\n", { - allow: "POST", - "content-type": "text/plain; charset=utf-8", - }); + send(response, 400, "/api requires a websocket upgrade\n", { + "content-type": "text/plain; charset=utf-8", + }); + return; + } + + // /api/watermarks — the same sync revisions the session serves, + // over plain HTTP, for samplers that want a few numbers on an + // interval rather than a session of their own. It reads through + // rpc.sync.watermarks(), so the two cannot drift. Part of the + // workspace API rather than daemon introspection, hence /api + // rather than /__computerd. + if (path === "/api/watermarks") { + if (request.method === "HEAD") { + send(response, 200, "", { "content-type": "application/json; charset=utf-8" }); return; } - void serveHTTPBatch(request, response, rpc).catch((error) => { - console.error("/api batch failed:", error); - if (!response.headersSent) { - send(response, 500, "internal error\n", { - "content-type": "text/plain; charset=utf-8", + void rpc.sync + .watermarks() + .then((watermarks) => { + send(response, 200, JSON.stringify(watermarks), { + "content-type": "application/json; charset=utf-8", }); - } - }); + }) + .catch((error: unknown) => { + console.error("/api/watermarks failed:", error); + if (!response.headersSent) { + send(response, 500, "internal error\n", { + "content-type": "text/plain; charset=utf-8", + }); + } + }); return; } - // /connect — POST { url } where url is the http(s) base of an - // egress endpoint the host wants us to dial back into. We open a - // capnweb WebSocket session against `${url}/ws` and serve our RPC - // over it, exactly like /ws but with the carrier inverted (we - // dial out instead of accepting an inbound upgrade). + // /connect — POST { base, health, api } naming an endpoint the + // host wants us to dial back into. We poll `base + health` until + // it answers, then open a capnweb WebSocket session against + // `base + api` and serve our RPC over it: the same session as an + // inbound upgrade, with the carrier inverted because we dial out. if (path === "/connect") { if (request.method !== "POST") { send(response, 405, "method not allowed\n", { @@ -277,24 +290,39 @@ function createHTTPServer( }); }); - // /ws — capnweb WebSocket endpoint. Long-lived, bidirectional, - // streaming-friendly. The container's primary sync carrier. - // perMessageDeflate compresses each WS frame with zlib. Defaults - // off in the `ws` package; we turn it on so computerd-to-computerd peers - // (and any Node-side client that negotiates the extension) save - // bytes on the wire. Clients that don't advertise the extension - // negotiate down to plain frames, so no flag day for workerd or - // browser callers. + // /api — the capnweb endpoint. Long-lived, bidirectional, and + // streaming-friendly, which is what the wire contract needs: every + // call that reads data back returns a ReadableStream. + // perMessageDeflate compresses each frame with zlib. Defaults off + // in the `ws` package; we turn it on so any Node-side client that + // negotiates the extension saves bytes on the wire. Clients that + // don't advertise it negotiate down to plain frames, so no flag day + // for workerd or browser callers. const wss = new WebSocketServer({ noServer: true, perMessageDeflate: true }); wss.on("connection", (ws) => { acceptWebSocketSession(ws, rpc); }); server.on("upgrade", (request, socket, head) => { - if (requestPath(request) !== "/ws") { + if (requestPath(request) !== "/api") { socket.write("HTTP/1.1 404 Not Found\r\n\r\n"); socket.destroy(); return; } + // `ws` answers a malformed handshake for us: 405 for a method + // other than GET, 400 for a bad Upgrade header or a missing key. + // It answers 400 for an unsupported version too, where the + // websocket specification calls for 426, so that one case is + // handled here before handing over. + const version = Number(request.headers["sec-websocket-version"]); + if (version !== 13 && version !== 8) { + socket.write( + "HTTP/1.1 426 Upgrade Required\r\n" + + "Sec-WebSocket-Version: 13, 8\r\n" + + "Connection: close\r\n\r\n", + ); + socket.destroy(); + return; + } wss.handleUpgrade(request, socket as Socket, head, (ws) => { wss.emit("connection", ws, request); }); @@ -322,16 +350,37 @@ async function closeServer(server: Server): Promise { }); } +// The caller names every part of the endpoint it wants us to reach. +// We assemble `base + health` and `base + api` and hold no opinion +// about either path, so the host is free to rename its own routes +// without a matching release of this binary. interface ConnectBody { - // Base URL of the egress endpoint. ws[s]:// or http[s]://; we - // normalise http(s) to ws(s) and append /ws. - url?: unknown; - // How long to poll the upstream /health before giving up. - // Defaults to 30s; the egress proxy is up at boot but the worker - // that hosts it may take a tick. + // http:// or https:// origin of the endpoint, optionally with a + // path prefix. Trailing slashes are trimmed. + base?: unknown; + // Absolute path of the readiness probe, e.g. "/health". + health?: unknown; + // Absolute path of the capnweb endpoint, e.g. "/api". Dialed as + // ws:// or wss:// according to the scheme on `base`. + api?: unknown; + // How long to poll the readiness probe before giving up. Defaults + // to 30s; the egress proxy is up at boot but the worker that hosts + // it may take a tick. healthTimeoutMs?: unknown; } +// A path we're willing to append to `base`. Rejecting anything that +// parses as an address of its own keeps a caller from redirecting the +// dial somewhere the base never pointed. "//host/path" is +// protocol-relative, which resolves against the base's scheme rather +// than staying under the base, so it goes too. +function isEndpointPath(value: unknown): value is string { + if (typeof value !== "string") return false; + if (!value.startsWith("/")) return false; + if (value.startsWith("//")) return false; + return !URL.canParse(value); +} + async function handleConnect( request: IncomingMessage, response: ServerResponse, @@ -348,22 +397,38 @@ async function handleConnect( return; } - if (typeof body.url !== "string" || body.url.length === 0) { - send(response, 400, "missing 'url' in body\n", { + if ( + typeof body.base !== "string" || + !(body.base.startsWith("http://") || body.base.startsWith("https://")) + ) { + send(response, 400, "'base' must be an http:// or https:// URL\n", { "content-type": "text/plain; charset=utf-8", }); return; } - const baseUrl = body.url.replace(/\/+$/, ""); + if (!isEndpointPath(body.health)) { + send(response, 400, "'health' must be an absolute path such as /health\n", { + "content-type": "text/plain; charset=utf-8", + }); + return; + } + if (!isEndpointPath(body.api)) { + send(response, 400, "'api' must be an absolute path such as /api\n", { + "content-type": "text/plain; charset=utf-8", + }); + return; + } + const baseUrl = body.base.replace(/\/+$/, ""); + const healthUrl = `${baseUrl}${body.health}`; const healthTimeoutMs = typeof body.healthTimeoutMs === "number" && body.healthTimeoutMs > 0 ? body.healthTimeoutMs : 30_000; try { - await waitForHealth(baseUrl, healthTimeoutMs); + await waitForHealth(healthUrl, healthTimeoutMs); } catch (error) { - send(response, 502, `upstream /health unreachable: ${(error as Error).message}\n`, { + send(response, 502, `readiness probe unreachable: ${(error as Error).message}\n`, { "content-type": "text/plain; charset=utf-8", }); return; @@ -383,7 +448,7 @@ async function handleConnect( } } - const wsUrl = `${toWebSocketUrl(baseUrl)}/ws`; + const wsUrl = `${toWebSocketUrl(baseUrl)}${body.api}`; const ws = new WebSocket(wsUrl); upstreamSlot.ws = ws; ws.once("open", () => { @@ -414,15 +479,13 @@ async function readJson(request: IncomingMessage): Promise { } function toWebSocketUrl(input: string): string { - if (input.startsWith("ws://") || input.startsWith("wss://")) return input; if (input.startsWith("http://")) return `ws://${input.slice("http://".length)}`; if (input.startsWith("https://")) return `wss://${input.slice("https://".length)}`; throw new Error(`unsupported URL scheme: ${input}`); } -async function waitForHealth(baseUrl: string, timeoutMs: number): Promise { +async function waitForHealth(healthUrl: string, timeoutMs: number): Promise { const deadline = Date.now() + timeoutMs; - const healthUrl = `${toHttpUrl(baseUrl)}/health`; let lastError: unknown; while (Date.now() < deadline) { try { @@ -439,12 +502,6 @@ async function waitForHealth(baseUrl: string, timeoutMs: number): Promise ); } -function toHttpUrl(input: string): string { - if (input.startsWith("ws://")) return `http://${input.slice("ws://".length)}`; - if (input.startsWith("wss://")) return `https://${input.slice("wss://".length)}`; - return input; -} - async function main(): Promise { // Install logging + crash handlers first thing so any early throws // (parsePort, parseMountPoint, resolveFuseBackend) still land in @@ -458,27 +515,13 @@ async function main(): Promise { // FUSE_MOUNT picks the backend. auto (default) probes /dev/fuse // or macFUSE and falls back to the userspace shim. fuse / macfuse // require their respective real backend. shim forces the userspace - // polling shim. none skips the mount entirely; HTTP + /api + /ws + // polling shim. none skips the mount entirely; HTTP and /api // still come up so tests and tooling can talk to computerd's RPC surface. const fuseMountMode = parseFuseMountMode(process.env.FUSE_MOUNT); const backend: FUSEBackend = await resolveFuseBackend(fuseMountMode); console.log(`[info] FUSE_MOUNT=${fuseMountMode} resolved to backend=${backend.kind}`); - const upstreamUrl = process.env.UPSTREAM_URL?.trim(); - let upstreamClient: WorkspaceClient | undefined; - if (upstreamUrl !== undefined && upstreamUrl.length > 0) { - // Use the `ws` package's WebSocket (not Node's built-in - // global) so the dial negotiates permessage-deflate against - // the upstream's WebSocketServer. Node 22's built-in - // WebSocket doesn't advertise the deflate extension. - upstreamClient = createWorkspaceClient({ - url: upstreamUrl, - WebSocketImpl: WebSocket as unknown as typeof globalThis.WebSocket, - }); - } - const { vfs, db, stopSync } = await createNodeVirtualFileSystem({ - upstream: upstreamClient?.sync, - }); + const { vfs, db } = await createNodeVirtualFileSystem(); const info: ComputerdInfo = { backend, mountPoint, port }; let fuse: FuseMount | undefined; @@ -584,14 +627,6 @@ async function main(): Promise { if (fuse !== undefined) { await unmount(fuse); } - if (upstreamClient !== undefined) { - try { - stopSync(); - await upstreamClient.close(); - } catch (error) { - console.error(error); - } - } teardownLogging(); process.exit(signal === "SIGINT" ? 130 : 143); }; diff --git a/packages/computerd/src/exec/runner.fuse.test.ts b/packages/computerd/src/exec/runner.fuse.test.ts index da5d2e8c..23b85877 100644 --- a/packages/computerd/src/exec/runner.fuse.test.ts +++ b/packages/computerd/src/exec/runner.fuse.test.ts @@ -107,7 +107,7 @@ describeIfReal("Runner shell.exec under real FUSE", () => { test("exec(cwd inside the FUSE mount) returns quickly and runs the command", async () => { if (!url) throw new Error("computerd container did not start"); const client = createWorkspaceClient({ - url: `${url.replace(/^http(s?):\/\//, "ws$1://")}/ws`, + url: `${url.replace(/^http(s?):\/\//, "ws$1://")}/api`, WebSocketImpl: WebSocket, }); try { diff --git a/packages/computerd/src/fuse/backend.ts b/packages/computerd/src/fuse/backend.ts index 5f90eb77..3d4f5cf3 100644 --- a/packages/computerd/src/fuse/backend.ts +++ b/packages/computerd/src/fuse/backend.ts @@ -8,7 +8,7 @@ import { access as defaultAccess } from "node:fs/promises"; // - fuse : require the linux kernel FUSE backend. // - macfuse: require macFUSE on darwin. // - shim : force the userspace shim. Works on any platform. -// - none : skip the mount entirely; HTTP + /api + /ws still come up. +// - none : skip the mount entirely; HTTP and /api still come up. export type FuseMountMode = "auto" | "fuse" | "macfuse" | "shim" | "none"; // The resolved backend selection. No `reason` field — the choice is diff --git a/packages/computerd/src/fuse/vfs.test.ts b/packages/computerd/src/fuse/vfs.test.ts index c6111f96..43ac6fbb 100644 --- a/packages/computerd/src/fuse/vfs.test.ts +++ b/packages/computerd/src/fuse/vfs.test.ts @@ -1,4 +1,3 @@ -import { createHash } from "node:crypto"; import { expect, test } from "vitest"; import { createNodeVirtualFileSystem } from "./index.js"; @@ -20,54 +19,3 @@ test("createNodeVirtualFileSystem returns a @platformatic/vfs filesystem", async vfs.unlinkSync("/project/greeting.txt"); expect(vfs.readdirSync("/project")).toEqual([]); }); - -test("createNodeVirtualFileSystem pulls initial state from an upstream SyncRPC", async () => { - const bytes = Buffer.from("hi"); - const hash = new Uint8Array(createHash("sha256").update(bytes).digest()); - - let fetchChangesCalls = 0; - const upstream = { - async fetchChanges() { - fetchChangesCalls++; - return { - currentCursor: { rev: 1, path: null }, - appliedPushCursor: { rev: 0, path: null }, - stream: new ReadableStream({ - start(c) { - c.enqueue({ - kind: "file", - rev: 1, - path: "/hi.txt", - mode: 0o644, - mtime: 100, - size: 2, - chunks: [{ hash, size: 2 }], - }); - c.close(); - }, - }), - }; - }, - async hasObjects(hashes) { - // The fake upstream is the source of truth for this file's - // chunk. Reply that we have every hash the client probes. - return hashes; - }, - async fetchObjects(hashes) { - return new ReadableStream({ - start(c) { - for (const h of hashes) c.enqueue({ hash: h, bytes }); - c.close(); - }, - }); - }, - async push() { - return { rev: 0, appliedPushCursor: { rev: 0, path: null } }; - }, - async pushObjects() {}, - }; - - const { vfs } = await createNodeVirtualFileSystem({ upstream }); - expect(fetchChangesCalls).toBe(1); - expect(vfs.readFileSync("/hi.txt").toString()).toBe("hi"); -}); diff --git a/packages/computerd/src/fuse/vfs.ts b/packages/computerd/src/fuse/vfs.ts index 432eeff1..c4023be7 100644 --- a/packages/computerd/src/fuse/vfs.ts +++ b/packages/computerd/src/fuse/vfs.ts @@ -1,5 +1,3 @@ -import type { SyncRPC } from "@cloudflare/computer-rpc"; -import { pullOnce, tick } from "@cloudflare/computer-rpc/driver"; import { Database, initializeSchema, SQLiteWorkspaceProvider } from "@cloudflare/dofs"; import { SQLiteTestStorage } from "@cloudflare/dofs/testing"; import { create, type VirtualFileSystem, VirtualProvider } from "@platformatic/vfs"; @@ -51,54 +49,24 @@ const EXTRA_VFS_METHODS = [ "releaseWriteBufferSync", ] as const; -export interface CreateOptions { - // Optional upstream sync surface. When set, the local store - // performs an initial pull on construction. When unset, computerd runs - // standalone against an in-memory store. - // - // The caller owns the carrier (WebSocket, in-process direct - // binding, or any future flavour). This package only needs the - // typed surface; the transport seam lives in computer-rpc. - // Future RPCs (exec, mounts, watchers) will travel beside - // SyncRPC on the same connection, so the caller may pass a - // composite stub — we accept the narrow SyncRPC subset - // structurally. - upstream?: SyncRPC; -} - export interface NodeVfsHandle { // @platformatic/vfs filesystem the FUSE driver consumes. vfs: NodeVirtualFileSystem; - // dofs Database backing the same store. Exposed so the - // CLI can construct a createSyncServer(db) and serve the local - // store to upstream callers over capnweb. + // dofs Database backing the same store. Exposed so the CLI can + // construct a createWorkspaceServer(db) and serve the local store + // to whoever holds the capnweb session. db: Database; - // Stop the periodic sync loop, if one was started. No-op when - // no upstream was provided. Idempotent. - stopSync: () => void; } -// Polling cadence for the background sync loop. Picked to match -// human-typing latency expectations without saturating the wire. -const SYNC_TICK_MS = 250; - -export async function createNodeVirtualFileSystem( - options: CreateOptions = {}, -): Promise { +// The store is local and process-lifetime. Sync is driven from the +// other end of the capnweb session: the host pushes changes in and +// pulls them back out, so nothing here polls. +export async function createNodeVirtualFileSystem(): Promise { ensureVirtualProviderPrototype(); const storage = new SQLiteTestStorage(); const db = new Database(storage); initializeSchema(db, () => Date.now()); - let stopSync = () => {}; - if (options.upstream !== undefined) { - // Initial pull. The polling loop would catch up eventually, - // but the FUSE mount comes up populated by waiting for the - // first pull to settle before returning. - await pullOnce(db, options.upstream); - stopSync = startSyncLoop(db, options.upstream); - } - const provider = new SQLiteWorkspaceProvider(db); const vfs = create(provider as unknown as VirtualProvider, { moduleHooks: false }); // Forward the extra dofs methods that @platformatic/vfs's @@ -117,27 +85,5 @@ export async function createNodeVirtualFileSystem( configurable: true, }); } - return { vfs, db, stopSync }; -} - -// Drive tick(db, upstream) on a setInterval. Errors during a tick -// log and continue — a transient upstream failure shouldn't -// kill the daemon. The watermarks are durable, so the next tick -// resumes from where the failed one would have. -function startSyncLoop(db: Database, upstream: SyncRPC): () => void { - let stopped = false; - const handle = setInterval(() => { - if (stopped) return; - tick(db, upstream).catch((error) => { - console.error("sync tick failed:", error); - }); - }, SYNC_TICK_MS); - // Don't block process exit on the timer. computerd's shutdown path - // calls stopSync() explicitly; this is belt-and-braces. - handle.unref?.(); - return () => { - if (stopped) return; - stopped = true; - clearInterval(handle); - }; + return { vfs, db }; } diff --git a/packages/rpc/package.json b/packages/rpc/package.json index c3eb78a2..518c59a5 100644 --- a/packages/rpc/package.json +++ b/packages/rpc/package.json @@ -35,7 +35,7 @@ }, "dependencies": { "@cloudflare/dofs": "*", - "capnweb": "^0.8.0" + "capnweb": "^0.10.0" }, "devDependencies": { "@cloudflare/workers-types": "^4.20260616.1", diff --git a/packages/rpc/src/client.ts b/packages/rpc/src/client.ts index e4fef044..300ff479 100644 --- a/packages/rpc/src/client.ts +++ b/packages/rpc/src/client.ts @@ -15,7 +15,7 @@ export interface RPCEvent { } export interface ClientOptions { - // WebSocket URL. Typically ws://container-host:45678/ws. + // WebSocket URL. Typically ws://container-host:45678/api. url: string; // Optional WebSocket constructor. Defaults to the global // WebSocket (node 22+ ships one; older runtimes can pass the diff --git a/packages/rpc/src/server.ts b/packages/rpc/src/server.ts index 9d0f23b8..ebe7c467 100644 --- a/packages/rpc/src/server.ts +++ b/packages/rpc/src/server.ts @@ -21,7 +21,7 @@ import { stageBlob, writeFetchCursor, } from "@cloudflare/dofs"; -import { newWebSocketRpcSession, nodeHttpBatchRpcResponse, RpcTarget } from "capnweb"; +import { newWebSocketRpcSession, RpcTarget } from "capnweb"; import { trackStub, untrackStub } from "./debug.js"; import type { ExecEvent, ShellRPC, SyncRPC, WorkspaceRPC } from "./interface.js"; @@ -323,7 +323,7 @@ export function createShellServer(runner: RunnerLike): ShellRPC { } // Construct the composite WorkspaceRPC. The wire serves this on -// /ws so clients reach `.sync` and `.shell` through one session. +// /api so clients reach `.sync` and `.shell` through one session. export function createWorkspaceServer( db: Database, runner: RunnerLike, @@ -349,18 +349,6 @@ export function acceptWebSocketSession( newWebSocketRpcSession(ws as unknown as WebSocket, rpc as unknown as RpcTarget); } -// Serve a single capnweb HTTP-batch session against a SyncRPC. Wraps -// capnweb's nodeHttpBatchRpcResponse so computerd never directly imports -// capnweb (which would split capnweb's module identity in mixed -// ESM/CJS contexts — the RpcTarget instanceof check then fails). -export function serveHTTPBatch( - request: import("node:http").IncomingMessage, - response: import("node:http").ServerResponse, - rpc: SyncRPC | ShellRPC | WorkspaceRPC, -): Promise { - return nodeHttpBatchRpcResponse(request, response, rpc as unknown as RpcTarget); -} - function iterableToReadableStream(it: AsyncIterable): ReadableStream { const iterator = it[Symbol.asyncIterator](); return new ReadableStream({ diff --git a/script/computerd-fuse-flush.mjs b/script/computerd-fuse-flush.mjs index 420cf3c7..e5d1e7b4 100755 --- a/script/computerd-fuse-flush.mjs +++ b/script/computerd-fuse-flush.mjs @@ -138,9 +138,12 @@ async function main() { // 2. Write through FUSE. echo > triggers FUSE create, write, // flush (on close), release \u2014 every spill point landed in // the 68407fc fix. + // The VFS keys everything under the mount point, so the path the + // receiver reads back is the absolute in-container path, not /x.txt. + const target = "/workspace/x.txt"; const payload = `from-fuse ${Date.now()}\n`; - await dockerExec(container.cid, "bash", "-c", `printf '%s' '${payload}' > /workspace/x.txt`); - process.stderr.write(` wrote ${payload.trim()} to /workspace/x.txt via FUSE\n`); + await dockerExec(container.cid, "bash", "-c", `printf '%s' '${payload}' > ${target}`); + process.stderr.write(` wrote ${payload.trim()} to ${target} via FUSE\n`); // 3. Pull from the container's WebSocket on the host. Mirrors // what the host DO does after exec returns. @@ -157,7 +160,7 @@ async function main() { `${REPO_ROOT}/node_modules/@cloudflare/computer-rpc/dist/sync-driver.js` ); - const wsUrl = `${container.url.replace("http://", "ws://")}/ws`; + const wsUrl = `${container.url.replace("http://", "ws://")}/api`; const client = createWorkspaceClient({ url: wsUrl }); const recvStorage = new SQLiteTestStorage(); @@ -165,16 +168,16 @@ async function main() { initializeSchema(recvDb, () => Date.now()); try { - const applied = await pullOnce(recvDb, client.sync); + const { applied } = await pullOnce(recvDb, client.sync); process.stderr.write(` pullOnce applied ${applied} entries\n`); if (applied === 0) { - throw new Error("pullOnce applied 0 entries; expected at least 1 for /x.txt"); + throw new Error(`pullOnce applied 0 entries; expected at least 1 for ${target}`); } // 4. Read the file back through the receiver-side provider. // The bytes have to match what we wrote through FUSE. const provider = new SQLiteWorkspaceProvider(recvDb, { now: () => Date.now() }); - const back = provider.readFileSync("/x.txt", "utf8"); + const back = provider.readFileSync(target, "utf8"); if (back !== payload) { throw new Error( `byte mismatch:\n wrote: ${JSON.stringify(payload)}\n read: ${JSON.stringify(back)}`, diff --git a/script/computerd-soak.mjs b/script/computerd-soak.mjs index fbbaf019..bf30822d 100755 --- a/script/computerd-soak.mjs +++ b/script/computerd-soak.mjs @@ -1,37 +1,48 @@ #!/usr/bin/env node -// computerd-soak.mjs — soak test for the computerd sync loop. +// computerd-soak.mjs — soak test for the sync loop between a host and +// one computerd. // -// Boots two computerd containers wired as peer-to-peer: -// A: standalone computerd, port mapped to the host. -// B: standalone computerd, port mapped to the host AND -// UPSTREAM_URL pointing at A's host port. B's sync -// loop pulls from A and pushes to A. +// Boots one computerd container and plays the part the durable object +// plays in production: this script holds the authoritative store and +// drives sync across the capnweb session. Files are written into the +// host store at a steady rate while a tick loop pushes them to the +// daemon. While that runs, sample: // -// Hammers A by writing files through its /api endpoint -// (capnweb HTTP batch). While the writes flow, sample: +// - the host store's currentRev — how far the writer has got. +// - the daemon's currentRev — how far sync has carried it. +// - docker stats — resident memory for the container. // -// - A.watermarks() — currentRev, pushRev, fetchRev. -// - B.watermarks() — same. -// - docker stats — RSS for both containers. +// The gap between the two revision columns is the convergence lag. The +// memory column is the other reason this script exists: it is the +// cheapest way to watch the daemon's footprint under write pressure. // -// Output is a TSV table on stdout, one row per sample, -// suitable for piping into a CSV reader or just eyeballing. +// This used to boot a second daemon and point UPSTREAM_URL at the +// first. That measured a daemon driving its own sync loop, which is a +// mode no deployment uses and which no longer exists. A workspace pairs +// one host with one container, so that is what this soaks. +// +// Output is a TSV table on stdout, one row per sample, suitable for +// piping into a CSV reader or just eyeballing. // // Knobs (env vars): // -// COMPUTERD_BINARY path to computerd-linux-x64 binary +// COMPUTERD_BINARY path to computerd-linux-x64 binary // SOAK_DURATION_MS total wall time of the soak phase (default 30000) -// SOAK_WRITES_PER_S target writes/second sustained against A (default 200) +// SOAK_WRITES_PER_S target writes/second sustained (default 200) // SOAK_PAYLOAD_B bytes per write (default 64) // SOAK_SAMPLE_MS sampling interval (default 250) +// SOAK_TICK_MS sync tick interval (default 100) import { execFile, spawn } from "node:child_process"; import { existsSync } from "node:fs"; import { dirname, resolve } from "node:path"; import { fileURLToPath } from "node:url"; import { promisify } from "node:util"; -import { newHttpBatchRpcSession, newWebSocketRpcSession } from "capnweb"; +import { createWorkspaceClient } from "@cloudflare/computer-rpc/client"; +import { pullOnce, pushOnce } from "@cloudflare/computer-rpc/driver"; +import { currentRev, Database, initializeSchema, SQLiteWorkspaceProvider } from "@cloudflare/dofs"; +import { SQLiteTestStorage } from "@cloudflare/dofs/testing"; import WebSocket from "ws"; const execFileP = promisify(execFile); @@ -55,10 +66,11 @@ const SAMPLE_MS = Number(process.env.SOAK_SAMPLE_MS ?? 250); // skipped. const DISABLE_FUSE = process.env.SOAK_DISABLE_FUSE === "1" || !existsSync("/dev/fuse"); -// On Linux, host.docker.internal isn't resolved automatically; -// we map it to the host-gateway address so B can reach A. -// docker-desktop on macOS/Windows already provides this. -const ADD_HOST_GATEWAY = process.platform === "linux"; +// Matches MOUNT_POINT in the container env below. The store keeps +// everything under the mount point so pushed paths land where the FUSE +// mount expects them. +const MOUNT_POINT = "/workspace"; +const TICK_MS = Number(process.env.SOAK_TICK_MS ?? 100); const IMAGE_TAG = "computerd-harness:libfuse2"; @@ -90,7 +102,7 @@ RUN apt-get update >/dev/null && apt-get install -y --no-install-recommends \\ }); } -async function bootContainer(extraEnv = {}) { +async function bootContainer() { const args = ["run", "--rm", "-d", "--platform", "linux/amd64"]; if (!DISABLE_FUSE) { args.push( @@ -107,9 +119,6 @@ async function bootContainer(extraEnv = {}) { "seccomp=unconfined", ); } - if (ADD_HOST_GATEWAY) { - args.push("--add-host", "host.docker.internal:host-gateway"); - } args.push( "-v", `${BINARY}:/usr/local/bin/computerd:ro`, @@ -118,14 +127,11 @@ async function bootContainer(extraEnv = {}) { "-e", "PORT=8080", "-e", - "MOUNT_POINT=/workspace", + `MOUNT_POINT=${MOUNT_POINT}`, ); if (DISABLE_FUSE) { args.push("-e", "FUSE_MOUNT=none"); } - for (const [k, v] of Object.entries(extraEnv)) { - args.push("-e", `${k}=${v}`); - } const image = DISABLE_FUSE ? "debian:stable-slim" : IMAGE_TAG; args.push(image, "/usr/local/bin/computerd"); const { stdout } = await execFileP("docker", args); @@ -174,16 +180,12 @@ async function dockerStats(cids) { return result; } -// Connect a capnweb HTTP batch client to /api. Each call is -// its own session; we accept the cost (one round-trip per -// hammer-write) so the soak measures the entire stack -// including session setup. -function batchStub(url) { - return newHttpBatchRpcSession(`${url}/api`); -} - +// The daemon's revision numbers over plain HTTP. A sampler wanting +// three integers on an interval does not need an RPC session. async function fetchWatermarks(url) { - return await batchStub(url).sync.watermarks(); + const res = await fetch(`${url}/api/watermarks`); + if (!res.ok) throw new Error(`watermarks HTTP ${res.status}`); + return await res.json(); } // Build a payload-bytes Uint8Array. @@ -207,149 +209,109 @@ function payloadBytes(seed) { return out; } -// Persistent capnweb WebSocket session against B. Reused -// across the soak; one upgrade, many push round-trips. +// Persistent capnweb session against the daemon. Reused across the +// soak: one upgrade, many push round-trips. // -// SOAK_NO_DEFLATE=1 forces the dial to negotiate without -// permessage-deflate so the soak can compare compressed and -// uncompressed wire costs without rebuilding the computerd binary. -function wsStub(url) { - const wsUrl = `${url.replace("http://", "ws://")}/ws`; - const ws = new WebSocket(wsUrl, { - perMessageDeflate: process.env.SOAK_NO_DEFLATE !== "1", +// SOAK_NO_DEFLATE=1 dials without permessage-deflate so the soak can +// compare compressed and uncompressed wire costs without rebuilding +// the computerd binary. +function openSession(url) { + return createWorkspaceClient({ + url: `${url.replace("http://", "ws://")}/api`, + WebSocketImpl: class extends WebSocket { + constructor(target) { + super(target, { perMessageDeflate: process.env.SOAK_NO_DEFLATE !== "1" }); + } + }, }); - return newWebSocketRpcSession(ws); } -// One write into computerd via the SyncRPC push path. senderRev=0 -// marks us as an external writer — the server applies as -// a local write (bumps vfs_meta.rev, leaves pushRev alone) so -// the outbound sync loop picks the entry up on the next tick. -// This is the F2 fix in practice; before it landed, this -// path silenced the outbound loop and A never saw any of B's -// writes. -async function pushOneWrite(stub, i, bytes) { - const hash = new Uint8Array(await crypto.subtle.digest("SHA-256", bytes)); - await stub.sync.pushObjects( - new ReadableStream({ - start(c) { - c.enqueue({ hash, bytes }); - c.close(); - }, - }), - ); - await stub.sync.push({ - senderRev: 0, - changes: new ReadableStream({ - start(c) { - c.enqueue({ - kind: "file", - path: `/soak_${i}.bin`, - mode: 0o644, - mtime: Date.now(), - size: bytes.byteLength, - chunks: [{ hash, size: bytes.byteLength }], - }); - c.close(); - }, - }), - }); +// The authoritative store, standing in for the durable object's SQLite. +function openHostStore() { + const db = new Database(new SQLiteTestStorage()); + initializeSchema(db, () => Date.now()); + return { db, provider: new SQLiteWorkspaceProvider(db, { now: () => Date.now() }) }; +} + +// One write into the host store. The sync tick below is what carries +// it to the daemon, so this is a plain filesystem write rather than a +// wire call. +function writeOne(provider, i, bytes) { + provider.writeFileSync(`${MOUNT_POINT}/soak_${i}.bin`, Buffer.from(bytes)); } async function main() { if (!DISABLE_FUSE) await ensureImage(); - process.stderr.write("booting A (sink) ...\n"); - const a = await bootContainer(); - process.stderr.write(` A: ${a.url} (${a.cid.slice(0, 12)})\n`); + process.stderr.write("booting computerd ...\n"); + const daemon = await bootContainer(); + process.stderr.write(` ${daemon.url} (${daemon.cid.slice(0, 12)})\n`); - process.stderr.write("booting B (source, UPSTREAM_URL -> A) ...\n"); - // B's sync loop will push to A. Hostname inside docker: - // we can't reach the host's 127.0.0.1 portably; use - // host.docker.internal which docker-desktop sets up on - // macOS/Windows. On linux we'd need --add-host=host.docker.internal:host-gateway. - // Inside a docker container we reach the host's mapped - // port via host.docker.internal. The capnweb client needs - // a ws:// URL pointing at the /ws endpoint (not just the - // host). - const upstreamForB = `${a.url.replace("http://127.0.0.1", "ws://host.docker.internal")}/ws`; - // Inside the docker container, the host port we're trying - // to reach is 127.0.0.1: on the host. Pass the - // mapped host port via host.docker.internal. - const b = await bootContainer({ - UPSTREAM_URL: upstreamForB, - }); - process.stderr.write(` B: ${b.url} (${b.cid.slice(0, 12)}) -> upstream ${upstreamForB}\n`); + const { db, provider } = openHostStore(); + provider.mkdirSync(MOUNT_POINT, { recursive: true }); + const session = openSession(daemon.url); - // Header row. - console.log( - "t_ms\tA_currentRev\tA_pushRev\tA_fetchRev\tB_currentRev\tB_pushRev\tB_fetchRev\tA_mem\tB_mem\twrites_sent", - ); + // Header row. host_rev is how far the writer has got; daemon_rev is + // how far sync has carried it. The difference is the lag. + console.log("t_ms\thost_rev\tdaemon_rev\tdaemon_pushRev\tmem\twrites_sent\tpushed"); const start = Date.now(); const stopAt = start + DURATION_MS; const intervalMs = Math.max(1, Math.floor(1000 / WRITES_PER_S)); let writeSeq = 0; let writesSent = 0; - let writesInFlight = 0; - // Writes go through B's SyncRPC /ws push path with - // senderRev=0. The server treats them as local writes; - // B's outbound sync loop ships them to A on the next - // tick. This is the path an external orchestrator (a - // DO accepting agent requests, the agent itself) would - // take — the same wire surface a computerd-to-computerd peer - // uses, just with a different senderRev value. - const writeStub = wsStub(b.url); - - // Fire-and-forget write loop. We don't await every write - // because the goal is to saturate; we cap the in-flight - // count to keep memory bounded. - const MAX_INFLIGHT = 32; + // Writes land in the host store. The sync tick below carries them + // over the wire, which is the direction production runs: the host + // owns the truth and pushes it to the container. const writeLoop = (async () => { while (Date.now() < stopAt) { - if (writesInFlight >= MAX_INFLIGHT) { - await new Promise((r) => setTimeout(r, 1)); - continue; - } - const i = writeSeq++; - writesInFlight++; - pushOneWrite(writeStub, i, payloadBytes(i)) - .then(() => { - writesSent++; - writesInFlight--; - }) - .catch((err) => { - writesInFlight--; - process.stderr.write(`write ${i} failed: ${err.message}\n`); - }); + const seq = writeSeq++; + writeOne(provider, seq, payloadBytes(seq)); + writesSent++; await new Promise((r) => setTimeout(r, intervalMs)); } })(); - // Sample loop. Runs in parallel with the writes. + // Sync tick. pushOnce ships whatever the writer has committed since + // the last tick; pullOnce brings back anything the daemon changed on + // its own, which is nothing here but exercises the other direction. + // A failed tick logs and continues: watermarks are durable, so the + // next tick resumes where this one stopped. + let pushed = 0; + let ticking = false; + const tickTimer = setInterval(() => { + if (ticking) return; + ticking = true; + (async () => { + pushed += await pushOnce(db, session.sync); + await pullOnce(db, session.sync); + })() + .catch((err) => { + process.stderr.write(`sync tick failed: ${err.message}\n`); + }) + .finally(() => { + ticking = false; + }); + }, TICK_MS); + + // Sample loop. Runs in parallel with the writes and the tick. const samples = []; const sampleLoop = (async () => { while (Date.now() < stopAt + 5000) { const t = Date.now() - start; - const [aWm, bWm, stats] = await Promise.all([ - fetchWatermarks(a.url).catch(() => null), - fetchWatermarks(b.url).catch(() => null), - dockerStats([a.cid, b.cid]).catch(() => ({})), + const [wm, stats] = await Promise.all([ + fetchWatermarks(daemon.url).catch(() => null), + dockerStats([daemon.cid]).catch(() => ({})), ]); - const aMem = stats[a.cid.slice(0, 12)] ?? "?"; - const bMem = stats[b.cid.slice(0, 12)] ?? "?"; const row = [ t, - aWm?.currentRev ?? -1, - aWm?.pushRev ?? -1, - aWm?.fetchRev ?? -1, - bWm?.currentRev ?? -1, - bWm?.pushRev ?? -1, - bWm?.fetchRev ?? -1, - aMem, - bMem, + currentRev(db), + wm?.currentRev ?? -1, + wm?.pushRev ?? -1, + stats[daemon.cid.slice(0, 12)] ?? "?", writesSent, + pushed, ]; console.log(row.join("\t")); samples.push(row); @@ -358,31 +320,22 @@ async function main() { })(); await writeLoop; - process.stderr.write(`writes done (${writesSent} sent, ${writesInFlight} still in flight)\n`); - // Let in-flight writes drain. - while (writesInFlight > 0) await new Promise((r) => setTimeout(r, 100)); - // Let the sync loop catch up. + process.stderr.write(`writes done (${writesSent} committed locally)\n`); + // Let the tick drain what the writer just committed. await new Promise((r) => setTimeout(r, 3000)); await sampleLoop; - - // capnweb's WebSocket session doesn't expose an explicit - // close on the stub; the container kill below tears it - // down at the computerd end. - await Promise.all([kill(a.cid), kill(b.cid)]); + clearInterval(tickTimer); + await session.close().catch(() => {}); + await kill(daemon.cid); // Summary on stderr. const final = samples[samples.length - 1] ?? []; process.stderr.write(`\n--- soak complete ---\n`); - process.stderr.write(`writes attempted: ${writeSeq}\n`); - process.stderr.write(`writes acked: ${writesSent}\n`); - process.stderr.write(`final B.currentRev: ${final[4]}\n`); - process.stderr.write( - `final B.pushRev: ${final[5]} (gap to currentRev: ${final[4] - final[5]})\n`, - ); - process.stderr.write( - `final A.fetchRev: ${final[3]} (lag behind B.currentRev: ${final[4] - final[3]})\n`, - ); - process.stderr.write(`final B mem: ${final[8]}\n`); + process.stderr.write(`writes committed: ${writeSeq}\n`); + process.stderr.write(`entries pushed: ${pushed}\n`); + process.stderr.write(`final host rev: ${final[1]}\n`); + process.stderr.write(`final daemon rev: ${final[2]} (lag: ${final[1] - final[2]})\n`); + process.stderr.write(`final mem: ${final[4]}\n`); } main().catch((err) => { diff --git a/script/computerd-stub-soak.mjs b/script/computerd-stub-soak.mjs index 2bfedc34..857cfe6d 100755 --- a/script/computerd-stub-soak.mjs +++ b/script/computerd-stub-soak.mjs @@ -236,7 +236,7 @@ async function main() { await waitForHealth(port, child); console.error("[soak] computerd healthy"); - const ws = new WebSocket(`ws://127.0.0.1:${port}/ws`); + const ws = new WebSocket(`ws://127.0.0.1:${port}/api`); await new Promise((res, rej) => { ws.once("open", res); ws.once("error", rej); diff --git a/script/exec-tests b/script/exec-tests index 05217693..a9d77d7e 100755 --- a/script/exec-tests +++ b/script/exec-tests @@ -13,7 +13,7 @@ # - artifacts/computerd/computerd-linux-x64 (run `npm run build:bin --workspace # @cloudflare/computerd` first) # - docker -# - node 22+ on the host +# - node on the host, recent enough for module.registerHooks (22.15+) set -euo pipefail repo_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" @@ -61,9 +61,12 @@ if ! curl -sf "http://localhost:${port}/health" >/dev/null 2>&1; then fi echo "==> computerd healthy on :${port}" -# Drive the smoke from a node script. Inline so we don't add a -# new file to mount; the script is self-contained. -PORT="${port}" REPO_ROOT="${repo_root}" node --input-type=module -e ' +# Drive the smoke from a node script, inline so nothing extra has to be +# mounted into the container. The --import hook makes the package's main +# entry loadable outside workerd; see the note at its use below. +PORT="${port}" REPO_ROOT="${repo_root}" node \ + --import "${repo_root}/script/lib/cloudflare-workers-stub.mjs" \ + --input-type=module -e ' const repoRoot = process.env.REPO_ROOT; const port = process.env.PORT; const { createWorkspaceClient } = await import( @@ -77,7 +80,7 @@ const assert = (cond, msg) => { } }; -const client = createWorkspaceClient({ url: `ws://localhost:${port}/ws` }); +const client = createWorkspaceClient({ url: `ws://localhost:${port}/api` }); try { // 1. echo + exit code. const h = await client.shell.exec({ source: "echo hello && exit 7" }); @@ -234,15 +237,12 @@ try { // 7. Workspace.runtime host wrapper (E3). Uses the composite // client through the high-level facade: encoding utf8 + // ExecHandle.result() in one call. - // Import the specific files rather than the package barrel. - // The barrel re-exports WorkspaceProxy, which extends - // WorkerEntrypoint from cloudflare:workers — that scheme - // only resolves under workerd, not plain Node. - const { TestBackend } = await import( - `${repoRoot}/packages/computer/dist/backends/test.js` - ); - const { Workspace } = await import( - `${repoRoot}/packages/computer/dist/workspace.js` + // The package barrel re-exports WorkspaceProxy, which extends + // WorkerEntrypoint from cloudflare:workers. That specifier only + // resolves under workerd, so this runs with a --import hook that + // stubs it; see script/lib/cloudflare-workers-stub.mjs. + const { TestBackend, Workspace } = await import( + `${repoRoot}/packages/computer/dist/index.js` ); // Workspace needs DurableObjectStorageLike; supply the // node:sqlite-backed test storage so this runs under plain diff --git a/script/lib/cloudflare-workers-stub.mjs b/script/lib/cloudflare-workers-stub.mjs new file mode 100644 index 00000000..4ab7f63d --- /dev/null +++ b/script/lib/cloudflare-workers-stub.mjs @@ -0,0 +1,59 @@ +// Makes `@cloudflare/computer`'s main entry importable under plain node. +// +// The entry re-exports WorkspaceProxy, which extends WorkerEntrypoint +// from `cloudflare:workers`. That specifier only resolves inside +// workerd, so a plain `import` of the entry fails with +// ERR_UNSUPPORTED_ESM_URL_SCHEME before any of the exports a host-side +// script actually wants become reachable. +// +// Registering a resolve hook for the specifier costs less than the +// alternatives. Importing dist files individually needs per-file build +// entries the bundler does not emit, and adding them would duplicate +// bundled code to suit a script. +// +// The stubs only have to satisfy `class X extends Y` at module +// evaluation time. Nothing here is ever constructed: a script that +// reaches for Workers runtime behavior wants workerd, not this. +// +// Use it with node's --import flag: +// +// node --import ./script/lib/cloudflare-workers-stub.mjs script.mjs +import { registerHooks } from "node:module"; + +const SPECIFIER = "cloudflare:workers"; +const STUB_URL = "cloudflare-workers-stub:main"; + +const SOURCE = ` +export class RpcTarget {} + +class Entrypoint { + constructor(ctx, env) { + this.ctx = ctx; + this.env = env; + } +} + +export class WorkerEntrypoint extends Entrypoint {} +export class DurableObject extends Entrypoint {} + +export const tracing = { + enterSpan(_name, callback) { + return callback(); + }, +}; +`; + +registerHooks({ + resolve(specifier, context, next) { + if (specifier === SPECIFIER) { + return { url: STUB_URL, shortCircuit: true }; + } + return next(specifier, context); + }, + load(url, context, next) { + if (url === STUB_URL) { + return { format: "module", source: SOURCE, shortCircuit: true }; + } + return next(url, context); + }, +});