From aa4f984986b8adbdd20e3dbac3f0f81bd8283db4 Mon Sep 17 00:00:00 2001 From: Pi Agent Date: Thu, 27 Aug 2026 23:00:44 +0000 Subject: [PATCH] computerd: Add an optional on-disk SQLite store computerd keeps its workspace in memory, so a restart loses it and the durable object has to send every path again. On a large workspace that replay is most of the time it takes to get back to work. Set COMPUTERD_DB to an absolute path and the store goes on the container's disk instead. The sync positions live in the same database, so a restarted daemon still knows what it was sent and the durable object only sends the difference: about 25ms instead of a full resend, whatever the size of the tree. dofs gains a ./node export with NodeSQLiteStorage, which runs node:sqlite against a file or memory. It stays out of the main entry point because that has to load under workerd, which has no node:sqlite. SQLiteTestStorage is now the in-memory pinning of the same class. Also adds POST /__computerd/checkpoint for folding the write-ahead log back before a disk snapshot, reports the store on /__computerd/info, and adds store size and free pages to /__computerd/stats. Measured with script/store-compare.mjs and script/restore-time.mjs; numbers in packages/computerd/bench-results.md. Reads are not slower on disk. Writes cost 10 to 20 percent more through a real FUSE mount. The in-memory store stays the default. --- .changeset/on-disk-computerd-store.md | 5 + docs/02_sync_protocol.md | 6 +- docs/11_lifecycle.md | 35 +- docs/19_performance.md | 63 ++++ packages/computerd/README.md | 40 ++- packages/computerd/bench-results.md | 117 +++++- packages/computerd/src/cli/computerd.test.ts | 358 ++++++++++++++++++- packages/computerd/src/cli/computerd.ts | 80 ++++- packages/computerd/src/fuse/index.ts | 4 +- packages/computerd/src/fuse/store.test.ts | 111 ++++++ packages/computerd/src/fuse/store.ts | 51 +++ packages/computerd/src/fuse/vfs.test.ts | 142 +++++++- packages/computerd/src/fuse/vfs.ts | 75 +++- packages/dofs/package.json | 4 + packages/dofs/src/node-storage.test.ts | 281 +++++++++++++++ packages/dofs/src/node-storage.ts | 212 +++++++++++ packages/dofs/src/testing.ts | 83 +---- packages/dofs/vitest.config.workers.ts | 17 +- script/restore-time.mjs | 157 ++++++++ script/store-compare.mjs | 122 +++++++ 20 files changed, 1847 insertions(+), 116 deletions(-) create mode 100644 .changeset/on-disk-computerd-store.md create mode 100644 packages/computerd/src/fuse/store.test.ts create mode 100644 packages/computerd/src/fuse/store.ts create mode 100644 packages/dofs/src/node-storage.test.ts create mode 100644 packages/dofs/src/node-storage.ts create mode 100644 script/restore-time.mjs create mode 100644 script/store-compare.mjs diff --git a/.changeset/on-disk-computerd-store.md b/.changeset/on-disk-computerd-store.md new file mode 100644 index 00000000..9f0717ea --- /dev/null +++ b/.changeset/on-disk-computerd-store.md @@ -0,0 +1,5 @@ +--- +"@cloudflare/dofs": minor +--- + +Let `computerd` keep its workspace on disk instead of in memory, so it survives a restart. Set `COMPUTERD_DB` to a file path — see [the `computerd` README](../packages/computerd/README.md#on-disk-store). diff --git a/docs/02_sync_protocol.md b/docs/02_sync_protocol.md index 610bd4f6..b744066f 100644 --- a/docs/02_sync_protocol.md +++ b/docs/02_sync_protocol.md @@ -444,9 +444,9 @@ needed; pure DO-side optimisation. ### Push backpressure A long-running exec can dirty container state faster than the DO can -pull. Today's process-lifetime container VFS caps this by OOMing, which -is a bad answer. Once a disk-backed container mirror lands the bound -shifts to path count, but the same problem persists. Likely shape: a +pull. An in-memory container VFS caps this by running out of memory. +Setting `COMPUTERD_DB` to a file path shifts the limit to path count, +but the same problem remains. Likely shape: a soft cap on the dirty set (say, 256 MiB pending bytes or 100k paths) above which FUSE write replies are delayed (real backpressure into the writer), or the container opportunistically initiates a push to the DO diff --git a/docs/11_lifecycle.md b/docs/11_lifecycle.md index 9e72d830..b129c16b 100644 --- a/docs/11_lifecycle.md +++ b/docs/11_lifecycle.md @@ -30,14 +30,14 @@ capnweb WebSocket session. │ └──────────┬──────────┘ │ │ └────────┬─────────┘ │ │ │ │ │ │ │ │ ┌──────────▼──────────┐ │ │ ┌────────▼─────────┐ │ -│ │ SQLite (ctx.storage)│ │ │ │ In-memory VFS │ │ -│ │ _vfs_watermark │ │ │ │ (process- │ │ -│ │ vfs_blobs / nodes │ │ │ │ lifetime DB) │ │ +│ │ SQLite (ctx.storage)│ │ │ │ SQLite VFS │ │ +│ │ _vfs_watermark │ │ │ │ (in memory, or │ │ +│ │ vfs_blobs / nodes │ │ │ │ COMPUTERD_DB) │ │ │ └─────────────────────┘ │ │ └──────────────────┘ │ └───────────────────────────┘ └────────────────────────┘ | | - | source of truth process-lifetime | - | (durable across restarts) (lost on restart)| + | source of truth mirror, in memory + | (durable across restarts) or on the container disk ``` The 1:1 mapping is load-bearing for several reasons: @@ -91,8 +91,8 @@ 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` + `/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), +a fresh capnweb session against the same container-side VFS. If the +container died too, the next sync round is a rev-0 baseline rebuild from the DO's store. ### Wake triggers @@ -127,11 +127,22 @@ lifetime policy. From the DO's perspective: exits, and the backend's `#monitoring` flag drops the cached handle at that point so the next call rebuilds from scratch (see the container host and backend implementations under `packages/computer/src/backends/container/`). -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. 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. +When `computerd` runs with its default in-memory store, the two sides +differ: the **container's VFS lasts only as long as the process**, +while the **DO's VFS is durable SQLite**. A container 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. + +Setting `COMPUTERD_DB` to a path changes this for a process restart. +The container-side store is written to disk, sync cursors included, +and reopened on the next start. The durable object then finds a peer +that still knows what it was sent, and only sends the difference. + +The container's disk does not survive a container restart, so this +helps a `computerd` crash inside a live container today. It will help +a container restart once the platform offers disk snapshots. See the +`computerd` README for the setting and its limits. ## Capnweb lifecycle diff --git a/docs/19_performance.md b/docs/19_performance.md index 50515710..9033441e 100644 --- a/docs/19_performance.md +++ b/docs/19_performance.md @@ -48,6 +48,69 @@ computerd is ~2x slower than the container's ext4 disk for the full `npm install`, and ~3.6x slower than tmpfs. The disk comparison is the more realistic baseline for general usage. +## In-memory store versus on-disk store + +`computerd` keeps its SQLite store in memory by default. Set +`COMPUTERD_DB` to a path and it goes on the container's disk instead. + +These numbers come from `script/store-compare.mjs`, which uses the +dofs filesystem directly with 2,000 files in one directory. There is +no FUSE mount involved, so any difference is down to the store. + +| Operation | memory | on disk (64 MiB cache) | ratio | +|---|---:|---:|---:| +| create 2000 files | 180.9 ms | 659.1 ms | 3.64x | +| stat 2000 paths, cold | 1580.4 ms | 1558.7 ms | **0.99x** | +| stat 2000 paths, warm | 12.1 ms | 15.1 ms | 1.25x | +| readdir x50 | 155.4 ms | 143.1 ms | **0.92x** | + +Reads are not slower on disk. That holds even when the cache is far +too small for the tree: a 2 MiB cache against a 3.8 MiB database still +reads at 0.99x. Two reasons. Most of the time goes on walking the path +rather than fetching pages, and the operating system caches whatever +SQLite drops. + +Writes are slower, and the reason is the cost of flushing to disk. +Creating 1,000 files takes 445 ms at `synchronous = full`, 292 ms at +`normal` (what we ship), and 148 ms at `off` — which matches the +in-memory store's 181 ms. + +Through a real FUSE mount the write cost shrinks, because the mount +itself is already the bigger expense: + +| Scenario | memory store | file store | baseline | +|---|---:|---:|---:| +| stat 1000 files | 2777.3 ms (1.10x) | 3114.9 ms (1.22x) | ~2540 ms | +| create 1000 files | 989.4 ms (0.98x) | 1178.7 ms (1.17x) | ~1010 ms | +| write 64 MiB | 238.1 ms (11.14x) | 221.9 ms (12.69x) | ~19 ms | +| overwrite 64 MiB | 294.3 ms (26.16x) | 304.6 ms (29.30x) | ~11 ms | + +## Restore time + +This is what the on-disk store buys. `script/restore-time.mjs` times +what a host waits after a restart: connect, compare sync positions, +and send whatever the other side is missing. + +| Tree | memory store | file store | +|---|---:|---:| +| 500 files | 480 ms (502 entries sent) | 26 ms (0 sent) | +| 3,000 files | 3749 ms (3002 entries sent) | 23 ms (0 sent) | + +An in-memory store sends the whole workspace again after every +restart, so its cost grows with the tree. A file store sends nothing, +because the sync positions came back along with the files. Restoring +takes about 25 ms whatever the size, so the saving grows too: 18x at +500 files, 161x at 3,000. + +The trade: small-file work costs 10 to 20 percent more, and a restart +costs a flat 25 ms instead of resending everything. + +Two things these numbers do not cover. They come from one Linux +container, not from Cloudflare Containers hardware, and a full +`cloudflare/sandbox-sdk` `npm install` has not been run. The restore +figures also drive the sync protocol in process, so they show the work +avoided but not the network round trips a real host would also skip. + ## Where computerd is faster than the disk baseline The in-memory inode store beats real disk on metadata-heavy work: diff --git a/packages/computerd/README.md b/packages/computerd/README.md index 64345557..cc1f5469 100644 --- a/packages/computerd/README.md +++ b/packages/computerd/README.md @@ -26,7 +26,8 @@ Current endpoints: - `GET /health` returns `200 OK` with `ok\n` once the HTTP server is up (it does not currently block on FUSE readiness). - `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 /__computerd/stats` returns JSON with DOFS table row counts, total inline and blob byte sizes, the orphan-blob subset, process resident memory, and the store's own size and free-page count. Useful for watching how the store grows under load. +- `POST /__computerd/checkpoint` folds the store's write-ahead log back into the database file and returns `{ walFrames, sizeBytes, durationMs }`. For a host about to take a disk snapshot. Any other method returns `405`. - `GET /` returns `200 OK` with an empty JSON object: `{}`. - `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. @@ -45,7 +46,41 @@ Current filesystem support: - Unsupported FUSE operations return `ENOSYS` to the kernel; the binding logs a one-shot warning per operation. - 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. +- Optional on-disk storage through `COMPUTERD_DB`. Unset, the in-memory store is rebuilt at each start and the host sends its state back. Set to a path, the store survives a restart and the host sends only what changed. See [On-disk store](#on-disk-store). + +## On-disk store + +`COMPUTERD_DB` picks where the workspace lives: + +```sh +COMPUTERD_DB=memory # default: in-memory, rebuilt on every start +COMPUTERD_DB=/var/lib/computerd/state.db # on-disk, survives a restart +``` + +The path must be absolute and must not sit inside `MOUNT_POINT`. A database file that the FUSE mount also shows would feed its own writes back to itself. `computerd` refuses to start on either mistake. + +Keeping the store on disk saves more than the files. The sync positions live in the same database (`_vfs_watermark`, `_vfs_fetch_cursor`, `_vfs_push_cursor`). Without them a restarted daemon looks further behind than it is, so the durable object sends every path in the workspace again. With them it sends only what changed. On a large workspace that is the difference between resending everything and doing nothing. + +The exec log does not persist. `computerd_exec_log` and `computerd_exec_meta` are cleared at every start, because the processes they describe are gone. + +### Settings + +A file store opens with write-ahead logging, `synchronous = normal`, a 64 MiB page cache, a 256 MiB memory map, temporary tables in memory, and a five-second busy timeout. + +`synchronous = normal` flushes to disk when the log is folded back rather than on every commit. That is safe here: the durable object holds the real copy, so a host crash that loses the last few writes costs a resend, not data. + +### Checkpointing + +- `POST /__computerd/checkpoint` folds the write-ahead log back into the database file and returns `{ walFrames, sizeBytes, durationMs }`. Call it before taking a disk snapshot, so the snapshot holds one file rather than a file plus a log. +- The same thing happens on `SIGTERM` and `SIGINT`, after the FUSE unmount. The order matters: the FUSE driver writes buffered bytes to the database when it releases a file, so unmounting first is what gets those bytes in. +- `GET /__computerd/stats` reports `store_size_bytes` and `store_freelist_count` next to the table counts, so you can watch the file grow. + +### Limits + +- Take snapshots between commands, not during one. A checkpoint keeps the database itself valid, but a snapshot taken mid-command catches a half-written workspace. A half-finished `npm install` is still half-finished after a restore. +- An older `computerd` exits with `EIO` rather than open a store written by a newer one. Restoring onto an older release fails loudly, which is intended. +- Mount rows (`_vfs_mounts`) come back with the store and may be out of date until the durable object rebuilds them. +- If the store is further ahead than the durable object, the daemon cannot fix it: it answers sync requests but never starts one. Begin from a fresh disk instead. ## FUSE write model @@ -115,6 +150,7 @@ Additional environment variables: EXEC_LOG_MAX_BYTES=1048576 # cap the in-memory exec log buffer (bytes) RPC_CLIENT_SECRET= # require Authorization: Bearer on every route but /health COMPUTER_VAR_NODE_ENV=production # forwarded into exec as NODE_ENV +COMPUTERD_DB=/var/lib/computerd/state.db # on-disk store; "memory" or unset keeps it in memory ``` `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. diff --git a/packages/computerd/bench-results.md b/packages/computerd/bench-results.md index e0629968..d38ae9db 100644 --- a/packages/computerd/bench-results.md +++ b/packages/computerd/bench-results.md @@ -1,4 +1,119 @@ -# FUSE mount option benchmarks +# Store and FUSE mount option benchmarks + +## In-memory store versus file-backed store + +Numbers from `script/store-compare.mjs`, which +drives the dofs filesystem directly against both storage backends. +It deliberately skips FUSE so a difference here is the store and +nothing else. 2,000 files in one directory, Node 24 on a Linux +container. + +| Store | create 2000 | stat cold | stat warm | readdir x50 | +|---|---:|---:|---:|---:| +| memory | 180.9 ms | 1580.4 ms | 12.1 ms | 155.4 ms | +| file, 64 MiB cache | 659.1 ms (3.64x) | 1558.7 ms (0.99x) | 15.1 ms (1.25x) | 143.1 ms (0.92x) | +| file, 256 MiB cache | 650.0 ms (3.59x) | 1509.8 ms (0.96x) | 13.8 ms (1.14x) | 146.9 ms (0.95x) | + +Two findings, one of which contradicts what we assumed when writing +the plan. + +**Metadata reads do not regress.** Cold `stat` of 2,000 paths is a +wash (0.96x to 0.99x), and `readdir` is if anything slightly faster +on the file store. The plan predicted this was where a file-backed +store would hurt. It does not, because the working set here is about +1.4 MiB — small enough to sit entirely in SQLite's page cache, so the +reads never reach the disk. Warm `stat` is 1.14x to 1.25x slower, +which is the resolve cache doing its job in both cases and the +remaining difference being page-cache lookup overhead rather than +input or output. + +**Writes are the real cost, and the cause is fsync.** Creating 2,000 +files is 3.6x slower on the file store. Varying `synchronous` isolates +it: + +| `synchronous` | create 1000 files | +|---|---:| +| `full` | 444.6 ms | +| `normal` | 291.9 ms | +| `off` | 148.1 ms | + +`off` matches the in-memory store, so the gap is entirely the cost of +flushing to disk. `normal` is the shipped default and already buys +back a third of `full`. Anything faster trades durability for speed, +which is defensible here because the durable object is the source of +truth, but `off` risks a corrupt database on host loss rather than +merely losing recent transactions, so it stays off the table. + +Sweeping the cache budget changes almost nothing. At 6,000 files the +database is 3.8 MiB; squeezing the cache to 2 MiB, so the working set +genuinely cannot fit, still leaves cold `stat` at 0.99x: + +| Store | create 6000 | stat cold | stat warm | readdir x50 | +|---|---:|---:|---:|---:| +| memory | 391.9 ms | 14389.1 ms | 34.4 ms | 437.1 ms | +| file, 2 MiB cache | 1682.0 ms (4.29x) | 14177.7 ms (0.99x) | 55.4 ms (1.61x) | 474.4 ms (1.09x) | + +That is the interesting result. The prediction was that a cache too +small for the tree would turn every resolve into a `pread` and wreck +the metadata numbers. It does not, because cold `stat` is dominated by +the resolve walk itself rather than by fetching pages, and the +operating system's own page cache absorbs what SQLite evicts. Warm +`stat` is where the difference shows, and it is 20 microseconds per +operation on a path that is already cheap. + +## Through a real FUSE mount + +The numbers above isolate the storage layer. These run the same +comparison through `script/fs-bench.sh` against a real kernel FUSE +mount, with `computerd` started on the host (`FUSE_MOUNT=fuse`), and +`/tmp` as the baseline. REPS=2, WARMUP=1. + +| Scenario | memory store | file store | baseline | +|---|---:|---:|---:| +| stat 1000 files | 2777.3 ms (1.10x) | 3114.9 ms (1.22x) | ~2540 ms | +| create 1000 files | 989.4 ms (0.98x) | 1178.7 ms (1.17x) | ~1010 ms | +| write 64 MiB | 238.1 ms (11.14x) | 221.9 ms (12.69x) | ~19 ms | +| overwrite 64 MiB | 294.3 ms (26.16x) | 304.6 ms (29.30x) | ~11 ms | + +Large-file input and output is unchanged between the two stores, which +is what the storage-layer numbers predicted: those paths are dominated +by chunking and the FUSE round trip, so the store barely registers. +The small-file scenarios cost 10 to 20 percent more on disk. That is a +real regression, and smaller than the 3.6x the storage-layer create +number would suggest on its own, because FUSE overhead dilutes it. + +## Restore time + +What the on-disk store buys, measured by `script/restore-time.mjs`. It +times the interval a host actually waits: from a healthy daemon to a +workspace the peer agrees is current, meaning connect, reconcile +watermarks, and push whatever the peer believes is missing. + +| Tree | store | first boot | restart | +|---|---|---:|---:| +| 500 files | memory | 454 ms (502 pushed) | 480 ms (502 pushed) | +| 500 files | file | 451 ms (502 pushed) | **26 ms (0 pushed)** | +| 3,000 files | memory | 3725 ms (3002 pushed) | 3749 ms (3002 pushed) | +| 3,000 files | file | 4063 ms (3002 pushed) | **23 ms (0 pushed)** | + +An in-memory store re-ships the whole tree on every restart, so its +restart cost tracks the tree size. A file store ships nothing, because +the sync cursors came back with the files and the peer can see there +is no difference to send. The saving is 18x at 500 files and 161x at +3,000, and it keeps growing: the restore side stays flat at roughly +25 ms while the memory side climbs with the workspace. + +This is the trade in one line. Small-file work costs 10 to 20 percent +more, and a restart costs a fixed 25 ms instead of a full replay. + +Caveats. These run on one Linux container, not on Cloudflare +Containers hardware. The restore measurement drives the sync protocol +directly rather than through a real durable object over a real +network, so it captures the work avoided but not the round-trip +latency a real host would also save. The full `cloudflare/sandbox-sdk` +`npm install` comparison has not been run. + +## FUSE mount option benchmarks Numbers from running `script/run-fs-bench.sh` against the linux-x64 `computerd` binary in a privileged docker container, with the bench's pure diff --git a/packages/computerd/src/cli/computerd.test.ts b/packages/computerd/src/cli/computerd.test.ts index cb5d2ce1..70bb9f06 100644 --- a/packages/computerd/src/cli/computerd.test.ts +++ b/packages/computerd/src/cli/computerd.test.ts @@ -98,6 +98,7 @@ test("computerd exposes file IO through real FUSE when FUSE_MOUNT=fuse", async ( backend: { kind: "fuse" }, mountPoint, port, + store: { kind: "memory" }, }); await fs.mkdir(path.join(mountPoint, "dir")); @@ -695,7 +696,7 @@ function rawRequest(port, lines) { function request(url, options = {}) { return new Promise((resolve, reject) => { - const request = http.get(url, options, (response) => { + const request = http.request(url, { method: "GET", ...options }, (response) => { response.setEncoding("utf8"); let body = ""; response.on("data", (chunk) => { @@ -710,6 +711,7 @@ function request(url, options = {}) { request.setTimeout(1_000, () => { request.destroy(new Error(`request timed out: ${url}`)); }); + request.end(); }); } @@ -862,3 +864,357 @@ test("without RPC_CLIENT_SECRET every route stays open", async (_ctx) => { expect((await request(`${base}${route}`)).statusCode, route).toBe(200); } }); + +test("/__computerd/info reports the in-memory store by default", async (_ctx) => { + const port = await getAvailablePort(); + const mountPoint = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-mount-")); + await startComputerd({ port, mountPoint, env: { FUSE_MOUNT: "none" } }); + + const response = await request(`http://127.0.0.1:${port}/__computerd/info`); + expect(response.statusCode).toBe(200); + expect(JSON.parse(response.body).store).toEqual({ kind: "memory" }); +}); + +test("/__computerd/info reports the file store and its path when COMPUTERD_DB is set", async (_ctx) => { + const port = await getAvailablePort(); + const mountPoint = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-mount-")); + const storeDir = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-store-")); + onTestFinished(() => fs.rm(storeDir, { recursive: true, force: true })); + const storePath = path.join(storeDir, "state.db"); + await startComputerd({ + port, + mountPoint, + env: { FUSE_MOUNT: "none", COMPUTERD_DB: storePath }, + }); + + const response = await request(`http://127.0.0.1:${port}/__computerd/info`); + expect(JSON.parse(response.body).store).toEqual({ + kind: "file", + path: storePath, + fresh: true, + }); +}); + +test("computerd rejects a relative COMPUTERD_DB value", async () => { + const port = await getAvailablePort(); + const child = spawn(cliPath, { + cwd: packageRoot, + env: { + ...process.env, + MOUNT_POINT: "/tmp/computerd-mount-not-used", + PORT: String(port), + FUSE_MOUNT: "none", + COMPUTERD_DB: "state.db", + }, + stdio: ["ignore", "ignore", "pipe"], + }); + + const { code, stderr } = await waitForExit(child); + expect(code).toBe(1); + expect(stderr).toMatch(/COMPUTERD_DB must be "memory" or an absolute path/); +}); + +test("computerd rejects a COMPUTERD_DB path inside the mount point", async () => { + const port = await getAvailablePort(); + const child = spawn(cliPath, { + cwd: packageRoot, + env: { + ...process.env, + MOUNT_POINT: "/tmp/computerd-mount-not-used", + PORT: String(port), + FUSE_MOUNT: "none", + COMPUTERD_DB: "/tmp/computerd-mount-not-used/state.db", + }, + stdio: ["ignore", "ignore", "pipe"], + }); + + const { code, stderr } = await waitForExit(child); + expect(code).toBe(1); + expect(stderr).toMatch(/must not sit inside the mount point/); +}); + +test("POST /__computerd/checkpoint leaves the store readable", 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-")); + const storeDir = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-store-")); + onTestFinished(() => fs.rm(storeDir, { recursive: true, force: true })); + await startComputerd({ + port, + mountPoint, + env: { FUSE_MOUNT: "none", COMPUTERD_DB: path.join(storeDir, "state.db") }, + }); + + const response = await request(`http://127.0.0.1:${port}/__computerd/checkpoint`, { + method: "POST", + }); + expect(response.statusCode).toBe(200); + const body = JSON.parse(response.body); + expect(typeof body.walFrames).toBe("number"); + expect(body.sizeBytes).toBeGreaterThan(0); + expect(typeof body.durationMs).toBe("number"); + + const client = createWorkspaceClient({ url: `ws://127.0.0.1:${port}/api` }); + try { + expect(await client.sync.hasObjects([])).toEqual([]); + } finally { + await client.close(); + } +}); + +test("POST /__computerd/checkpoint on an in-memory store reports no frames", async (_ctx) => { + const port = await getAvailablePort(); + const mountPoint = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-mount-")); + await startComputerd({ port, mountPoint, env: { FUSE_MOUNT: "none" } }); + + const response = await request(`http://127.0.0.1:${port}/__computerd/checkpoint`, { + method: "POST", + }); + expect(response.statusCode).toBe(200); + expect(JSON.parse(response.body).walFrames).toBe(0); +}); + +test("/__computerd/checkpoint refuses a GET", async (_ctx) => { + const port = await getAvailablePort(); + const mountPoint = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-mount-")); + await startComputerd({ port, mountPoint, env: { FUSE_MOUNT: "none" } }); + + const response = await request(`http://127.0.0.1:${port}/__computerd/checkpoint`); + expect(response.statusCode).toBe(405); +}); + +test("/__computerd/checkpoint requires the shared secret when one is set", async (_ctx) => { + const secret = "checkpoint-secret"; + const port = await getAvailablePort(); + const mountPoint = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-mount-")); + await startComputerd({ + port, + mountPoint, + env: { FUSE_MOUNT: "none", RPC_CLIENT_SECRET: secret }, + }); + + const unauthorized = await request(`http://127.0.0.1:${port}/__computerd/checkpoint`, { + method: "POST", + }); + expect(unauthorized.statusCode).toBe(401); + + const authorized = await request(`http://127.0.0.1:${port}/__computerd/checkpoint`, { + method: "POST", + headers: { authorization: `Bearer ${secret}` }, + }); + expect(authorized.statusCode).toBe(200); +}); + +test("/__computerd/stats reports the store size and free pages", async (_ctx) => { + const port = await getAvailablePort(); + const mountPoint = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-mount-")); + const storeDir = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-store-")); + onTestFinished(() => fs.rm(storeDir, { recursive: true, force: true })); + await startComputerd({ + port, + mountPoint, + env: { FUSE_MOUNT: "none", COMPUTERD_DB: path.join(storeDir, "state.db") }, + }); + + const response = await request(`http://127.0.0.1:${port}/__computerd/stats`); + const stats = JSON.parse(response.body); + expect(stats.store_size_bytes).toBeGreaterThan(0); + expect(typeof stats.store_freelist_count).toBe("number"); +}); + +test("a write survives SIGTERM because the store is checkpointed after the unmount", async (_ctx) => { + const { createWorkspaceClient } = await import("@cloudflare/computer-rpc/client"); + const storeDir = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-store-")); + onTestFinished(() => fs.rm(storeDir, { recursive: true, force: true })); + const storePath = path.join(storeDir, "state.db"); + + const firstPort = await getAvailablePort(); + const firstMount = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-mount-")); + const child = await startComputerd({ + port: firstPort, + mountPoint: firstMount, + env: { FUSE_MOUNT: "none", COMPUTERD_DB: storePath }, + }); + + const { Database, initializeSchema, WorkspaceFilesystem } = await import("@cloudflare/dofs"); + const { SQLiteTestStorage } = await import("@cloudflare/dofs/testing"); + const { pushOnce } = await import("@cloudflare/computer-rpc/driver"); + + const writer = createWorkspaceClient({ url: `ws://127.0.0.1:${firstPort}/api` }); + try { + const local = new Database(new SQLiteTestStorage()); + initializeSchema(local, Date.now); + await new WorkspaceFilesystem(local).writeFile("/survives.txt", "written before SIGTERM"); + expect(await pushOnce(local, writer.sync)).toBeGreaterThan(0); + } finally { + await writer.close(); + } + + await stopProcess(child); + + const secondPort = await getAvailablePort(); + const secondMount = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-mount-")); + await startComputerd({ + port: secondPort, + mountPoint: secondMount, + env: { FUSE_MOUNT: "none", COMPUTERD_DB: storePath }, + }); + + const reader = createWorkspaceClient({ url: `ws://127.0.0.1:${secondPort}/api` }); + try { + const entry = await reader.sync.readEntry("/survives.txt"); + expect(entry).not.toBeNull(); + } finally { + await reader.close(); + } +}); + +test("a file store brings the tree and the sync cursors back after a restart", async (_ctx) => { + const { Database, initializeSchema, WorkspaceFilesystem } = await import("@cloudflare/dofs"); + const { SQLiteTestStorage } = await import("@cloudflare/dofs/testing"); + const { createWorkspaceClient } = await import("@cloudflare/computer-rpc/client"); + const { pushOnce } = await import("@cloudflare/computer-rpc/driver"); + + const storeDir = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-store-")); + onTestFinished(() => fs.rm(storeDir, { recursive: true, force: true })); + const storePath = path.join(storeDir, "state.db"); + + const firstPort = await getAvailablePort(); + const firstMount = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-mount-")); + const first = await startComputerd({ + port: firstPort, + mountPoint: firstMount, + env: { FUSE_MOUNT: "none", COMPUTERD_DB: storePath }, + }); + + const writer = createWorkspaceClient({ url: `ws://127.0.0.1:${firstPort}/api` }); + try { + const local = new Database(new SQLiteTestStorage()); + initializeSchema(local, Date.now); + const localFs = new WorkspaceFilesystem(local); + await localFs.mkdir("/restart", { recursive: true }); + await localFs.writeFile("/restart/a.txt", "before the restart"); + expect(await pushOnce(local, writer.sync)).toBeGreaterThan(0); + } finally { + await writer.close(); + } + + const before = JSON.parse((await request(`http://127.0.0.1:${firstPort}/api/watermarks`)).body); + expect(before.fetchCursor.rev).toBeGreaterThan(0); + + await stopProcess(first); + + const secondPort = await getAvailablePort(); + const secondMount = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-mount-")); + await startComputerd({ + port: secondPort, + mountPoint: secondMount, + env: { FUSE_MOUNT: "none", COMPUTERD_DB: storePath }, + }); + + const after = JSON.parse((await request(`http://127.0.0.1:${secondPort}/api/watermarks`)).body); + expect(after.currentRev).toBe(before.currentRev); + expect(after.fetchCursor).toEqual(before.fetchCursor); + + const reader = createWorkspaceClient({ url: `ws://127.0.0.1:${secondPort}/api` }); + try { + const entry = await reader.sync.readEntry("/restart/a.txt"); + expect(entry).not.toBeNull(); + } finally { + await reader.close(); + } +}); + +test("an in-memory store comes back empty after a restart", async (_ctx) => { + const { Database, initializeSchema, WorkspaceFilesystem } = await import("@cloudflare/dofs"); + const { SQLiteTestStorage } = await import("@cloudflare/dofs/testing"); + const { createWorkspaceClient } = await import("@cloudflare/computer-rpc/client"); + const { pushOnce } = await import("@cloudflare/computer-rpc/driver"); + + const firstPort = await getAvailablePort(); + const firstMount = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-mount-")); + const first = await startComputerd({ + port: firstPort, + mountPoint: firstMount, + env: { FUSE_MOUNT: "none" }, + }); + + const writer = createWorkspaceClient({ url: `ws://127.0.0.1:${firstPort}/api` }); + try { + const local = new Database(new SQLiteTestStorage()); + initializeSchema(local, Date.now); + const localFs = new WorkspaceFilesystem(local); + await localFs.mkdir("/restart", { recursive: true }); + await localFs.writeFile("/restart/a.txt", "before the restart"); + expect(await pushOnce(local, writer.sync)).toBeGreaterThan(0); + } finally { + await writer.close(); + } + + await stopProcess(first); + + const secondPort = await getAvailablePort(); + const secondMount = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-mount-")); + await startComputerd({ + port: secondPort, + mountPoint: secondMount, + env: { FUSE_MOUNT: "none" }, + }); + + const after = JSON.parse((await request(`http://127.0.0.1:${secondPort}/api/watermarks`)).body); + expect(after.fetchCursor).toEqual({ rev: 0, path: null }); + + const reader = createWorkspaceClient({ url: `ws://127.0.0.1:${secondPort}/api` }); + try { + expect(await reader.sync.readEntry("/restart/a.txt")).toBeNull(); + } finally { + await reader.close(); + } +}); + +test("the exec log does not come back after a restart on a file store", async (_ctx) => { + const { createWorkspaceClient } = await import("@cloudflare/computer-rpc/client"); + + const storeDir = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-store-")); + onTestFinished(() => fs.rm(storeDir, { recursive: true, force: true })); + const storePath = path.join(storeDir, "state.db"); + + const firstPort = await getAvailablePort(); + const firstMount = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-mount-")); + const first = await startComputerd({ + port: firstPort, + mountPoint: firstMount, + env: { FUSE_MOUNT: "none", COMPUTERD_DB: storePath }, + }); + + let execId = ""; + const runner = createWorkspaceClient({ url: `ws://127.0.0.1:${firstPort}/api` }); + try { + const handle = await runner.shell.exec({ source: "echo hello-from-before" }); + execId = handle.id; + const reader = handle.events.getReader(); + while (true) { + const { done } = await reader.read(); + if (done) break; + } + } finally { + await runner.close(); + } + + await stopProcess(first); + + const secondPort = await getAvailablePort(); + const secondMount = await fs.mkdtemp(path.join(os.tmpdir(), "computerd-mount-")); + await startComputerd({ + port: secondPort, + mountPoint: secondMount, + env: { FUSE_MOUNT: "none", COMPUTERD_DB: storePath }, + }); + + const after = createWorkspaceClient({ url: `ws://127.0.0.1:${secondPort}/api` }); + try { + await expect(after.shell.getExec({ id: execId })).rejects.toThrow(); + } finally { + await after.close(); + } +}); diff --git a/packages/computerd/src/cli/computerd.ts b/packages/computerd/src/cli/computerd.ts index 5036db4b..9b5077f2 100644 --- a/packages/computerd/src/cli/computerd.ts +++ b/packages/computerd/src/cli/computerd.ts @@ -19,7 +19,10 @@ import { type FuseMount, mountFuse, parseFuseMountMode, + parseStoreMode, + type ResolvedStore, resolveFuseBackend, + resolveStore, } from "../fuse/index.js"; import { mountShim, type ShimMount } from "../shim/index.js"; import { installLogging } from "./logger.js"; @@ -147,6 +150,7 @@ interface ComputerdInfo { backend: FUSEBackend; mountPoint: string; port: number; + store: ResolvedStore; } // Snapshot DOFS table sizes and process memory so an external caller @@ -193,6 +197,7 @@ function createHTTPServer( rpc: ReturnType, secret: string | undefined, getStats?: () => Record, + checkpoint?: () => { walFrames: number; sizeBytes: number; durationMs: number }, ): HTTPHandle { // Holds the current outbound capnweb session opened via /connect. // Re-POSTing /connect (e.g. after a DO hibernate + new incarnation) @@ -241,6 +246,37 @@ function createHTTPServer( return; } + // /__computerd/checkpoint — fold the write-ahead log back into + // the main database file. A host about to ask the platform for a + // disk snapshot calls this first so the snapshot captures one + // file rather than a file plus a log segment. + if (path === "/__computerd/checkpoint") { + if (request.method !== "POST") { + send(response, 405, "method not allowed\n", { + allow: "POST", + "content-type": "text/plain; charset=utf-8", + }); + return; + } + if (checkpoint === undefined) { + send(response, 404, "checkpoint unavailable\n", { + "content-type": "text/plain; charset=utf-8", + }); + return; + } + try { + send(response, 200, JSON.stringify(checkpoint()), { + "content-type": "application/json; charset=utf-8", + }); + } catch (error) { + console.error("/__computerd/checkpoint failed:", error); + send(response, 500, "internal error\n", { + "content-type": "text/plain; charset=utf-8", + }); + } + return; + } + if (request.method !== "GET" && request.method !== "HEAD") { send(response, 405, "method not allowed\n", { allow: "GET, HEAD", @@ -595,8 +631,21 @@ async function main(): Promise { const backend: FUSEBackend = await resolveFuseBackend(fuseMountMode); console.log(`[info] FUSE_MOUNT=${fuseMountMode} resolved to backend=${backend.kind}`); - const { vfs, db } = await createNodeVirtualFileSystem(); - const info: ComputerdInfo = { backend, mountPoint, port }; + const store = resolveStore(parseStoreMode(process.env.COMPUTERD_DB), mountPoint); + console.log( + `[info] COMPUTERD_DB resolved to store=${store.kind}${ + store.kind === "file" ? ` path=${store.path} fresh=${store.fresh}` : "" + }`, + ); + + const { + vfs, + db, + checkpoint: checkpointStore, + storeStats, + close: closeStore, + } = await createNodeVirtualFileSystem({ store }); + const info: ComputerdInfo = { backend, mountPoint, port, store }; let fuse: FuseMount | undefined; // When running on the userspace shim, capture the typed handle @@ -677,10 +726,21 @@ async function main(): Promise { } : {}), }); - const http = createHTTPServer(info, rpc, clientSecret, () => ({ - ...collectDbStats(db), - ...(fuse?.getBufferStats?.() ?? {}), - })); + const http = createHTTPServer( + info, + rpc, + clientSecret, + () => { + const { sizeBytes, freelistCount } = storeStats(); + return { + ...collectDbStats(db), + ...(fuse?.getBufferStats?.() ?? {}), + store_size_bytes: sizeBytes, + store_freelist_count: freelistCount, + }; + }, + checkpointStore, + ); let shuttingDown = false; const shutdown = async (signal: NodeJS.Signals): Promise => { @@ -701,6 +761,14 @@ async function main(): Promise { if (fuse !== undefined) { await unmount(fuse); } + // Unmount first. The FUSE driver writes any buffered bytes to the + // database when it releases a file, and those writes need to land + // before the store closes. + try { + closeStore(); + } catch (error) { + console.error("failed to close the store:", error); + } teardownLogging(); process.exit(signal === "SIGINT" ? 130 : 143); }; diff --git a/packages/computerd/src/fuse/index.ts b/packages/computerd/src/fuse/index.ts index 8b82f9cc..e508fa06 100644 --- a/packages/computerd/src/fuse/index.ts +++ b/packages/computerd/src/fuse/index.ts @@ -2,5 +2,7 @@ export type { FUSEBackend, FuseMountMode, ResolveFuseBackendOptions } from "./ba export { parseFuseMountMode, resolveFuseBackend } from "./backend.js"; export type { FuseMount, FuseOps, FuseStat } from "./driver.js"; export { makeFUSEOps, mountFuse } from "./driver.js"; -export type { NodeVirtualFileSystem } from "./vfs.js"; +export type { ResolvedStore, StoreMode } from "./store.js"; +export { parseStoreMode, resolveStore } from "./store.js"; +export type { CreateNodeVFSOptions, NodeVFSHandle, NodeVirtualFileSystem } from "./vfs.js"; export { createNodeVirtualFileSystem } from "./vfs.js"; diff --git a/packages/computerd/src/fuse/store.test.ts b/packages/computerd/src/fuse/store.test.ts new file mode 100644 index 00000000..2d95c2b6 --- /dev/null +++ b/packages/computerd/src/fuse/store.test.ts @@ -0,0 +1,111 @@ +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { describe, expect, onTestFinished, test } from "vitest"; + +import { parseStoreMode, resolveStore } from "./store.js"; + +// A temp directory that removes itself when the calling test ends. +function createTempDir(): string { + const dir = mkdtempSync(join(tmpdir(), "computerd-store-")); + onTestFinished(() => rmSync(dir, { recursive: true, force: true })); + return dir; +} + +describe("parseStoreMode", () => { + test("treats an unset value as the in-memory store", () => { + expect(parseStoreMode(undefined)).toEqual({ kind: "memory" }); + }); + + test("treats an empty value as the in-memory store", () => { + expect(parseStoreMode("")).toEqual({ kind: "memory" }); + }); + + test("treats the literal 'memory' as the in-memory store", () => { + expect(parseStoreMode("memory")).toEqual({ kind: "memory" }); + }); + + test("treats an absolute path as a file store", () => { + expect(parseStoreMode("/var/lib/computerd/state.db")).toEqual({ + kind: "file", + path: "/var/lib/computerd/state.db", + }); + }); + + test("rejects a relative path and names the variable", () => { + expect(() => parseStoreMode("state.db")).toThrow(/COMPUTERD_DB/); + }); + + test("rejects a relative path that walks upward", () => { + expect(() => parseStoreMode("../state.db")).toThrow(/COMPUTERD_DB/); + }); + + test("normalizes a path with redundant segments", () => { + expect(parseStoreMode("/var/lib/../lib/computerd/state.db")).toEqual({ + kind: "file", + path: "/var/lib/computerd/state.db", + }); + }); +}); + +describe("resolveStore", () => { + test("passes the in-memory store through", () => { + expect(resolveStore({ kind: "memory" }, "/workspace")).toEqual({ kind: "memory" }); + }); + + test("reports a file store as fresh when the file does not exist", () => { + const path = join(createTempDir(), "state.db"); + expect(resolveStore({ kind: "file", path }, "/workspace")).toEqual({ + kind: "file", + path, + fresh: true, + }); + }); + + test("reports a file store as not fresh when the file already exists", () => { + const path = join(createTempDir(), "state.db"); + writeFileSync(path, ""); + expect(resolveStore({ kind: "file", path }, "/workspace")).toEqual({ + kind: "file", + path, + fresh: false, + }); + }); + + test("rejects a store inside the mount point", () => { + expect(() => resolveStore({ kind: "file", path: "/workspace/state.db" }, "/workspace")).toThrow( + /mount point/i, + ); + }); + + test("rejects a store deeper inside the mount point", () => { + expect(() => + resolveStore({ kind: "file", path: "/workspace/nested/state.db" }, "/workspace"), + ).toThrow(/mount point/i); + }); + + test("rejects a store at the mount point itself", () => { + expect(() => resolveStore({ kind: "file", path: "/workspace" }, "/workspace")).toThrow( + /mount point/i, + ); + }); + + test("allows a sibling directory whose name merely starts with the mount point", () => { + const resolved = resolveStore( + { kind: "file", path: "/workspace-other/state.db" }, + "/workspace", + ); + expect(resolved).toEqual({ kind: "file", path: "/workspace-other/state.db", fresh: true }); + }); + + test("allows a store outside a mount point given with a trailing slash", () => { + const resolved = resolveStore({ kind: "file", path: "/var/lib/state.db" }, "/workspace/"); + expect(resolved).toEqual({ kind: "file", path: "/var/lib/state.db", fresh: true }); + }); + + test("rejects a store inside a mount point given with a trailing slash", () => { + expect(() => + resolveStore({ kind: "file", path: "/workspace/state.db" }, "/workspace/"), + ).toThrow(/mount point/i); + }); +}); diff --git a/packages/computerd/src/fuse/store.ts b/packages/computerd/src/fuse/store.ts new file mode 100644 index 00000000..538539c3 --- /dev/null +++ b/packages/computerd/src/fuse/store.ts @@ -0,0 +1,51 @@ +// Reads the COMPUTERD_DB environment variable and works out where the +// workspace database should live. +// +// "memory", or nothing at all, keeps the database in memory. An +// absolute path puts it on disk, where it survives a restart. + +import { existsSync } from "node:fs"; +import { isAbsolute, normalize, resolve as resolvePath, sep } from "node:path"; + +export type StoreMode = { kind: "memory" } | { kind: "file"; path: string }; + +export type ResolvedStore = { kind: "memory" } | { kind: "file"; path: string; fresh: boolean }; + +const MEMORY_VALUE = "memory"; + +export function parseStoreMode(value: string | undefined): StoreMode { + if (value === undefined || value === "") return { kind: "memory" }; + const trimmed = value.trim(); + if (trimmed === "" || trimmed === MEMORY_VALUE) return { kind: "memory" }; + // Same rule MOUNT_POINT enforces. A relative path resolves against + // the working directory, which is not something a container author + // should have to reason about. + if (!isAbsolute(trimmed)) { + throw new Error( + `COMPUTERD_DB must be "memory" or an absolute path, got ${JSON.stringify(value)}`, + ); + } + return { kind: "file", path: normalize(trimmed) }; +} + +export function resolveStore(mode: StoreMode, mountPoint: string): ResolvedStore { + if (mode.kind === "memory") return { kind: "memory" }; + assertOutsideMountPoint(mode.path, mountPoint); + return { kind: "file", path: mode.path, fresh: !existsSync(mode.path) }; +} + +// A store the FUSE mount also projects is a loop: writing to the +// database produces filesystem entries, which produce writes to the +// database. Cheap to check here, confusing to debug in the field. +function assertOutsideMountPoint(path: string, mountPoint: string): void { + const mount = resolvePath(mountPoint); + const candidate = resolvePath(path); + // Compare against the mount plus a separator, not the bare prefix. + // A plain startsWith would also reject /workspace-other, which is a + // sibling and perfectly legal. + if (candidate === mount || candidate.startsWith(`${mount}${sep}`)) { + throw new Error( + `COMPUTERD_DB must not sit inside the mount point (${mount}), got ${JSON.stringify(path)}`, + ); + } +} diff --git a/packages/computerd/src/fuse/vfs.test.ts b/packages/computerd/src/fuse/vfs.test.ts index 43ac6fbb..a66061a8 100644 --- a/packages/computerd/src/fuse/vfs.test.ts +++ b/packages/computerd/src/fuse/vfs.test.ts @@ -1,7 +1,17 @@ -import { expect, test } from "vitest"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { describe, expect, onTestFinished, test } from "vitest"; import { createNodeVirtualFileSystem } from "./index.js"; +// A temp directory that removes itself when the calling test ends. +function createTempDir(): string { + const dir = mkdtempSync(join(tmpdir(), "computerd-vfs-store-")); + onTestFinished(() => rmSync(dir, { recursive: true, force: true })); + return dir; +} + test("createNodeVirtualFileSystem returns a @platformatic/vfs filesystem", async () => { const { vfs } = await createNodeVirtualFileSystem(); @@ -19,3 +29,133 @@ test("createNodeVirtualFileSystem returns a @platformatic/vfs filesystem", async vfs.unlinkSync("/project/greeting.txt"); expect(vfs.readdirSync("/project")).toEqual([]); }); + +describe("createNodeVirtualFileSystem store selection", () => { + test("defaults to an in-memory store", async () => { + const handle = await createNodeVirtualFileSystem(); + expect(handle.store).toEqual({ kind: "memory" }); + handle.close(); + }); + + test("an in-memory store starts empty on every call", async () => { + const first = await createNodeVirtualFileSystem(); + first.vfs.writeFileSync("/only-in-first.txt", Buffer.from("x")); + first.close(); + + const second = await createNodeVirtualFileSystem(); + const exists = second.vfs.existsSync("/only-in-first.txt"); + second.close(); + + expect(exists).toBe(false); + }); + + test("a file store keeps its files across close and reopen", async () => { + const path = join(createTempDir(), "state.db"); + + const first = await createNodeVirtualFileSystem({ + store: { kind: "file", path, fresh: true }, + }); + first.vfs.mkdirSync("/workspace/repo", { recursive: true }); + first.vfs.writeFileSync("/workspace/repo/a.txt", Buffer.from("persisted")); + first.close(); + + const second = await createNodeVirtualFileSystem({ + store: { kind: "file", path, fresh: false }, + }); + const contents = second.vfs.readFileSync("/workspace/repo/a.txt").toString(); + second.close(); + + expect(contents).toBe("persisted"); + }); + + test("a file store reports the resolved path back to the caller", async () => { + const path = join(createTempDir(), "state.db"); + const handle = await createNodeVirtualFileSystem({ + store: { kind: "file", path, fresh: true }, + }); + const store = handle.store; + handle.close(); + + expect(store).toEqual({ kind: "file", path, fresh: true }); + }); + + test("a file store forwards the extra dofs methods onto the vfs instance", async () => { + const path = join(createTempDir(), "state.db"); + const handle = await createNodeVirtualFileSystem({ + store: { kind: "file", path, fresh: true }, + }); + + const forwarded = [ + "linkSync", + "createFileSync", + "writeRangeSync", + "truncateFileSync", + "chmodSync", + "readRangeSync", + "openWriteBufferSync", + "openWriteBufferForCreateSync", + "releaseWriteBufferSync", + ] as const; + const missing = forwarded.filter( + (name) => typeof (handle.vfs as unknown as Record)[name] !== "function", + ); + handle.close(); + + expect(missing).toEqual([]); + }); + + test("a file store serves range reads through the forwarded dofs method", async () => { + const path = join(createTempDir(), "state.db"); + const handle = await createNodeVirtualFileSystem({ + store: { kind: "file", path, fresh: true }, + }); + handle.vfs.writeFileSync("/ranged.txt", Buffer.from("abcdefghij")); + + const readRange = (handle.vfs as unknown as Record).readRangeSync as ( + p: string, + offset: number, + length: number, + ) => Uint8Array; + const slice = Buffer.from(readRange("/ranged.txt", 2, 3)).toString(); + handle.close(); + + expect(slice).toBe("cde"); + }); + + test("refuses a store written by a newer schema version", async () => { + const path = join(createTempDir(), "state.db"); + const first = await createNodeVirtualFileSystem({ + store: { kind: "file", path, fresh: true }, + }); + first.db.run("UPDATE vfs_meta SET v = v + 1 WHERE k = ?", "schema_version"); + first.close(); + + await expect( + createNodeVirtualFileSystem({ store: { kind: "file", path, fresh: false } }), + ).rejects.toThrow(/schema version/i); + }); + + test("migrates a store written by an older schema version", async () => { + const path = join(createTempDir(), "state.db"); + const first = await createNodeVirtualFileSystem({ + store: { kind: "file", path, fresh: true }, + }); + first.vfs.writeFileSync("/pre-migration.txt", Buffer.from("kept")); + const current = first.db.scalar("SELECT v FROM vfs_meta WHERE k = ?", "schema_version"); + first.db.run("UPDATE vfs_meta SET v = ? WHERE k = ?", 2, "schema_version"); + first.close(); + + const second = await createNodeVirtualFileSystem({ + store: { kind: "file", path, fresh: false }, + }); + const migrated = second.db.scalar( + "SELECT v FROM vfs_meta WHERE k = ?", + "schema_version", + ); + const contents = second.vfs.readFileSync("/pre-migration.txt").toString(); + second.close(); + + expect(migrated).toBe(current); + expect(contents).toBe("kept"); + }); +}); diff --git a/packages/computerd/src/fuse/vfs.ts b/packages/computerd/src/fuse/vfs.ts index c4023be7..13b5c0ee 100644 --- a/packages/computerd/src/fuse/vfs.ts +++ b/packages/computerd/src/fuse/vfs.ts @@ -1,7 +1,14 @@ -import { Database, initializeSchema, SQLiteWorkspaceProvider } from "@cloudflare/dofs"; -import { SQLiteTestStorage } from "@cloudflare/dofs/testing"; +import { + Database, + initializeSchema, + invalidateReadOnlyMountCache, + SQLiteWorkspaceProvider, +} from "@cloudflare/dofs"; +import { NodeSQLiteStorage } from "@cloudflare/dofs/node"; import { create, type VirtualFileSystem, VirtualProvider } from "@platformatic/vfs"; +import type { ResolvedStore } from "./store.js"; + export type NodeVirtualFileSystem = VirtualFileSystem; // @platformatic/vfs's create() guards on `provider instanceof @@ -49,23 +56,58 @@ const EXTRA_VFS_METHODS = [ "releaseWriteBufferSync", ] as const; -export interface NodeVfsHandle { +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 createWorkspaceServer(db) and serve the local store // to whoever holds the capnweb session. db: Database; + // Which store backs this handle, as resolved by store.ts. The CLI + // reports it on /__computerd/info. + store: ResolvedStore; + // Fold any write-ahead log back into the main database file. + checkpoint: () => { walFrames: number; sizeBytes: number; durationMs: number }; + // Byte size of the store, and the pages SQLite is holding free + // inside it. Both feed /__computerd/stats. + storeStats: () => { sizeBytes: number; freelistCount: number }; + // Checkpoint and close the underlying database. + close: () => void; +} + +export interface CreateNodeVFSOptions { + store?: ResolvedStore; } -// 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 { +export async function createNodeVirtualFileSystem( + options: CreateNodeVFSOptions = {}, +): Promise { ensureVirtualProviderPrototype(); - const storage = new SQLiteTestStorage(); + const store: ResolvedStore = options.store ?? { kind: "memory" }; + const storage = new NodeSQLiteStorage({ + location: store.kind === "file" ? store.path : ":memory:", + }); const db = new Database(storage); - initializeSchema(db, () => Date.now()); + try { + // On a restored store this is the migration step, and the guard + // against a stale binary: initializeSchema migrates a database + // written by an older computerd forward, and throws EIO on one + // written by a newer computerd rather than corrupting it. Closing + // the storage on the way out keeps a rejected open from leaking + // the file handle. + initializeSchema(db, () => Date.now()); + } catch (error) { + storage.close(); + throw error; + } + // A restored store arrives with whatever _vfs_mounts rows the + // previous process wrote. The read-only mount guard caches those + // roots per Database, so drop the cache before anything reads + // through it. The durable object's mount indexer re-asserts the + // real set on connect; until it does, restored rows may be stale. + if (store.kind === "file" && !store.fresh) { + invalidateReadOnlyMountCache(db); + } const provider = new SQLiteWorkspaceProvider(db); const vfs = create(provider as unknown as VirtualProvider, { moduleHooks: false }); @@ -85,5 +127,18 @@ export async function createNodeVirtualFileSystem(): Promise { configurable: true, }); } - return { vfs, db }; + return { + vfs, + db, + store, + checkpoint: () => storage.checkpoint(), + storeStats: () => ({ + sizeBytes: storage.sizeBytes(), + freelistCount: storage.freelistCount(), + }), + close: () => { + storage.checkpoint(); + storage.close(); + }, + }; } diff --git a/packages/dofs/package.json b/packages/dofs/package.json index 40872848..e0ff8238 100644 --- a/packages/dofs/package.json +++ b/packages/dofs/package.json @@ -8,6 +8,10 @@ "types": "./dist/index.d.ts", "default": "./dist/index.js" }, + "./node": { + "types": "./dist/node-storage.d.ts", + "default": "./dist/node-storage.js" + }, "./testing": { "types": "./dist/testing.d.ts", "default": "./dist/testing.js" diff --git a/packages/dofs/src/node-storage.test.ts b/packages/dofs/src/node-storage.test.ts new file mode 100644 index 00000000..0b1d3c0e --- /dev/null +++ b/packages/dofs/src/node-storage.test.ts @@ -0,0 +1,281 @@ +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; + +import { WorkspaceFilesystem } from "./fs/filesystem.js"; +import { NodeSQLiteStorage } from "./node-storage.js"; +import { initializeSchema, ROOT_INODE, SCHEMA_VERSION } from "./schema/index.js"; +import { Database } from "./storage.js"; +import { + readFetchCursor, + readPushCursor, + writeFetchCursor, + writePushCursor, +} from "./sync/watermarks.js"; + +let dir: string; + +beforeEach(() => { + dir = mkdtempSync(join(tmpdir(), "dofs-node-storage-")); +}); + +afterEach(() => { + rmSync(dir, { recursive: true, force: true }); +}); + +function openStore(path: string): { storage: NodeSQLiteStorage; db: Database } { + const storage = new NodeSQLiteStorage({ location: path }); + const db = new Database(storage); + initializeSchema(db, () => 1000); + return { storage, db }; +} + +describe("NodeSQLiteStorage on a file", () => { + it("keeps files and their bytes across a close and reopen", async () => { + const path = join(dir, "state.db"); + + const first = openStore(path); + const fs = new WorkspaceFilesystem(first.db, { now: () => 1000 }); + await fs.mkdir("/workspace/repo", { recursive: true }); + await fs.writeFile("/workspace/repo/a.txt", "hello from the first process"); + first.storage.close(); + + const second = openStore(path); + const reopened = new WorkspaceFilesystem(second.db, { now: () => 2000 }); + const contents = await reopened.readFile("/workspace/repo/a.txt", "utf8"); + second.storage.close(); + + expect(contents).toBe("hello from the first process"); + }); + + it("keeps the sync cursors across a close and reopen", () => { + const path = join(dir, "state.db"); + + const first = openStore(path); + writePushCursor(first.db, { rev: 42, path: null }); + writeFetchCursor(first.db, { rev: 41, path: "/workspace/b.txt" }); + first.storage.close(); + + const second = openStore(path); + const push = readPushCursor(second.db); + const fetch = readFetchCursor(second.db); + second.storage.close(); + + expect(push).toEqual({ rev: 42, path: null }); + expect(fetch).toEqual({ rev: 41, path: "/workspace/b.txt" }); + }); + + it("does not re-seed the root inode when reopening an existing store", () => { + const path = join(dir, "state.db"); + + const first = openStore(path); + first.db.run("UPDATE vfs_nodes SET mtime = ? WHERE inode = ?", 5555, ROOT_INODE); + first.storage.close(); + + const second = openStore(path); + const mtime = second.db.scalar( + "SELECT mtime FROM vfs_nodes WHERE inode = ?", + ROOT_INODE, + ); + const version = second.db.scalar( + "SELECT v FROM vfs_meta WHERE k = ?", + "schema_version", + ); + second.storage.close(); + + expect(mtime).toBe(5555); + expect(version).toBe(SCHEMA_VERSION); + }); + + it("turns on write-ahead logging", () => { + const path = join(dir, "state.db"); + const { storage, db } = openStore(path); + + const mode = db.scalar("PRAGMA journal_mode"); + storage.close(); + + expect(mode).toBe("wal"); + }); + + it("converts a byte cache budget into SQLite's negative kibibyte form", () => { + const path = join(dir, "state.db"); + const storage = new NodeSQLiteStorage({ + location: path, + cacheSizeBytes: 64 * 1024 * 1024, + }); + const db = new Database(storage); + + const cacheSize = db.scalar("PRAGMA cache_size"); + storage.close(); + + expect(cacheSize).toBe(-65536); + }); + + it("applies the memory map size it was given", () => { + const path = join(dir, "state.db"); + const storage = new NodeSQLiteStorage({ + location: path, + mmapSizeBytes: 256 * 1024 * 1024, + }); + const db = new Database(storage); + + const mmapSize = db.scalar("PRAGMA mmap_size"); + storage.close(); + + expect(mmapSize).toBe(268435456); + }); + + it("applies the synchronous level it was given", () => { + const path = join(dir, "state.db"); + const storage = new NodeSQLiteStorage({ location: path, synchronous: "normal" }); + const db = new Database(storage); + + const level = db.scalar("PRAGMA synchronous"); + storage.close(); + + expect(level).toBe(1); + }); + + it("keeps temporary tables in memory", () => { + const path = join(dir, "state.db"); + const { storage, db } = openStore(path); + + const tempStore = db.scalar("PRAGMA temp_store"); + storage.close(); + + expect(tempStore).toBe(2); + }); + + it("creates the database file when it does not exist", async () => { + const path = join(dir, "nested", "state.db"); + const { storage, db } = openStore(path); + const fs = new WorkspaceFilesystem(db, { now: () => 1000 }); + + await fs.writeFile("/probe.txt", "x"); + storage.close(); + + const reopened = openStore(path); + const contents = await new WorkspaceFilesystem(reopened.db).readFile("/probe.txt", "utf8"); + reopened.storage.close(); + + expect(contents).toBe("x"); + }); + + it("leaves the store readable after a checkpoint", async () => { + const path = join(dir, "state.db"); + const { storage, db } = openStore(path); + const fs = new WorkspaceFilesystem(db, { now: () => 1000 }); + await fs.writeFile("/checkpointed.txt", "still here"); + + storage.checkpoint(); + + const contents = await fs.readFile("/checkpointed.txt", "utf8"); + storage.close(); + + expect(contents).toBe("still here"); + }); + + it("reports the bytes a checkpoint folded back into the main file", async () => { + const path = join(dir, "state.db"); + const { storage, db } = openStore(path); + const fs = new WorkspaceFilesystem(db, { now: () => 1000 }); + await fs.writeFile("/sized.txt", "x".repeat(4096)); + + const result = storage.checkpoint(); + storage.close(); + + expect(result.sizeBytes).toBeGreaterThan(0); + }); + + it("rolls back a failed transaction", () => { + const path = join(dir, "state.db"); + const { storage, db } = openStore(path); + + expect(() => { + db.transactionSync(() => { + db.run("INSERT INTO vfs_meta (k, v) VALUES (?, ?)", "rollback_probe", 1); + throw new Error("forced"); + }); + }).toThrow("forced"); + + const value = db.scalar("SELECT v FROM vfs_meta WHERE k = ?", "rollback_probe"); + storage.close(); + + expect(value).toBeUndefined(); + }); + + it("commits an inner savepoint while the outer transaction continues", () => { + const path = join(dir, "state.db"); + const { storage, db } = openStore(path); + + db.transactionSync(() => { + db.run("INSERT INTO vfs_meta (k, v) VALUES (?, ?)", "outer", 1); + db.transactionSync(() => { + db.run("INSERT INTO vfs_meta (k, v) VALUES (?, ?)", "inner", 2); + }); + }); + + const outer = db.scalar("SELECT v FROM vfs_meta WHERE k = ?", "outer"); + const inner = db.scalar("SELECT v FROM vfs_meta WHERE k = ?", "inner"); + storage.close(); + + expect(outer).toBe(1); + expect(inner).toBe(2); + }); + + it("rolls the outer transaction back over a committed inner savepoint", () => { + const path = join(dir, "state.db"); + const { storage, db } = openStore(path); + + expect(() => { + db.transactionSync(() => { + db.transactionSync(() => { + db.run("INSERT INTO vfs_meta (k, v) VALUES (?, ?)", "nested_probe", 1); + }); + throw new Error("forced"); + }); + }).toThrow("forced"); + + const value = db.scalar("SELECT v FROM vfs_meta WHERE k = ?", "nested_probe"); + storage.close(); + + expect(value).toBeUndefined(); + }); +}); + +describe("NodeSQLiteStorage in memory", () => { + it("accepts the in-memory location and serves a working schema", () => { + const storage = new NodeSQLiteStorage({ location: ":memory:" }); + const db = new Database(storage); + initializeSchema(db, () => 1234); + + const row = db.one<{ inode: number; type: string }>( + "SELECT inode, type FROM vfs_nodes WHERE inode = ?", + ROOT_INODE, + ); + storage.close(); + + expect(row).toEqual({ inode: ROOT_INODE, type: "dir" }); + }); + + it("leaves journal mode alone for an in-memory store", () => { + const storage = new NodeSQLiteStorage({ location: ":memory:" }); + const db = new Database(storage); + + const mode = db.scalar("PRAGMA journal_mode"); + storage.close(); + + expect(mode).toBe("memory"); + }); + + it("reports nothing to fold when checkpointing an in-memory store", () => { + const storage = new NodeSQLiteStorage({ location: ":memory:" }); + new Database(storage); + + const result = storage.checkpoint(); + storage.close(); + + expect(result.walFrames).toBe(0); + }); +}); diff --git a/packages/dofs/src/node-storage.ts b/packages/dofs/src/node-storage.ts new file mode 100644 index 00000000..430793f6 --- /dev/null +++ b/packages/dofs/src/node-storage.ts @@ -0,0 +1,212 @@ +// A DurableObjectStorageLike backed by node:sqlite, either in memory +// or on a file. +// +// This module cannot be re-exported from the package's main entry +// point. That entry has to load under workerd, which has no +// node:sqlite, and an import of it there fails at module load rather +// than at the call site. It ships under the "./node" export instead, +// and the in-memory pinning used by tests lives in ./testing.ts. + +import { mkdirSync } from "node:fs"; +import { dirname } from "node:path"; +import { DatabaseSync, type StatementSync } from "node:sqlite"; + +import type { DurableObjectStorageLike, SQLCursorLike } from "./types.js"; + +export type JournalMode = "wal" | "delete" | "memory"; +export type SynchronousLevel = "off" | "normal" | "full"; + +export interface NodeSQLiteStorageOptions { + // A filesystem path, or ":memory:". + location: string; + journalMode?: JournalMode; + // Page cache budget, in bytes. SQLite's own pragma wants a negative + // number of kibibytes; the conversion happens at the pragma so + // every size on this interface stays in one unit. + cacheSizeBytes?: number; + mmapSizeBytes?: number; + synchronous?: SynchronousLevel; + tempStore?: "default" | "file" | "memory"; + busyTimeoutMs?: number; +} + +export interface CheckpointResult { + walFrames: number; + sizeBytes: number; + durationMs: number; +} + +const IN_MEMORY_LOCATION = ":memory:"; + +// Defaults for a file-backed store. Cache size makes little +// difference to read speed in practice: a 2 MiB cache against a +// 3.8 MiB database reads as fast as a 256 MiB one, because the +// operating system caches the pages SQLite drops. Numbers in +// packages/computerd/bench-results.md. +const DEFAULT_CACHE_SIZE_BYTES = 64 * 1024 * 1024; +const DEFAULT_MMAP_SIZE_BYTES = 256 * 1024 * 1024; +// One writer, but a snapshot tool or a second connection should wait +// rather than fail outright. +const DEFAULT_BUSY_TIMEOUT_MS = 5000; + +class NodeCursor implements SQLCursorLike { + private readonly rows: Row[]; + + constructor(rows: Row[]) { + this.rows = rows; + } + + toArray(): Row[] { + return this.rows; + } +} + +export class NodeSQLiteStorage implements DurableObjectStorageLike { + readonly location: string; + private readonly db: DatabaseSync; + private readonly cache = new Map(); + readonly sql: { + exec: (query: string, ...bindings: unknown[]) => SQLCursorLike; + }; + + constructor(options: NodeSQLiteStorageOptions) { + this.location = options.location; + const onDisk = options.location !== IN_MEMORY_LOCATION; + if (onDisk) { + mkdirSync(dirname(options.location), { recursive: true }); + } + this.db = new DatabaseSync(options.location); + this.applyPragmas(options, onDisk); + this.sql = { + exec: (query: string, ...bindings: unknown[]): SQLCursorLike => { + let stmt = this.cache.get(query); + if (stmt === undefined) { + stmt = this.db.prepare(query); + this.cache.set(query, stmt); + } + const normalized = bindings.map(toSQLiteValue); + const rows = (stmt.all(...(normalized as never[])) as Row[]) ?? []; + return new NodeCursor(rows); + }, + }; + } + + // Every default here applies only to a file-backed store. An + // in-memory database has no disk to tune against, and forcing a + // journal mode on it would change behavior the several hundred + // tests built on SQLiteTestStorage already depend on. + private applyPragmas(options: NodeSQLiteStorageOptions, onDisk: boolean): void { + const journalMode = options.journalMode ?? (onDisk ? "wal" : undefined); + if (journalMode !== undefined) { + this.db.exec(`PRAGMA journal_mode = ${journalMode}`); + } + // "normal" fsyncs on checkpoint rather than on every commit. Safe + // here because the durable object holds the authoritative copy: + // losing the last few transactions to a host crash costs a + // re-push, not data. Measured at roughly a third faster than + // "full" for file creation. "off" is faster still and closes the + // gap to the in-memory store entirely, but it risks a corrupt + // database rather than merely losing recent writes, so it is not + // the default. + const synchronous = options.synchronous ?? (onDisk ? "normal" : undefined); + if (synchronous !== undefined) { + this.db.exec(`PRAGMA synchronous = ${synchronous}`); + } + const cacheSizeBytes = + options.cacheSizeBytes ?? (onDisk ? DEFAULT_CACHE_SIZE_BYTES : undefined); + if (cacheSizeBytes !== undefined) { + // SQLite reads a negative cache_size as a budget in kibibytes + // and a positive one as a count of pages. We want the byte + // budget, so the sign is deliberate rather than a slip. + const kibibytes = Math.max(1, Math.floor(cacheSizeBytes / 1024)); + this.db.exec(`PRAGMA cache_size = -${kibibytes}`); + } + const mmapSizeBytes = options.mmapSizeBytes ?? (onDisk ? DEFAULT_MMAP_SIZE_BYTES : undefined); + if (mmapSizeBytes !== undefined) { + this.db.exec(`PRAGMA mmap_size = ${Math.max(0, Math.floor(mmapSizeBytes))}`); + } + const tempStore = options.tempStore ?? (onDisk ? "memory" : undefined); + if (tempStore !== undefined) { + this.db.exec(`PRAGMA temp_store = ${tempStore}`); + } + const busyTimeoutMs = options.busyTimeoutMs ?? (onDisk ? DEFAULT_BUSY_TIMEOUT_MS : undefined); + if (busyTimeoutMs !== undefined) { + this.db.exec(`PRAGMA busy_timeout = ${Math.max(0, Math.floor(busyTimeoutMs))}`); + } + } + + transactionSync(closure: () => T): T { + this.db.exec("BEGIN"); + try { + const result = closure(); + this.db.exec("COMMIT"); + return result; + } catch (error) { + this.db.exec("ROLLBACK"); + throw error; + } + } + + // Fold the write-ahead log back into the main database file. A host + // about to ask the platform for a disk snapshot calls this first so + // the snapshot captures one file rather than a file plus a log + // segment that a restore would have to replay. + // + // TRUNCATE rather than PASSIVE: it waits until the log is fully + // folded and then empties it, which is the state a snapshot wants. + checkpoint(): CheckpointResult { + const startedAt = Date.now(); + let walFrames = 0; + if (this.location !== IN_MEMORY_LOCATION) { + const rows = this.db.prepare("PRAGMA wal_checkpoint(TRUNCATE)").all() as unknown as Array< + Record + >; + const row = rows[0]; + if (row !== undefined) { + // The pragma answers with three unnamed columns: a busy flag, + // the frames the log held, and the frames folded down. Column + // names differ across SQLite builds, so read by position. + const values = Object.values(row); + walFrames = typeof values[1] === "number" && values[1] > 0 ? values[1] : 0; + } + } + return { + walFrames, + sizeBytes: this.sizeBytes(), + durationMs: Date.now() - startedAt, + }; + } + + sizeBytes(): number { + const pageCount = this.scalarNumber("PRAGMA page_count"); + const pageSize = this.scalarNumber("PRAGMA page_size"); + return pageCount * pageSize; + } + + freelistCount(): number { + return this.scalarNumber("PRAGMA freelist_count"); + } + + private scalarNumber(pragma: string): number { + const rows = this.db.prepare(pragma).all() as unknown as Array>; + const row = rows[0]; + if (row === undefined) return 0; + const value = Object.values(row)[0]; + return typeof value === "number" ? value : 0; + } + + close(): void { + this.cache.clear(); + this.db.close(); + } +} + +function toSQLiteValue(value: unknown): string | number | bigint | null | Uint8Array { + if (value === undefined || value === null) return null; + if (typeof value === "boolean") return value ? 1 : 0; + if (value instanceof Uint8Array) return value; + if (typeof value === "string" || typeof value === "number" || typeof value === "bigint") { + return value; + } + throw new TypeError(`NodeSQLiteStorage cannot bind value of type ${typeof value}`); +} diff --git a/packages/dofs/src/testing.ts b/packages/dofs/src/testing.ts index 3b42ec02..f9d271b4 100644 --- a/packages/dofs/src/testing.ts +++ b/packages/dofs/src/testing.ts @@ -3,7 +3,12 @@ // surface is a subset of this, so anything that works here works on // the real platform too. // -// This module imports node:sqlite at the top level and therefore +// The implementation lives in ./node-storage.ts, which serves both +// this fixture and the on-disk store computerd runs in production. +// This class is the in-memory pinning of it, kept as its own name +// because several hundred tests construct it with no arguments. +// +// This module reaches node:sqlite through that import and therefore // cannot be loaded under workerd. RecordingStorage — the // pure-JS fixture that also lives in dofs's testing surface // — has moved to ./testing-recording.ts so it can be imported @@ -11,80 +16,16 @@ // `import { RecordingStorage } from "@cloudflare/dofs/testing"` // call sites keep working under node. -import { DatabaseSync, type StatementSync } from "node:sqlite"; - -import type { DurableObjectStorageLike, SQLCursorLike } from "./types.js"; +import { NodeSQLiteStorage } from "./node-storage.js"; export type { ExecutedStatement } from "./testing-recording.js"; export { RecordingStorage } from "./testing-recording.js"; -class TestCursor implements SQLCursorLike { - private readonly rows: Row[]; - - constructor(rows: Row[]) { - this.rows = rows; - } - - toArray(): Row[] { - return this.rows; - } -} - -export class SQLiteTestStorage implements DurableObjectStorageLike { - private readonly db: DatabaseSync; - private readonly cache = new Map(); - readonly sql: { - exec: (query: string, ...bindings: unknown[]) => SQLCursorLike; - }; - +// Kept as its own name, and kept taking no arguments, because +// several hundred call sites construct it that way. Collapsing it +// into NodeSQLiteStorage would touch every one of them for no gain. +export class SQLiteTestStorage extends NodeSQLiteStorage { constructor() { - this.db = new DatabaseSync(":memory:"); - this.sql = { - exec: (query: string, ...bindings: unknown[]): SQLCursorLike => { - // node:sqlite refuses statements with trailing whitespace through - // prepare(); also we cache prepared statements per unique query - // string to keep the fixture fast. - const key = query; - let stmt = this.cache.get(key); - if (stmt === undefined) { - stmt = this.db.prepare(query); - this.cache.set(key, stmt); - } - const normalizedBindings = bindings.map(toSQLiteValue); - const rows = (stmt.all(...(normalizedBindings as never[])) as Row[]) ?? []; - return new TestCursor(rows); - }, - }; - } - - transactionSync(closure: () => T): T { - this.db.exec("BEGIN"); - try { - const result = closure(); - this.db.exec("COMMIT"); - return result; - } catch (error) { - this.db.exec("ROLLBACK"); - throw error; - } - } - - close(): void { - // StatementSync instances are released when the database closes. - this.cache.clear(); - this.db.close(); - } -} - -// node:sqlite is strict about input shapes: it accepts strings, numbers, -// bigints, null, and Uint8Array but not undefined, Buffer subclasses -// other than Uint8Array, or booleans. Normalize. -function toSQLiteValue(value: unknown): string | number | bigint | null | Uint8Array { - if (value === undefined || value === null) return null; - if (typeof value === "boolean") return value ? 1 : 0; - if (value instanceof Uint8Array) return value; - if (typeof value === "string" || typeof value === "number" || typeof value === "bigint") { - return value; + super({ location: ":memory:" }); } - throw new TypeError(`SQLiteTestStorage cannot bind value of type ${typeof value}`); } diff --git a/packages/dofs/vitest.config.workers.ts b/packages/dofs/vitest.config.workers.ts index 2632b169..7f24bfbb 100644 --- a/packages/dofs/vitest.config.workers.ts +++ b/packages/dofs/vitest.config.workers.ts @@ -24,13 +24,14 @@ export default defineConfig({ test: { globals: true, include: ["src/**/*.test.ts"], - // These files exercise SQLiteTestStorage directly. The - // node:sqlite-backed fixture has no analogue under workerd and - // importing it crashes the pool worker instead of reporting a - // module-resolution error. schema/index.test.ts also stages raw, - // pre-migration databases, which withDB cannot represent. - // All other tests run under both backends; helpers delegate to - // withDB, which this config aliases to a DO-backed implementation. - exclude: ["src/schema/index.test.ts", "src/testing.test.ts"], + // These files exercise SQLiteTestStorage or NodeSQLiteStorage + // directly. The node:sqlite-backed store has no analogue under + // workerd and importing it crashes the pool worker instead of + // reporting a module-resolution error. schema/index.test.ts also + // stages raw, pre-migration databases, which withDB cannot + // represent. All other tests run under both backends; helpers + // delegate to withDB, which this config aliases to a DO-backed + // implementation. + exclude: ["src/schema/index.test.ts", "src/testing.test.ts", "src/node-storage.test.ts"], }, }); diff --git a/script/restore-time.mjs b/script/restore-time.mjs new file mode 100644 index 00000000..1cd34e2c --- /dev/null +++ b/script/restore-time.mjs @@ -0,0 +1,157 @@ +// Measures what an on-disk store saves when computerd restarts. +// +// Two runs, same workload: +// memory - the store is rebuilt empty, so the peer re-ships the tree. +// file - the store is reopened, so the peer ships only the delta. +// +// "Restore" here is the wall time from a started daemon to a workspace +// the peer agrees is up to date: connect, reconcile watermarks, and +// push whatever the peer thinks is missing. That is the cost a real +// host pays before the first command can run. +// +// Usage, from the repository root: +// node script/restore-time.mjs [fileCount] + +import { spawn } from "node:child_process"; +import { mkdtempSync, rmSync } from "node:fs"; +import net from "node:net"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +const FILES = Number(process.argv[2] ?? 2000); +const CLI = new URL("../packages/computerd/dist/cli/computerd.cjs", import.meta.url).pathname; + +const { Database, initializeSchema, WorkspaceFilesystem } = await import("@cloudflare/dofs"); +const { SQLiteTestStorage } = await import("@cloudflare/dofs/testing"); +const { createWorkspaceClient } = await import("@cloudflare/computer-rpc/client"); +const { pushOnce, reconcileWatermarks } = await import("@cloudflare/computer-rpc/driver"); + +function freePort() { + return new Promise((resolve, reject) => { + const s = net.createServer(); + s.once("error", reject); + s.listen(0, "127.0.0.1", () => { + const { port } = s.address(); + s.close(() => resolve(port)); + }); + }); +} + +async function waitHealthy(port, timeoutMs = 20000) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + try { + const res = await fetch(`http://127.0.0.1:${port}/health`); + if (res.ok) return; + } catch {} + await new Promise((r) => setTimeout(r, 50)); + } + throw new Error("daemon never became healthy"); +} + +function start(port, mountPoint, storePath) { + const child = spawn("node", [CLI], { + env: { + ...process.env, + PORT: String(port), + MOUNT_POINT: mountPoint, + FUSE_MOUNT: "none", + COMPUTERD_DB: storePath ?? "memory", + }, + stdio: ["ignore", "pipe", "pipe"], + }); + return child; +} + +async function stop(child) { + if (child.exitCode !== null) return; + await new Promise((resolve) => { + child.once("exit", resolve); + child.kill("SIGTERM"); + setTimeout(() => { + child.kill("SIGKILL"); + resolve(); + }, 8000); + }); +} + +// The durable-object side: an authoritative store holding the tree. +function buildPeer() { + const db = new Database(new SQLiteTestStorage()); + initializeSchema(db, Date.now); + return db; +} + +async function seedPeer(db) { + const fs = new WorkspaceFilesystem(db, { now: () => 1000 }); + await fs.mkdir("/w/pkg", { recursive: true }); + for (let i = 0; i < FILES; i++) { + await fs.writeFile(`/w/pkg/f${i}.js`, `module.exports = ${i};\n`); + } +} + +// One measured cycle. Returns the milliseconds from "daemon is +// healthy" to "peer has finished syncing it". +async function cycle({ storePath, mountPoint, peer }) { + const port = await freePort(); + const child = start(port, mountPoint, storePath); + await waitHealthy(port); + + const started = process.hrtime.bigint(); + const client = createWorkspaceClient({ url: `ws://127.0.0.1:${port}/api` }); + let pushed = 0; + try { + await reconcileWatermarks(peer, client.sync); + pushed = await pushOnce(peer, client.sync); + } finally { + await client.close(); + } + const ms = Number(process.hrtime.bigint() - started) / 1e6; + + await stop(child); + return { ms, pushed }; +} + +const dir = mkdtempSync(join(tmpdir(), "restore-time-")); +const rows = []; +try { + for (const mode of ["memory", "file"]) { + const mountPoint = join(dir, `mount-${mode}`); + const storePath = mode === "file" ? join(dir, `${mode}.db`) : undefined; + + // A fresh peer per mode so both start from the same place. + const peer = buildPeer(); + await seedPeer(peer); + + // First boot: the daemon is empty either way, so this is the + // cold cost of shipping the whole tree. + const first = await cycle({ storePath, mountPoint, peer }); + + // Second boot: this is the restore. With a file store the daemon + // reopens what it had; with memory it starts empty again. + const second = await cycle({ storePath, mountPoint, peer }); + + rows.push({ mode, first, second }); + } +} finally { + rmSync(dir, { recursive: true, force: true }); +} + +console.log(`\nfiles=${FILES}\n`); +console.log(`${"store".padEnd(8)} ${"first boot".padStart(22)} ${"restart".padStart(22)}`); +console.log("-".repeat(56)); +for (const r of rows) { + const f = `${r.first.ms.toFixed(0)}ms (${r.first.pushed} pushed)`; + const s = `${r.second.ms.toFixed(0)}ms (${r.second.pushed} pushed)`; + console.log(`${r.mode.padEnd(8)} ${f.padStart(22)} ${s.padStart(22)}`); +} +const mem = rows.find((r) => r.mode === "memory"); +const file = rows.find((r) => r.mode === "file"); +if (mem && file) { + console.log( + `\nrestart saving: ${(mem.second.ms - file.second.ms).toFixed(0)}ms ` + + `(${(mem.second.ms / Math.max(file.second.ms, 0.01)).toFixed(1)}x faster), ` + + `${mem.second.pushed - file.second.pushed} fewer entries pushed\n`, + ); +} +console.log(`${JSON.stringify({ files: FILES, rows })}\n`); diff --git a/script/store-compare.mjs b/script/store-compare.mjs new file mode 100644 index 00000000..70890e79 --- /dev/null +++ b/script/store-compare.mjs @@ -0,0 +1,122 @@ +// Compares computerd's two stores: in memory against a file on disk. +// +// It drives the dofs filesystem directly, without a FUSE mount, so any +// difference in the numbers comes from the store and nothing else. The +// operations are the ones most likely to suffer when reads have to +// reach a disk: path resolution, stat, readdir, and small file writes. +// +// Run from the repository root, after a build: +// +// node script/store-compare.mjs +// +// Environment: +// FILES=2000 files created per tree +// CACHE_MIB=64,256 page cache budgets to compare, in mebibytes +// +// For the same comparison through a real FUSE mount, see +// script/fs-bench.sh. For what the file store saves on a restart, see +// script/restore-time.mjs. + +import { mkdtempSync, rmSync, statSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { Database, initializeSchema, WorkspaceFilesystem } from "@cloudflare/dofs"; +import { NodeSQLiteStorage } from "@cloudflare/dofs/node"; + +const FILES = Number(process.env.FILES ?? 2000); +const CACHE_MIB = (process.env.CACHE_MIB ?? "64,256") + .split(",") + .map((v) => Number(v.trim())) + .filter((v) => Number.isFinite(v) && v > 0); + +function ms(fn) { + const started = process.hrtime.bigint(); + fn(); + return Number(process.hrtime.bigint() - started) / 1e6; +} + +async function msAsync(fn) { + const started = process.hrtime.bigint(); + await fn(); + return Number(process.hrtime.bigint() - started) / 1e6; +} + +function open(location, cacheSizeBytes) { + const storage = new NodeSQLiteStorage( + cacheSizeBytes === undefined ? { location } : { location, cacheSizeBytes }, + ); + const db = new Database(storage); + initializeSchema(db, () => 1000); + return { storage, db, fs: new WorkspaceFilesystem(db, { now: () => 1000 }) }; +} + +async function buildTree(fs) { + await fs.mkdir("/w/pkg", { recursive: true }); + for (let i = 0; i < FILES; i++) { + await fs.writeFile(`/w/pkg/f${i}.js`, `module.exports = ${i};\n`); + } +} + +const paths = Array.from({ length: FILES }, (_, i) => `/w/pkg/f${i}.js`); + +// A file store can be reopened to get a genuinely cold cache. An +// in-memory store cannot, so its first pass over a freshly built tree +// is the closest equivalent. +async function measure(label, location, cacheSizeBytes) { + let handle = open(location, cacheSizeBytes); + const create = await msAsync(() => buildTree(handle.fs)); + + if (location !== ":memory:") { + handle.storage.close(); + handle = open(location, cacheSizeBytes); + } + + const statCold = ms(() => { + for (const p of paths) handle.fs.stat(p); + }); + const statWarm = ms(() => { + for (const p of paths) handle.fs.stat(p); + }); + const readdir = ms(() => { + for (let i = 0; i < 50; i++) handle.fs.ls("/w/pkg"); + }); + + const sizeBytes = handle.storage.sizeBytes(); + handle.storage.close(); + return { label, create, statCold, statWarm, readdir, sizeBytes }; +} + +const dir = mkdtempSync(join(tmpdir(), "store-compare-")); +const rows = []; +try { + rows.push(await measure("memory", ":memory:")); + for (const mib of CACHE_MIB) { + const path = join(dir, `file-${mib}.db`); + const row = await measure(`file cache=${mib}MiB`, path, mib * 1024 * 1024); + row.fileBytes = statSync(path).size; + rows.push(row); + } +} finally { + rmSync(dir, { recursive: true, force: true }); +} + +const base = rows[0]; +const pad = (s, n) => String(s).padEnd(n); +const num = (v, n) => v.toFixed(1).padStart(n); + +console.log(`\nfiles=${FILES} node=${process.version}\n`); +console.log( + `${pad("store", 20)} ${pad("create", 10)} ${pad("stat cold", 12)} ${pad("stat warm", 12)} ${pad("readdir x50", 12)} ${pad("db MiB", 8)}`, +); +console.log("-".repeat(80)); +for (const r of rows) { + const ratio = (v, b) => (r === base ? "" : ` (${(v / b).toFixed(2)}x)`); + console.log( + `${pad(r.label, 20)} ${num(r.create, 7)}ms${ratio(r.create, base.create)} ` + + `${num(r.statCold, 7)}ms${ratio(r.statCold, base.statCold)} ` + + `${num(r.statWarm, 7)}ms${ratio(r.statWarm, base.statWarm)} ` + + `${num(r.readdir, 7)}ms${ratio(r.readdir, base.readdir)} ` + + `${((r.fileBytes ?? r.sizeBytes) / 1024 / 1024).toFixed(1)}`, + ); +} +console.log(`\n${JSON.stringify({ files: FILES, rows })}\n`);