fix: defer archive source replacement during harness preparation

Wrap only archive input filesystem operations in the existing typed source error boundary. A disappearing source can defer online while database, output-file and cancellation failures remain fatal.

Prove a source renamed after preparation still allows the next backup to publish, and that an output ENOENT stops the same batch. All 68 backup and archive tests pass.

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

server/src/services/native-runtime/native-harness-backup-archive.ts

@@ -4,6 +4,7 @@ import { join, posix } from "node:path";
 import { Transform } from "node:stream";
 import { pipeline } from "node:stream/promises";
 import tar from "tar-stream";
+import { readNativeHarnessBackupSourceInput } from "./native-harness-backup-source.js";
 
 export const MAX_NATIVE_HARNESS_BYTES = 4 * 1024 * 1024 * 1024;
 export const MAX_NATIVE_HARNESS_OBJECT_BYTES = MAX_NATIVE_HARNESS_BYTES + 768 * 1024 * 1024;
@@ -45,7 +46,7 @@ const boundedArchive = (limit = MAX_NATIVE_HARNESS_OBJECT_BYTES) => {
 
 export async function captureNativeHarnessArchive(root: string, target: string, signal?: AbortSignal, maxArchiveBytes = MAX_NATIVE_HARNESS_OBJECT_BYTES) {
   if (!Number.isSafeInteger(maxArchiveBytes) || maxArchiveBytes <= 0 || maxArchiveBytes > MAX_NATIVE_HARNESS_OBJECT_BYTES) throw fail();
-  if (!(await lstat(root)).isDirectory()) throw fail();
+  if (!(await readNativeHarnessBackupSourceInput(() => lstat(root))).isDirectory()) throw fail();
   const archive = tar.pack();
   let outputFailure: unknown;
   const output = pipeline(archive, boundedArchive(maxArchiveBytes), createWriteStream(target, { flags: "wx", mode: 0o600 }), { signal });
@@ -53,11 +54,11 @@ export async function captureNativeHarnessArchive(root: string, target: string,
   let count = 0, bytes = 0;
   const links = new Map<string, string>();
   const visit = async (relative: string): Promise<void> => {
-    for (const name of (await readdir(join(root, relative))).sort()) {
+    for (const name of (await readNativeHarnessBackupSourceInput(() => readdir(join(root, relative)))).sort()) {
       signal?.throwIfAborted();
       const entry = relative ? `${relative}/${name}` : name;
       if (!safeName(entry) || ++count > MAX_NATIVE_HARNESS_ENTRIES) throw fail();
-      const file = join(root, entry), stat = await lstat(file);
+      const file = join(root, entry), stat = await readNativeHarnessBackupSourceInput(() => lstat(file));
       if (stat.isDirectory()) {
         await new Promise<void>((resolve, reject) => archive.entry({ name: entry, type: "directory", mode: stat.mode & 0o777 },
           error => error ? reject(error) : resolve()));
@@ -65,17 +66,17 @@ export async function captureNativeHarnessArchive(root: string, target: string,
       } else if (stat.isFile()) {
         bytes += stat.size;
         if (bytes > MAX_NATIVE_HARNESS_BYTES) throw fail();
-        const handle = await open(file, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK);
+        const handle = await readNativeHarnessBackupSourceInput(() => open(file, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK));
         try {
-          const before = await handle.stat();
+          const before = await readNativeHarnessBackupSourceInput(() => handle.stat());
           if (!before.isFile() || before.ino !== stat.ino || before.dev !== stat.dev || before.size !== stat.size) throw fail();
           const stream = archive.entry({ name: entry, size: stat.size, type: "file", mode: stat.mode & 0o777 });
           await pipeline(handle.createReadStream({ autoClose: false }), stream, { signal });
-          const after = await handle.stat();
+          const after = await readNativeHarnessBackupSourceInput(() => handle.stat());
           if (before.size !== after.size || before.mtimeMs !== after.mtimeMs || before.ctimeMs !== after.ctimeMs) throw fail();
         } finally { await handle.close(); }
       } else if (stat.isSymbolicLink()) {
-        const linkname = await readlink(file);
+        const linkname = await readNativeHarnessBackupSourceInput(() => readlink(file));
         if (!safeLink(entry, linkname)) throw fail();
         links.set(entry, linkname);
         await new Promise<void>((resolve, reject) => archive.entry({ name: entry, type: "symlink", linkname },

server/src/services/native-runtime/native-harness-backup-source.ts

@@ -5,16 +5,28 @@ import { join } from "node:path";
  * retain their identity, even if their class or code resembles a source error. */
 export class NativeHarnessBackupSourceError extends Error {}
 
+/** Wrap input reads individually, so archive output and database errors never
+ * become deferrable just because they share a filesystem error code. */
+export async function readNativeHarnessBackupSourceInput<T>(read: () => Promise<T>): Promise<T> {
+  try { return await read(); }
+  catch (error) {
+    if (["ENOENT", "ENOTDIR", "EISDIR", "ELOOP", "EACCES", "EPERM"].includes((error as NodeJS.ErrnoException)?.code ?? "")) {
+      throw new NativeHarnessBackupSourceError("native_harness_migration_source_unavailable", { cause: error });
+    }
+    throw error;
+  }
+}
+
 export async function readNativeHarnessBackupSource(root: string) {
   try {
-    const stat = await lstat(root).catch((error: NodeJS.ErrnoException) => {
+    const stat = await readNativeHarnessBackupSourceInput(() => lstat(root).catch((error: NodeJS.ErrnoException) => {
       if (error.code === "ENOENT") return null; throw error;
-    });
+    }));
     if (!stat) return null;
     if (!stat.isDirectory() || stat.isSymbolicLink()) throw new NativeHarnessBackupSourceError("native_harness_migration_unsafe_source");
-    const manifestBytes = await readFile(join(root, "manifest.json"));
+    const manifestBytes = await readNativeHarnessBackupSourceInput(() => readFile(join(root, "manifest.json")));
     const manifest: unknown = JSON.parse(manifestBytes.toString("utf8"));
-    const state: unknown = JSON.parse(await readFile(join(root, "runner", "runner-state.json"), "utf8"));
+    const state: unknown = JSON.parse(await readNativeHarnessBackupSourceInput(() => readFile(join(root, "runner", "runner-state.json"), "utf8")));
     if (!manifest || typeof manifest !== "object" || Array.isArray(manifest)
       || !state || typeof state !== "object" || Array.isArray(state)) {
       throw new NativeHarnessBackupSourceError("native_harness_migration_source_unsettled");
@@ -22,9 +34,6 @@ export async function readNativeHarnessBackupSource(root: string) {
     return { manifestBytes, manifest: manifest as Record<string, unknown>, state: state as Record<string, unknown> };
   } catch (error) {
     if (error instanceof SyntaxError) throw new NativeHarnessBackupSourceError("native_harness_migration_source_invalid_json", { cause: error });
-    if (["ENOENT", "ENOTDIR", "EISDIR", "ELOOP", "EACCES", "EPERM"].includes((error as NodeJS.ErrnoException)?.code ?? "")) {
-      throw new NativeHarnessBackupSourceError("native_harness_migration_source_unavailable", { cause: error });
-    }
     throw error;
   }
 }

server/src/services/native-runtime/native-harness-backup-storage.integration.test.ts

@@ -15,6 +15,7 @@ import { createNativeHarnessBackupStamp, verifyNativeHarnessBackupStamp, verifyN
 import { hasLocalNativeHarnessObligations, migrateNativeHarnessBackups, prestageNativeHarnessBackups } from "./native-harness-backup-migration.js";
 import { nativeHarnessBackupExpirationService } from "./native-harness-backup-expiration.js";
 import { legacyControllerBootId } from "../legacy-controller-lease.js";
+import * as backupArchive from "./native-harness-backup-archive.js";
 
 describeEmbeddedPostgres("durable native provider backups", () => {
   const environment = nativeRestartFixture();
@@ -238,6 +239,31 @@ describeEmbeddedPostgres("durable native provider backups", () => {
       await symlink(f.root + "-retained", f.root);
     } else await writeFile(join(f.root, kind === "manifest" ? "manifest.json" : "runner/runner-state.json"), "{");
   }
+  it.each(["source", "output"])("distinguishes %s ENOENT during archive capture after source preparation", async kind => {
+    const [bad, good] = await releasedFixtures(), capture = backupArchive.captureNativeHarnessArchive;
+    const warning = vi.spyOn(console, "warn").mockImplementation(() => {});
+    const archive = vi.spyOn(backupArchive, "captureNativeHarnessArchive").mockImplementation(async (root, target, signal, limit) => {
+      if (root === bad.root) {
+        if (kind === "source") await rename(root, root + "-retained");
+        else target = join(target, "missing-parent", "archive");
+      }
+      return capture(root, target, signal, limit);
+    });
+    try {
+      const preparation = prestageNativeHarnessBackups(bad.db, { storage: bad.storage, signal: AbortSignal.timeout(15_000) });
+      if (kind === "source") {
+        expect(await preparation).toEqual({ inspected: 2, published: 1, current: 0, deferred: { source_changed: 1 } });
+        expect(JSON.parse(warning.mock.calls[0][0])).toMatchObject({ scopeDigest: bad.scopeDigest,
+          reason: "native_harness_migration_source_unavailable" });
+        expect((await good.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, good.scopeDigest)))[0].status).toBe("current");
+      } else {
+        await expect(preparation).rejects.toMatchObject({ code: "ENOENT" });
+        expect(warning).not.toHaveBeenCalled();
+        expect(await good.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, good.scopeDigest))).toHaveLength(0);
+      }
+      expect(await bad.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, bad.scopeDigest))).toHaveLength(0);
+    } finally { archive.mockRestore(); warning.mockRestore(); }
+  });
   it.each(["manifest", "runner", "unsafe"])("defers a damaged %s source and still copies the next eligible backup", async kind => {
     const [bad, good] = await releasedFixtures(), warning = vi.spyOn(console, "warn").mockImplementation(() => {});
     try {