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.

Author: Caio-Nogueira

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

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;

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

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<ChangeEntry, { kind: "file" }>,
+  objects: Map<string, Uint8Array>,
+): 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.