perf(workspaces): bound parallel snapshot reads (#16239)

Bound workspace snapshot file reads in batches while preserving snapshot contents, ordering, confinement, and failure settlement.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

packages/adapter-utils/src/sandbox-managed-runtime.ts

@@ -1335,9 +1335,6 @@ export async function prepareSandboxManagedRuntime(input: {
   );
   const workspaceRestoreExclude = mergeExcludes(restoreExclude, repositories.map((repo) => repo.path));
   // The baseline "before" snapshot: a recursive walk of the whole workspace that
-  // `lstat`s every entry and SHA-256-hashes every file's bytes. This is the
-  // dominant cost in the pre-`pack` window — it reads the content of every
-  // non-excluded file, serially — so it earns its own span.
   baselineSnapshot = syncWorkspace
     ? (input.workspaceBaseline ??
       (await runStepSpan("snapshot.baseline", () =>

packages/adapter-utils/src/workspace-restore-merge.ts

@@ -147,21 +147,37 @@ async function hashFile(filePath: string): Promise<string> {
 async function* walkDirectory(
   root: string, exclude: readonly string[], ignored: ReturnType<typeof workspacePathMatcher>, relative = "",
 ): AsyncGenerator<[string, SnapshotEntry]> {
+  const readEntry = async (nextRelative: string): Promise<[string, SnapshotEntry | null]> => {
+    const fullPath = path.join(root, nextRelative);
+    const stats = await fs.lstat(fullPath);
+    if (stats.isDirectory()) return [nextRelative, { kind: "dir" }];
+    if (stats.isSymbolicLink()) return [nextRelative, { kind: "symlink", target: await fs.readlink(fullPath) }];
+    if (stats.isFile()) return [nextRelative, { kind: "file", mode: stats.mode, hash: await hashFile(fullPath) }];
+    return [nextRelative, null];
+  };
+  const visitBatch = async function* (paths: string[]): AsyncGenerator<[string, SnapshotEntry]> {
+    const results = await Promise.allSettled(paths.map(readEntry));
+    for (const result of results) if (result.status === "rejected") throw result.reason;
+    for (const result of results) {
+      if (result.status !== "fulfilled") continue;
+      const [nextRelative, entry] = result.value;
+      if (!entry) continue;
+      yield [nextRelative, entry];
+      if (entry.kind === "dir") yield* walkDirectory(root, exclude, ignored, nextRelative);
+    }
+  };
   const current = relative ? path.join(root, relative) : root;
+  let batch: string[] = [];
   for await (const entry of await fs.opendir(current)) {
     const nextRelative = relative ? path.posix.join(relative, entry.name) : entry.name;
     if (shouldExcludePath(nextRelative, exclude) || ignored.matches(nextRelative)) continue;
-    const fullPath = path.join(root, nextRelative);
-    const stats = await fs.lstat(fullPath);
-    if (stats.isDirectory()) {
-      yield [nextRelative, { kind: "dir" }];
-      yield* walkDirectory(root, exclude, ignored, nextRelative);
-    } else if (stats.isSymbolicLink()) {
-      yield [nextRelative, { kind: "symlink", target: await fs.readlink(fullPath) }];
-    } else if (stats.isFile()) {
-      yield [nextRelative, { kind: "file", mode: stats.mode, hash: await hashFile(fullPath) }];
+    batch.push(nextRelative);
+    if (batch.length === 16) {
+      yield* visitBatch(batch);
+      batch = [];
     }
   }
+  if (batch.length) yield* visitBatch(batch);
 }
 
 async function readSnapshotEntry(root: string, relative: string): Promise<SnapshotEntry | null> {

packages/adapter-utils/src/workspace-snapshot-concurrency.test.ts

@@ -0,0 +1,176 @@
+import { createHash } from "node:crypto";
+import { createReadStream, promises as fs } from "node:fs";
+import os from "node:os";
+import path from "node:path";
+import { setImmediate } from "node:timers/promises";
+import { afterEach, describe, expect, it, vi } from "vitest";
+import {
+  captureDirectorySnapshot,
+  disposeDirectorySnapshot,
+  type SnapshotEntry,
+} from "./workspace-restore-merge.js";
+
+vi.mock("node:fs", async importOriginal => {
+  const original = await importOriginal<typeof import("node:fs")>();
+  return { ...original, createReadStream: vi.fn(original.createReadStream) };
+});
+
+const originalFs = await vi.importActual<typeof import("node:fs")>("node:fs");
+const roots: string[] = [];
+
+async function fixture() {
+  const root = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-snapshot-concurrency-"));
+  roots.push(root);
+  return root;
+}
+
+afterEach(async () => {
+  vi.restoreAllMocks();
+  vi.mocked(createReadStream).mockImplementation(originalFs.createReadStream);
+  await Promise.all(roots.splice(0).map(root => fs.rm(root, { recursive: true, force: true })));
+});
+
+describe("bounded workspace snapshot reads", () => {
+  it("preserves file hashes, modes, nested paths, exclusions, and symlink entries across batches", async () => {
+    const root = await fixture();
+    const outside = await fixture();
+    const expected = new Map<string, SnapshotEntry>();
+    for (const directory of ["nested", "nested/deeper"]) {
+      await fs.mkdir(path.join(root, directory));
+      expected.set(directory, { kind: "dir" });
+    }
+    for (let index = 0; index < 73; index++) {
+      const relative = `${index < 35 ? "" : "nested/deeper/"}${index % 3 === 0 ? "line\nbreak" : "-space name"}-${index}`;
+      const content = Buffer.from(`snapshot ${index}\0bytes`);
+      const filePath = path.join(root, relative);
+      await fs.writeFile(filePath, content);
+      await fs.chmod(filePath, index % 2 === 0 ? 0o600 : 0o444);
+      expected.set(relative, {
+        kind: "file", mode: (await fs.lstat(filePath)).mode,
+        hash: createHash("sha256").update(content).digest("hex"),
+      });
+    }
+    await fs.writeFile(path.join(outside, "private"), "outside bytes");
+    await fs.symlink(outside, path.join(root, "outside-link"));
+    expected.set("outside-link", { kind: "symlink", target: outside });
+    for (const ignored of ["excluded", "ignored"]) {
+      await fs.mkdir(path.join(root, ignored));
+      await fs.writeFile(path.join(root, ignored, "private"), "omitted bytes");
+    }
+    for (const diskBacked of [false, true]) {
+      const snapshot = await captureDirectorySnapshot(root, {
+        exclude: ["excluded"], ignoredPaths: ["ignored"], diskBacked,
+      });
+      try {
+        expect(new Map(snapshot.entries)).toEqual(expected);
+      } finally {
+        await disposeDirectorySnapshot(snapshot);
+      }
+    }
+  });
+
+  it("overlaps reads within a fixed bound even across nested directories", async () => {
+    const root = await fixture();
+    for (const directory of ["", "nested", "nested/deeper"]) {
+      await fs.mkdir(path.join(root, directory), { recursive: true });
+      await Promise.all(Array.from({ length: 35 }, (_, index) =>
+        fs.writeFile(path.join(root, directory, `file-${index}`), String(index))));
+    }
+    let active = 0;
+    let peak = 0;
+    vi.mocked(createReadStream).mockImplementation((...args) => {
+      const stream = originalFs.createReadStream(...args);
+      stream.pause();
+      active++;
+      peak = Math.max(peak, active);
+      const settled = () => { active--; };
+      stream.once("end", settled);
+      stream.once("error", settled);
+      setTimeout(() => stream.resume(), 10);
+      return stream;
+    });
+    const snapshot = await captureDirectorySnapshot(root, { diskBacked: true });
+    try {
+      expect(snapshot.entries.size).toBe(107);
+      expect(peak).toBeGreaterThan(1);
+      expect(peak).toBeLessThanOrEqual(16);
+      expect(active).toBe(0);
+    } finally {
+      await disposeDirectorySnapshot(snapshot);
+    }
+  });
+
+  it("settles started metadata reads before propagating a peer failure", async () => {
+    const root = await fixture();
+    await fs.writeFile(path.join(root, "denied"), "denied");
+    await fs.writeFile(path.join(root, "pending"), "pending");
+    const lstat = fs.lstat.bind(fs);
+    let release!: () => void;
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    let pending = false;
+    let active = 0;
+    vi.spyOn(fs, "lstat").mockImplementation(async (...args: Parameters<typeof fs.lstat>) => {
+      if (String(args[0]) === path.join(root, "denied")) throw Object.assign(new Error("denied"), { code: "EACCES" });
+      active++;
+      pending = true;
+      try {
+        await gate;
+        return await lstat(...args);
+      } finally {
+        active--;
+      }
+    });
+    let settled = false;
+    const result = captureDirectorySnapshot(root, { diskBacked: true }).then(
+      () => { settled = true; return null; },
+      error => { settled = true; return error; },
+    );
+    try {
+      await vi.waitFor(() => expect(pending).toBe(true));
+      expect(settled).toBe(false);
+    } finally {
+      release();
+    }
+    expect(await result).toMatchObject({ code: "EACCES" });
+    expect(active).toBe(0);
+  });
+
+  it("settles delayed peer content reads before returning a content failure", async () => {
+    const root = await fixture();
+    await fs.writeFile(path.join(root, "failed"), "failed");
+    await fs.writeFile(path.join(root, "pending"), "pending bytes");
+    let release!: () => void;
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    let pending = false;
+    let failed = false;
+    let active = 0;
+    vi.mocked(createReadStream).mockImplementation((...args) => {
+      const stream = originalFs.createReadStream(...args);
+      stream.pause();
+      active++;
+      stream.once("end", () => { active--; });
+      stream.once("error", () => { active--; failed = true; });
+      if (String(args[0]) === path.join(root, "failed")) {
+        void setImmediate().then(() => stream.destroy(Object.assign(new Error("read failed"), { code: "EIO" })));
+      } else {
+        pending = true;
+        void gate.then(() => stream.resume());
+      }
+      return stream;
+    });
+    let settled = false;
+    const result = captureDirectorySnapshot(root, { diskBacked: true }).then(
+      () => { settled = true; return null; },
+      error => { settled = true; return error; },
+    );
+    try {
+      await vi.waitFor(() => expect(failed && pending).toBe(true));
+      expect(settled).toBe(false);
+      expect(active).toBe(1);
+    } finally {
+      release();
+    }
+    expect(await result).toMatchObject({ code: "EIO" });
+    expect(active).toBe(0);
+  });
+});