From 2fc081cda7d6c594bcd44145423b5cc9c7451c08 Mon Sep 17 00:00:00 2001 From: Caio Nogueira Date: Thu, 6 Aug 2026 10:35:36 +0100 Subject: [PATCH 1/2] dofs: Link staged chunks during sync apply Sync receivers already stage chunk payloads in content-addressed storage before applying file entries. Reading them back and concatenating a whole-file buffer makes peak isolate memory roughly twice the file size. Link each file's existing chunk references directly while preserving metadata and validating declared sizes. This keeps payload bytes out of the apply path and relies on vfs_chunks reachability to protect linked blobs from collection. --- packages/dofs/src/fs/gc.test.ts | 43 ++++++++++ packages/dofs/src/fs/writeFile.ts | 20 ++++- packages/dofs/src/sync/apply.test.ts | 73 ++++++++++++++++- packages/dofs/src/sync/apply.ts | 112 +++++++++++++-------------- 4 files changed, 186 insertions(+), 62 deletions(-) diff --git a/packages/dofs/src/fs/gc.test.ts b/packages/dofs/src/fs/gc.test.ts index 64cce055..127cebf5 100644 --- a/packages/dofs/src/fs/gc.test.ts +++ b/packages/dofs/src/fs/gc.test.ts @@ -1,7 +1,12 @@ +import { createHash } from "node:crypto"; + import { describe, expect, it } from "vitest"; import type { Database } from "../storage.js"; +import { applyChanges } from "../sync/apply.js"; +import { stageBlob } from "../sync/blobs.js"; import { gc } from "./gc.js"; +import { readFile } from "./readFile.js"; import { rm } from "./rm.js"; import { withDB } from "./with-db.js"; import { writeFile } from "./writeFile.js"; @@ -33,6 +38,44 @@ describe("gc", () => { }); }); + it("does not free blobs a sync apply linked", async () => { + await withDB(async (db) => { + // A pull stages chunks before linking them, so vfs_chunks reachability + // must protect blobs independently of their staging timestamp. + const bytes = new TextEncoder().encode("staged content"); + const hash = new Uint8Array(createHash("sha256").update(bytes).digest()); + stageBlob(db, hash, bytes, 1000); + + await applyChanges( + db, + [ + { + kind: "file", + rev: 1, + path: "/staged.txt", + mode: 0o644, + mtime: 1000, + size: bytes.byteLength, + chunks: [{ hash, size: bytes.byteLength }], + }, + ], + new Map(), + { source: "upstream" }, + ); + expect(blobCount(db)).toBe(1); + + // A zero safety window and far-future clock make every blob stale; + // only references from vfs_chunks protect this blob and manifest. + expect(gc(db, { now: () => 999_999_999, safetyWindowMs: 0 })).toEqual({ + blobsFreed: 0, + manifestsFreed: 0, + }); + expect(blobCount(db)).toBe(1); + expect(blobBytesCount(db)).toBe(1); + expect(await readFile(db, "/staged.txt", "utf8")).toBe("staged content"); + }); + }); + it("frees orphan blobs left behind by overwrite", async () => { await withDB(async (db) => { await writeFile(db, "/x.txt", "first", {}, () => 1000); diff --git a/packages/dofs/src/fs/writeFile.ts b/packages/dofs/src/fs/writeFile.ts index 48f6fc53..d03f1532 100644 --- a/packages/dofs/src/fs/writeFile.ts +++ b/packages/dofs/src/fs/writeFile.ts @@ -211,10 +211,20 @@ async function writeFileStreaming( flush(carry); } - // Wire up the inode against the staged blobs in one short - // transaction. From this point on the SQL is the same shape as the - // synchronous path — only the chunk-bytes step is skipped because - // stageBlob already landed them above. + linkStagedChunksSync(db, canonical, parts, chunkRefs, { ...options, mode }, mtime); +} + +// Link a path to chunks already staged in content-addressed storage. +// This keeps payload bytes out of memory during sync apply. +export function linkStagedChunksSync( + db: Database, + canonical: string, + parts: string[], + chunkRefs: { hash: Uint8Array; size: number }[], + options: WriteFileOptions, + mtime: number, +): void { + const mode = (options.mode ?? 0o644) & 0o7777; db.transactionSync(() => { const parentInode = resolveParent(db, parts, canonical); const leafName = parts[parts.length - 1]; @@ -251,6 +261,8 @@ async function writeFileStreaming( ref.size, ); } + // Referenced chunks cannot be collected, so last_seen only protects + // blobs during the staging window before this transaction. const manifestHash = buildManifest(db, chunkRefs, mtime); const rev = incrementRev(db); let totalSize = 0; diff --git a/packages/dofs/src/sync/apply.test.ts b/packages/dofs/src/sync/apply.test.ts index 5ced0d0e..baadeb46 100644 --- a/packages/dofs/src/sync/apply.test.ts +++ b/packages/dofs/src/sync/apply.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import { link } from "../fs/link.js"; import { mkdir } from "../fs/mkdir.js"; @@ -10,8 +10,9 @@ import { resolveInode } from "../fs/resolve.js"; import { rm } from "../fs/rm.js"; import { symlink } from "../fs/symlink.js"; import { withDB, withTwoDBs } from "../fs/with-db.js"; -import { writeFile, writeFileSync } from "../fs/writeFile.js"; +import { CHUNK_SIZE, writeFile, writeFileSync } from "../fs/writeFile.js"; import { applyChanges, applyChangesSync } from "./apply.js"; +import { stageBlob } from "./blobs.js"; import type { ChangeEntry } from "./changes.js"; import { coalesceChanges } from "./coalesce.js"; import { fetchObjects } from "./fetch.js"; @@ -73,6 +74,74 @@ describe("applyChanges", () => { ); }); + it("links staged chunks without reading payload bytes", async () => { + const content = `${"a".repeat(CHUNK_SIZE)}b`; + await withTwoDBs( + async (a) => { + await writeFile(a, "/large.txt", content, {}, () => 1); + const entries = await drain(coalesceChanges(a, 0)); + const entry = entries.find((candidate) => candidate.path === "/large.txt"); + if (entry?.kind !== "file") throw new Error("missing file entry"); + expect(entry.chunks).toHaveLength(2); + return { entry, objects: await collectObjects(a, [entry]) }; + }, + async (b, { entry, objects }) => { + for (const chunk of entry.chunks) { + const bytes = objects.get(hex(chunk.hash)); + if (bytes === undefined) throw new Error("missing chunk bytes"); + stageBlob(b, chunk.hash, bytes, 2); + } + + const all = vi.spyOn(b, "all"); + try { + await applyChanges(b, [entry], new Map(), { source: "upstream" }); + const payloadReads = all.mock.calls.filter(([query]) => + query.includes("SELECT bytes FROM vfs_blob_bytes"), + ); + expect(all.mock.calls.length).toBeGreaterThan(0); + expect(payloadReads).toHaveLength(0); + } finally { + all.mockRestore(); + } + expect(await readFile(b, "/large.txt", "utf8")).toBe(content); + }, + ); + }); + + it("links staged chunks synchronously without reading payload bytes", async () => { + const content = `${"a".repeat(CHUNK_SIZE)}b`; + await withTwoDBs( + async (a) => { + await writeFile(a, "/large.txt", content, {}, () => 1); + const entries = await drain(coalesceChanges(a, 0)); + const entry = entries.find((candidate) => candidate.path === "/large.txt"); + if (entry?.kind !== "file") throw new Error("missing file entry"); + expect(entry.chunks).toHaveLength(2); + return { entry, objects: await collectObjects(a, [entry]) }; + }, + async (b, { entry, objects }) => { + for (const chunk of entry.chunks) { + const bytes = objects.get(hex(chunk.hash)); + if (bytes === undefined) throw new Error("missing chunk bytes"); + stageBlob(b, chunk.hash, bytes, 2); + } + + const all = vi.spyOn(b, "all"); + try { + applyChangesSync(b, [entry], new Map(), { source: "upstream" }); + const payloadReads = all.mock.calls.filter(([query]) => + query.includes("SELECT bytes FROM vfs_blob_bytes"), + ); + expect(all.mock.calls.length).toBeGreaterThan(0); + expect(payloadReads).toHaveLength(0); + } finally { + all.mockRestore(); + } + expect(await readFile(b, "/large.txt", "utf8")).toBe(content); + }, + ); + }); + it("commits in batches capped by byte budget", async () => { // Force many small files; with a tiny byte budget the apply // path should still converge, just across more batches. We diff --git a/packages/dofs/src/sync/apply.ts b/packages/dofs/src/sync/apply.ts index 4986c98e..d11974fe 100644 --- a/packages/dofs/src/sync/apply.ts +++ b/packages/dofs/src/sync/apply.ts @@ -5,10 +5,11 @@ import { invalidateResolveSubtree } from "../fs/resolveCache.js"; import { rm } from "../fs/rm.js"; import { symlink } from "../fs/symlink.js"; import { unlinkDirent } from "../fs/unlink.js"; -import { writeFile, writeFileSync } from "../fs/writeFile.js"; +import { linkStagedChunksSync } from "../fs/writeFile.js"; import { canonicalizePath } from "../path.js"; import { incrementRev } from "../rev.js"; import type { Database } from "../storage.js"; +import { stageBlob } from "./blobs.js"; import type { ChangeEntry } from "./changes.js"; import { computeManifestHash } from "./manifests.js"; @@ -289,35 +290,7 @@ export async function applyChanges( if (pathsInBatch >= maxPaths) flush(); continue; } - // file: assemble chunk bytes. First check the in-memory map - // (the streaming hand-off); fall back to vfs_blob_bytes (the - // staged-via-pushObjects path). - const parts: Uint8Array[] = []; - let total = 0; - for (const c of entry.chunks) { - const k = hex(c.hash); - let bytes = objects.get(k); - if (bytes === undefined) { - const row = db.one<{ bytes: Uint8Array }>( - "SELECT bytes FROM vfs_blob_bytes WHERE hash = ?", - c.hash, - ); - bytes = row?.bytes; - } - if (bytes === undefined) { - throw new Error(`applyChanges: missing object ${k} for ${entry.path}`); - } - parts.push(bytes); - total += bytes.byteLength; - } - const buf = new Uint8Array(total); - let off = 0; - for (const p of parts) { - buf.set(p, off); - off += p.byteLength; - } - removeReplaceableFinalEntry(db, entry.path, "file"); - await writeFile(db, entry.path, buf, { mode: entry.mode }, () => entry.mtime); + const total = applyFileEntry(db, entry, objects); applied++; bytesInBatch += total; pathsInBatch++; @@ -408,32 +381,7 @@ export function applyChangesSync( if (pathsInBatch >= maxPaths) flush(); continue; } - const parts: Uint8Array[] = []; - let total = 0; - for (const c of entry.chunks) { - const k = hex(c.hash); - let bytes = objects.get(k); - if (bytes === undefined) { - const row = db.one<{ bytes: Uint8Array }>( - "SELECT bytes FROM vfs_blob_bytes WHERE hash = ?", - c.hash, - ); - bytes = row?.bytes; - } - if (bytes === undefined) { - throw new Error(`applyChanges: missing object ${k} for ${entry.path}`); - } - parts.push(bytes); - total += bytes.byteLength; - } - const buf = new Uint8Array(total); - let off = 0; - for (const p of parts) { - buf.set(p, off); - off += p.byteLength; - } - removeReplaceableFinalEntry(db, entry.path, "file"); - writeFileSync(db, entry.path, buf, { mode: entry.mode }, () => entry.mtime); + const total = applyFileEntry(db, entry, objects); applied++; bytesInBatch += total; pathsInBatch++; @@ -448,6 +396,58 @@ export function applyChangesSync( return { applied, skipped }; } +// Link a file entry to staged chunks without loading payload bytes. +// In-memory objects are staged individually before the link. + +// Validate declared sizes without loading payloads; chunk hashes remain +// trusted here, matching stageBlob's existing contract. +function applyFileEntry( + db: Database, + entry: Extract, + objects: Map, +): number { + let total = 0; + for (const c of entry.chunks) { + total += c.size; + const staged = stagedBlobSize(db, c.hash); + if (staged === undefined) { + const k = hex(c.hash); + const bytes = objects.get(k); + if (bytes === undefined) { + throw new Error(`applyChanges: missing object ${k} for ${entry.path}`); + } + assertChunkSize(bytes.byteLength, c.size, c.hash, entry.path); + stageBlob(db, c.hash, bytes, entry.mtime); + continue; + } + assertChunkSize(staged, c.size, c.hash, entry.path); + } + removeReplaceableFinalEntry(db, entry.path, "file"); + const { parts, path: canonical } = canonicalizePath(entry.path); + linkStagedChunksSync(db, canonical, parts, entry.chunks, { mode: entry.mode }, entry.mtime); + return total; +} + +// Return a staged chunk's size without loading its payload bytes. +// A short byte row is treated as an interrupted write. +function stagedBlobSize(db: Database, hash: Uint8Array): number | undefined { + return db.one<{ size: number }>( + `SELECT b.size AS size + FROM vfs_blobs b + JOIN vfs_blob_bytes bb ON bb.hash = b.hash + WHERE b.hash = ? + AND length(bb.bytes) = b.size`, + hash, + )?.size; +} + +function assertChunkSize(actual: number, declared: number, hash: Uint8Array, path: string): void { + if (actual === declared) return; + throw new Error( + `applyChanges: chunk ${hex(hash)} for ${path} declares ${declared} bytes but holds ${actual}`, + ); +} + // Compare an entry against the local node graph. Returns true when // the entry would be a no-op apply: the manifest hash (files), mode // (dirs), or mode + symlink target (symlinks) already matches. From 91139bc889d447d712e8dc90ef45d58a5910b4a0 Mon Sep 17 00:00:00 2001 From: aron <263346377+aron-cf@users.noreply.github.com> Date: Fri, 7 Aug 2026 15:16:11 +0000 Subject: [PATCH 2/2] dofs: Guard the staged-chunk link path linkStagedChunksSync writes chunk rows from a list it did not produce, so the checks writeFile used to run on the way into the sync apply path have to live in the helper itself. Positional reads find the chunk covering an offset by dividing that offset by CHUNK_SIZE, and take a chunk's start offset to be its index times CHUNK_SIZE. A local writer chunks with chunksOf and satisfies that by construction, but a chunk list that arrived over the wire only satisfies it while both sides window at the same size. Reject a list whose interior chunks are short or whose chunks overflow a window, so a sender that windows differently fails loudly instead of producing a file whose bytes read back from the wrong offsets. Also restore the read-only mount check. applyChanges gates read-only mount roots itself and reports the skipped entries, so behavior over sync is unchanged, but the exported helper is a write primitive and the guard belongs at the data layer. Run both checks in applyFileEntry before the existing entry at the path is removed. A batch is a sequence of independent transactions, so a throw after the removal would drop a file and put nothing in its place. --- .changeset/link-staged-chunks-on-apply.md | 6 ++ packages/dofs/src/fs/mount-guard.test.ts | 28 ++++++- packages/dofs/src/fs/writeFile.ts | 27 +++++++ packages/dofs/src/sync/apply.test.ts | 99 +++++++++++++++++++++++ packages/dofs/src/sync/apply.ts | 13 ++- 5 files changed, 168 insertions(+), 5 deletions(-) create mode 100644 .changeset/link-staged-chunks-on-apply.md diff --git a/.changeset/link-staged-chunks-on-apply.md b/.changeset/link-staged-chunks-on-apply.md new file mode 100644 index 00000000..d247baa3 --- /dev/null +++ b/.changeset/link-staged-chunks-on-apply.md @@ -0,0 +1,6 @@ +--- +"@cloudflare/dofs": patch +"@cloudflare/computer": patch +--- + +Cut peak memory during a sync pull. Applying a file entry now links the chunks the sender already staged instead of reading them back and joining them into one whole-file buffer, which used to hold roughly twice the file size in the isolate at once. diff --git a/packages/dofs/src/fs/mount-guard.test.ts b/packages/dofs/src/fs/mount-guard.test.ts index cc729a48..8e4d71c4 100644 --- a/packages/dofs/src/fs/mount-guard.test.ts +++ b/packages/dofs/src/fs/mount-guard.test.ts @@ -1,6 +1,9 @@ +import { createHash } from "node:crypto"; + import { describe, expect, it } from "vitest"; import type { Database } from "../storage.js"; +import { stageBlob } from "../sync/blobs.js"; import { mkdir } from "./mkdir.js"; import { assertNotReadOnly, @@ -12,7 +15,7 @@ import { resolveInode } from "./resolve.js"; import { rm } from "./rm.js"; import { symlink } from "./symlink.js"; import { withDB } from "./with-db.js"; -import { writeFile, writeFileSync } from "./writeFile.js"; +import { linkStagedChunksSync, writeFile, writeFileSync } from "./writeFile.js"; // Stage a read-only mount the way the workspace-side indexer // eventually will: a row in `_vfs_mounts` plus an actual subtree @@ -141,6 +144,29 @@ describe("writeFile under a read-only mount", () => { }); }); + it("rejects linkStagedChunksSync under the mount root with EROFS", async () => { + await withDB(async (db) => { + mkdir(db, "/workspace/r2", { recursive: true }, () => 0); + stageMount(db, "/workspace/r2", "read-only"); + + const bytes = new TextEncoder().encode("blocked"); + const hash = new Uint8Array(createHash("sha256").update(bytes).digest()); + stageBlob(db, hash, bytes, 0); + + expect(() => + linkStagedChunksSync( + db, + "/workspace/r2/hello.txt", + ["workspace", "r2", "hello.txt"], + [{ hash, size: bytes.byteLength }], + {}, + 0, + ), + ).toThrow(/EROFS|read-only/); + expect(resolveInode(db, "/workspace/r2/hello.txt")).toBeNull(); + }); + }); + it("allows writes under a read-write mount", async () => { await withDB(async (db) => { mkdir(db, "/workspace/rw", { recursive: true }, () => 0); diff --git a/packages/dofs/src/fs/writeFile.ts b/packages/dofs/src/fs/writeFile.ts index d03f1532..512e8b4d 100644 --- a/packages/dofs/src/fs/writeFile.ts +++ b/packages/dofs/src/fs/writeFile.ts @@ -214,8 +214,33 @@ async function writeFileStreaming( linkStagedChunksSync(db, canonical, parts, chunkRefs, { ...options, mode }, mtime); } +// Reject a chunk list that positional reads could not address. +// readRangeSync finds the chunk covering an offset by dividing that +// offset by CHUNK_SIZE, and takes a chunk's start offset to be its +// index times CHUNK_SIZE, so every chunk but the last has to fill a +// whole window and none may overflow one. Local writers chunk with +// chunksOf and satisfy this by construction; a chunk list that +// arrived over sync does not have to. +export function assertChunkWindows(chunkRefs: ChunkRef[], canonical: string): void { + for (let idx = 0; idx < chunkRefs.length; idx++) { + const { size } = chunkRefs[idx]; + const last = idx === chunkRefs.length - 1; + if (size === CHUNK_SIZE || (last && size < CHUNK_SIZE)) continue; + throw createWorkspaceError( + "EINVAL", + `chunk ${idx} of ${chunkRefs.length} holds ${size} bytes; only the last chunk may be shorter than ${CHUNK_SIZE}: ${canonical}`, + canonical, + ); + } +} + // Link a path to chunks already staged in content-addressed storage. // This keeps payload bytes out of memory during sync apply. +// +// The chunk list comes from a caller that did its own chunking, so +// the guards every other write path gets from writeFile have to run +// here too: the read-only mount check, and the fixed-window layout +// that positional reads depend on. export function linkStagedChunksSync( db: Database, canonical: string, @@ -224,6 +249,8 @@ export function linkStagedChunksSync( options: WriteFileOptions, mtime: number, ): void { + assertNotReadOnly(db, canonical); + assertChunkWindows(chunkRefs, canonical); const mode = (options.mode ?? 0o644) & 0o7777; db.transactionSync(() => { const parentInode = resolveParent(db, parts, canonical); diff --git a/packages/dofs/src/sync/apply.test.ts b/packages/dofs/src/sync/apply.test.ts index baadeb46..8c3245c2 100644 --- a/packages/dofs/src/sync/apply.test.ts +++ b/packages/dofs/src/sync/apply.test.ts @@ -1,3 +1,5 @@ +import { createHash } from "node:crypto"; + import { describe, expect, it, vi } from "vitest"; import { link } from "../fs/link.js"; @@ -142,6 +144,103 @@ describe("applyChanges", () => { ); }); + it("rejects a file entry whose interior chunks are not chunk-aligned", async () => { + // Positional reads locate a chunk by dividing the offset by + // CHUNK_SIZE, so only the final chunk may be short. Linking a + // sender's chunk list verbatim has to enforce that. + await withDB(async (db) => { + const first = new TextEncoder().encode("first"); + const second = new TextEncoder().encode("second"); + const chunks = [first, second].map((bytes) => { + const hash = new Uint8Array(createHash("sha256").update(bytes).digest()); + stageBlob(db, hash, bytes, 1000); + return { hash, size: bytes.byteLength }; + }); + + await expect( + applyChanges( + db, + [ + { + kind: "file", + rev: 1, + path: "/ragged.txt", + mode: 0o644, + mtime: 1000, + size: first.byteLength + second.byteLength, + chunks, + }, + ], + new Map(), + { source: "upstream" }, + ), + ).rejects.toThrow(/chunk/); + expect(resolveInode(db, "/ragged.txt")).toBeNull(); + }); + }); + + it("leaves the existing file in place when it rejects a ragged entry", async () => { + await withDB(async (db) => { + await writeFile(db, "/keep.txt", "original", {}, () => 1000); + + const first = new TextEncoder().encode("first"); + const second = new TextEncoder().encode("second"); + const chunks = [first, second].map((bytes) => { + const hash = new Uint8Array(createHash("sha256").update(bytes).digest()); + stageBlob(db, hash, bytes, 1000); + return { hash, size: bytes.byteLength }; + }); + + await expect( + applyChanges( + db, + [ + { + kind: "file", + rev: 2, + path: "/keep.txt", + mode: 0o644, + mtime: 2000, + size: first.byteLength + second.byteLength, + chunks, + }, + ], + new Map(), + { source: "upstream" }, + ), + ).rejects.toThrow(/chunk/); + expect(await readFile(db, "/keep.txt", "utf8")).toBe("original"); + }); + }); + + it("rejects a file entry with a chunk larger than the chunk size", async () => { + await withDB(async (db) => { + const bytes = new Uint8Array(CHUNK_SIZE + 1); + const hash = new Uint8Array(createHash("sha256").update(bytes).digest()); + stageBlob(db, hash, bytes, 1000); + + await expect( + applyChanges( + db, + [ + { + kind: "file", + rev: 1, + path: "/oversized.bin", + mode: 0o644, + mtime: 1000, + size: bytes.byteLength, + chunks: [{ hash, size: bytes.byteLength }], + }, + ], + new Map(), + { source: "upstream" }, + ), + ).rejects.toThrow(/chunk/); + expect(resolveInode(db, "/oversized.bin")).toBeNull(); + }); + }); + it("commits in batches capped by byte budget", async () => { // Force many small files; with a tiny byte budget the apply // path should still converge, just across more batches. We diff --git a/packages/dofs/src/sync/apply.ts b/packages/dofs/src/sync/apply.ts index d11974fe..b9df856c 100644 --- a/packages/dofs/src/sync/apply.ts +++ b/packages/dofs/src/sync/apply.ts @@ -5,7 +5,7 @@ import { invalidateResolveSubtree } from "../fs/resolveCache.js"; import { rm } from "../fs/rm.js"; import { symlink } from "../fs/symlink.js"; import { unlinkDirent } from "../fs/unlink.js"; -import { linkStagedChunksSync } from "../fs/writeFile.js"; +import { assertChunkWindows, linkStagedChunksSync } from "../fs/writeFile.js"; import { canonicalizePath } from "../path.js"; import { incrementRev } from "../rev.js"; import type { Database } from "../storage.js"; @@ -397,15 +397,20 @@ export function applyChangesSync( } // Link a file entry to staged chunks without loading payload bytes. -// In-memory objects are staged individually before the link. - -// Validate declared sizes without loading payloads; chunk hashes remain +// In-memory objects are staged individually before the link. Declared +// sizes are validated without loading payloads; chunk hashes remain // trusted here, matching stageBlob's existing contract. +// +// Every check runs before removeReplaceableFinalEntry, so a rejected +// entry leaves whatever was already at the path alone. Batches are a +// sequence of independent transactions, so a throw part way through +// does not roll the removal back. function applyFileEntry( db: Database, entry: Extract, objects: Map, ): number { + assertChunkWindows(entry.chunks, entry.path); let total = 0; for (const c of entry.chunks) { total += c.size;