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.

Author: aron-cf

.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.

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);

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);

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

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<ChangeEntry, { kind: "file" }>,
   objects: Map<string, Uint8Array>,
 ): number {
+  assertChunkWindows(entry.chunks, entry.path);
   let total = 0;
   for (const c of entry.chunks) {
     total += c.size;