fix(checkpoint): bound artifact member IO

loopx/control_plane/runtime/checkpoint_artifacts.ts

@@ -21,6 +21,23 @@ const SCHEMA = "loopx_checkpoint_artifact_set_v0";
 type Component = {id: string; source_path: string};
 type Member = {path: string; kind: "file" | "directory"; mode: number; size?: number; sha256?: string};
 
+// Keep independent member IO bounded, including on failure: staging cleanup
+// must not race copies or flushes that are still running after a rejection.
+async function mapMembers<T, R>(items: T[], visit: (item: T) => Promise<R>): Promise<R[]> {
+  const result = new Array<R>(items.length);
+  let next = 0, failed = false;
+  let failure: unknown;
+  await Promise.all(Array.from({length: Math.min(8, items.length)}, async () => {
+    while (!failed && next < items.length) {
+      const index = next++;
+      try {result[index] = await visit(items[index]);}
+      catch (error) {if (!failed) {failed = true; failure = error;}}
+    }
+  }));
+  if (failed) throw failure;
+  return result;
+}
+
 function absolute(value: unknown, label: string): string {
   const path = requireNonEmptyString(value, label);
   if (!isAbsolute(path) || resolve(path) !== path) throw new Error(`${label} must be normalized and absolute`);
@@ -61,21 +78,41 @@ async function members(root: string, sourceObservation: boolean): Promise<Member
   await physical(root);
   const output: Member[] = [];
   async function visit(path: string, name: string): Promise<void> {
-    const stat = await lstat(path);
-    if (stat.isSymbolicLink()) throw new Error("checkpoint source contains a symlink");
-    // SQLite's reader index and an empty WAL contain no committed data. A
-    // read-only backup can create either on a closed WAL database. Exclude
-    // only these recognized sidecars, not a nonempty WAL or an explicitly
-    // requested standalone file. Artifact verification excludes nothing.
-    if (sourceObservation && name !== "" && stat.isFile() &&
-        (await sqliteSidecar(path, ["-shm"]) ||
-         (stat.size === 0 && await sqliteSidecar(path, ["-wal"])))) return;
-    if (stat.isDirectory()) {
-      output.push({path: name, kind: "directory", mode: stat.mode & 0o777});
-      for (const child of (await readdir(path)).sort()) await visit(join(path, child), name ? `${name}/${child}` : child);
-    } else if (stat.isFile()) {
-      output.push({path: name, kind: "file", mode: stat.mode & 0o777, size: stat.size, sha256: await digest(path)});
-    } else throw new Error("checkpoint source contains an unsupported file type");
+    type Node = {member?: Member; children: Node[]};
+    const tree: Node = {children: []};
+    let frontier = [{path, name, node: tree}];
+    while (frontier.length) {
+      frontier = (await mapMembers(frontier, async entry => {
+        const {path, name, node} = entry, stat = await lstat(path);
+        if (stat.isSymbolicLink()) throw new Error("checkpoint source contains a symlink");
+        // SQLite's reader index and an empty WAL contain no committed data. A
+        // read-only backup can create either on a closed WAL database. Exclude
+        // only these recognized sidecars, not a nonempty WAL or an explicitly
+        // requested standalone file. Artifact verification excludes nothing.
+        if (sourceObservation && name !== "" && stat.isFile() &&
+            (await sqliteSidecar(path, ["-shm"]) ||
+             (stat.size === 0 && await sqliteSidecar(path, ["-wal"])))) return [];
+        if (stat.isDirectory()) {
+          node.member = {path: name, kind: "directory", mode: stat.mode & 0o777};
+          return (await readdir(path)).sort().map(child => {
+            const childNode: Node = {children: []};
+            node.children.push(childNode);
+            return {path: join(path, child), name: name ? `${name}/${child}` : child, node: childNode};
+          });
+        }
+        if (!stat.isFile()) throw new Error("checkpoint source contains an unsupported file type");
+        node.member = {path: name, kind: "file", mode: stat.mode & 0o777, size: stat.size, sha256: await digest(path)};
+        return [];
+      })).flat();
+    }
+    // Traversal concurrency cannot change the existing depth-first manifest
+    // order (which differs from sorting complete relative path strings).
+    const pending = [tree];
+    while (pending.length) {
+      const node = pending.pop()!;
+      if (node.member) output.push(node.member);
+      for (let i = node.children.length - 1; i >= 0; i--) pending.push(node.children[i]);
+    }
   }
   await visit(root, "");
   if ((await lstat(root)).isFile() && await sqliteFile(root)) {
@@ -137,28 +174,27 @@ async function flushCopiedFile(path: string): Promise<void> {
   }
 }
 async function copyComponent(source: string, destination: string, snapshotDatabases = true): Promise<void> {
-  const stat = await lstat(source);
-  if (stat.isSymbolicLink()) throw new Error("checkpoint source became a symlink");
-  if (stat.isDirectory()) {
-    await mkdir(destination, {mode: 0o700});
-    const children = (await readdir(source)).sort();
-    const databases = new Set<string>();
-    for (const child of snapshotDatabases ? children : []) {
-      const path = join(source, child);
-      if ((await lstat(path)).isFile()) {
-        if (await sqliteFile(path)) databases.add(child);
+  let frontier = [{source, destination, nested: false}];
+  while (frontier.length) {
+    frontier = (await mapMembers(frontier, async entry => {
+      const {source, destination, nested} = entry, stat = await lstat(source);
+      if (stat.isSymbolicLink()) throw new Error("checkpoint source became a symlink");
+      // An explicitly selected standalone sidecar remains a byte artifact.
+      // Directory participants seal their databases through the existing WAL
+      // snapshot owner rather than copying those databases' live sidecars.
+      if (snapshotDatabases && nested && await sqliteSidecar(source, ["-wal", "-shm", "-journal"])) return [];
+      if (stat.isDirectory()) {
+        await mkdir(destination, {mode: 0o700});
+        return (await readdir(source)).sort().map(child => ({source: join(source, child),
+          destination: join(destination, child), nested: true}));
       }
-    }
-    for (const child of children) {
-      if (["-wal", "-shm", "-journal"].some(suffix => child.endsWith(suffix) && databases.has(child.slice(0, -suffix.length)))) continue;
-      await copyComponent(join(source, child), join(destination, child), snapshotDatabases);
-    }
-  } else if (stat.isFile()) {
-    // SQLite's existing physical snapshot owner seals committed WAL contents.
-    if (snapshotDatabases && await sqliteFile(source)) await snapshotSqliteDatabase(source, destination);
-    else await copyFile(source, destination);
-    await flushCopiedFile(destination);
-  } else throw new Error("checkpoint source contains an unsupported file type");
+      if (!stat.isFile()) throw new Error("checkpoint source contains an unsupported file type");
+      if (snapshotDatabases && await sqliteFile(source)) await snapshotSqliteDatabase(source, destination);
+      else await copyFile(source, destination);
+      await flushCopiedFile(destination);
+      return [];
+    })).flat();
+  }
 }
 function manifestBody(raw: unknown): JsonObject {
   const value = requireJsonObject(raw, "checkpoint manifest");
@@ -357,15 +393,17 @@ export async function checkpointArtifacts(value: unknown): Promise<JsonObject> {
     // that could otherwise escape a before/after comparison. SQLite is sealed
     // by its existing snapshot owner and has a separate physical witness.
     for (const c of sources) {
-      for (const member of before[c.id] as Member[]) {
-        if (member.kind !== "file") continue;
-        const source = (await lstat(c.source_path)).isDirectory() ? join(c.source_path, member.path) : c.source_path + member.path;
-        if (await sqliteFile(source) || (member.path !== "" && await sqliteSidecar(source, ["-wal", "-shm", "-journal"]))) continue;
-        const copied = (artifacts[c.id] as Member[]).find(m => m.path === member.path && m.kind === "file");
+      const directory = (await lstat(c.source_path)).isDirectory();
+      const copiedMembers = new Map((artifacts[c.id] as Member[]).filter(m => m.kind === "file").map(m => [m.path, m]));
+      await mapMembers(before[c.id] as Member[], async member => {
+        if (member.kind !== "file") return;
+        const source = directory ? join(c.source_path, member.path) : c.source_path + member.path;
+        if (await sqliteFile(source) || (member.path !== "" && await sqliteSidecar(source, ["-wal", "-shm", "-journal"]))) return;
+        const copied = copiedMembers.get(member.path);
         if (!copied || copied.sha256 !== member.sha256 || copied.size !== member.size || copied.mode !== member.mode) {
           throw new Error("checkpoint copied bytes do not match the reviewed source");
         }
-      }
+      });
     }
     const body: JsonObject = {schema_version: SCHEMA, components: sources, point_metadata: requireJsonObject(request.point_metadata, "checkpoint point metadata"),
       source_witness: before, artifact_witness: artifacts, research_point_qualified: false,

tests/control_plane_ts/checkpoint_artifacts.test.ts

@@ -1,4 +1,7 @@
 import assert from "node:assert/strict";
+import {createHash} from "node:crypto";
+import fs from "node:fs/promises";
+import {syncBuiltinESMExports} from "node:module";
 import test from "node:test";
 import {chmod, mkdir, mkdtemp, readFile, readdir, realpath, rm, stat, symlink, writeFile} from "node:fs/promises";
 import {tmpdir} from "node:os";
@@ -56,6 +59,44 @@ test("a read-only participant's copied mode survives the durability flush and ma
   assert.equal((await stat(restored)).mode & 0o777, 0o444);
 });
 
+test("nested inventories retain depth-first sibling order and every byte and mode", async t => {
+  const {root, request} = await fixture(t);
+  const source = join(root, "tree");
+  await mkdir(source, {mode: 0o700});
+  await mkdir(join(source, "a"), {mode: 0o700});
+  await mkdir(join(source, "a/b"), {mode: 0o700});
+  const leaves = ["a/b/deep", "a/b-", "a/b.txt", "a-", "a.txt",
+    ...Array.from({length: 64}, (_, i) => `z${String(i).padStart(2, "0")}`)];
+  for (const [i, name] of [...leaves].reverse().entries()) {
+    await writeFile(join(source, name), `evidence:${name}\n`, {mode: i % 2 ? 0o600 : 0o400});
+  }
+  // The filesystem owns its supported permission bits (Windows exposes a
+  // different subset). Observe those independently before calling capture.
+  const modes = new Map<string, number>();
+  for (const name of ["", "a", "a/b", ...leaves]) modes.set(name, (await stat(join(source, name))).mode & 0o777);
+  const file = (name: string) => ({path: name, kind: "file", mode: modes.get(name)!,
+    size: Buffer.byteLength(`evidence:${name}\n`), sha256: createHash("sha256").update(`evidence:${name}\n`).digest("hex")});
+  const directory = (name: string) => ({path: name, kind: "directory", mode: modes.get(name)!});
+  // Flat string sorting would put a- before a/b; completion order is also
+  // irrelevant. This is the existing manifest's depth-first traversal contract.
+  const expected = [directory(""), directory("a"), directory("a/b"), ...leaves.map(file)];
+  const components = [{id: "tree", source_path: source}];
+  const probe = await checkpointArtifacts({action: "probe", components});
+  assert.deepEqual(probe.source_witness, {tree: expected});
+  const captured = await checkpointArtifacts({...request, components, expected_source_sha256: probe.source_sha256});
+  const manifest = JSON.parse(await readFile(join(request.destination, "checkpoint.json"), "utf8"));
+  assert.deepEqual(manifest.source_witness, {tree: expected});
+  assert.deepEqual(manifest.artifact_witness, {tree: expected});
+  const restored = join(root, "restored");
+  await checkpointArtifacts({action: "materialize", root: request.destination,
+    expected_sha256: captured.sha256, destination: restored});
+  for (const name of leaves) {
+    const copy = join(restored, "components/tree", name);
+    assert.equal(await readFile(copy, "utf8"), `evidence:${name}\n`);
+    assert.equal((await stat(copy)).mode & 0o777, file(name).mode);
+  }
+});
+
 test("changed source, duplicate identities and overlapping sources publish no unit", async t => {
   const {root, request} = await fixture(t);
   await writeFile(join(root, "conversation"), "new native cursor");
@@ -65,6 +106,48 @@ test("changed source, duplicate identities and overlapping sources publish no un
   assert.deepEqual((await readdir(root)).sort(), ["code", "conversation"]);
 });
 
+test("copied bytes reject ABA source movement even when the final source matches the probe", async t => {
+  const {root, request} = await fixture(t), original = fs.copyFile;
+  const source = join(root, "code/solution.sh"), bytes = await readFile(source);
+  const copy = t.mock.method(fs, "copyFile", async (...args: Parameters<typeof fs.copyFile>) => {
+    if (args[0] !== source) return original(...args);
+    await writeFile(source, "unreviewed intermediate bytes\n");
+    try {await original(...args);}
+    finally {await writeFile(source, bytes);}
+  });
+  syncBuiltinESMExports();
+  try {
+    await assert.rejects(checkpointArtifacts(request), /copied bytes do not match/);
+    assert.deepEqual(await readFile(source), bytes);
+    assert.deepEqual((await readdir(root)).sort(), ["code", "conversation"]);
+  } finally {copy.mock.restore(); syncBuiltinESMExports();}
+});
+
+test("a failed copy drains started member IO before removing the unpublished staging tree", async t => {
+  const {root, request} = await fixture(t), original = fs.copyFile;
+  for (let i = 0; i < 24; i++) await writeFile(join(root, "code", `member-${i}`), `retained evidence ${i}`);
+  const probe = await checkpointArtifacts({action: "probe", components: request.components});
+  let active = 0, peak = 0, calls = 0;
+  const copy = t.mock.method(fs, "copyFile", async (...args: Parameters<typeof fs.copyFile>) => {
+    active++; peak = Math.max(peak, active);
+    const first = calls++ === 0;
+    try {
+      // Controlled storage latency lets sibling copies overlap the injected
+      // failure. It is a cleanup invariant, not a performance threshold.
+      await new Promise(resolve => setTimeout(resolve, first ? 30 : 80));
+      if (first) throw new Error("injected member copy failure");
+      await original(...args);
+    } finally {active--;}
+  });
+  syncBuiltinESMExports();
+  try {
+    await assert.rejects(checkpointArtifacts({...request, expected_source_sha256: probe.source_sha256}), /injected member copy failure/);
+    assert.equal(active, 0, "no member IO may outlive staging cleanup");
+    assert.ok(peak <= 8, "member IO must remain bounded");
+    assert.deepEqual((await readdir(root)).sort(), ["code", "conversation"]);
+  } finally {copy.mock.restore(); syncBuiltinESMExports();}
+});
+
 test("missing, changed and extra artifact members or foreign expected digest reject", async t => {
   const {root, request} = await fixture(t), captured = await checkpointArtifacts(request);
   const verify = {action: "verify", root: request.destination, expected_sha256: captured.sha256};