fix(native): bound cached pack verification and preflight archive limits

Co-Authored-By: Paperclip <[email protected]>

doc/opencode-cold-recovery.md

@@ -94,8 +94,9 @@ The temporary cache contains public code and source provenance only. It uses
 private, process-account-owned directories and exact manifest keys. An empty
 host can reconstruct the cache from the registry. A cached source is checked
 again before upload. Imports stream data with a four-GiB code limit and a
-three-minute deadline, allow two active imports, and bound queued distinct
-sources to 32. Import waits and downloads respond to cancellation; one cancelled
+three-minute deadline, allow two active imports or cache verifications, and bound queued distinct
+sources to 32. Concurrent callers share verification of the same cached pack.
+Import waits and downloads respond to cancellation; one cancelled
 caller does not cancel a shared import still needed by another caller. Inactive
 instances perform no registry reads or cache scans. Each import requires disk
 headroom before writing; completed cached code remains until temporary storage

server/src/services/native-runtime/provider-pack-archive.test.ts

@@ -66,6 +66,25 @@ it("keeps historical/source packs without an archive on the existing path", asyn
   expect(await readProviderPackArchive("/does-not-exist/provider-pack", digest)).toBeNull();
 });
 
+it.each(["v1", "v2"])("cancels a real %s archive hash after its first read chunk", async version => {
+  const { pack } = version === "v2" ? await layeredFixture() : await fixture();
+  const archivePath = version === "v2" ? `${pack}.archives/overlay.tar.zst` : `${pack}.tar.zst`;
+  const controller = new AbortController();
+  const handle = await fs.open(archivePath, "r");
+  const prototype = Object.getPrototypeOf(handle) as typeof handle;
+  await handle.close();
+  const originalStream = prototype.createReadStream;
+  const streaming = vi.spyOn(prototype, "createReadStream").mockImplementation(function(this: typeof handle, options) {
+    const stream = originalStream.call(this, options);
+    stream.once("data", () => controller.abort(new Error("owned archive read stopped")));
+    return stream;
+  });
+  try {
+    await expect(readProviderPackArchive(pack, digest, controller.signal)).rejects.toThrow(/abort|owned archive read stopped/i);
+    expect(controller.signal.aborted).toBe(true);
+  } finally { streaming.mockRestore(); }
+});
+
 async function layeredFixture() {
   const { root, pack } = await fixture();
   await fs.writeFile(path.join(pack, "bin", "large"), Buffer.alloc(1024 * 1024, 42), { mode: 0o755 });

server/src/services/native-runtime/provider-pack-archive.ts

@@ -7,16 +7,18 @@ import { providerPackExtractionProbe } from "./provider-pack-extraction.js";
 
 const quote = (value: string) => `'${value.replaceAll("'", "'\\''")}'`;
 
-export async function readProviderPackArchive(root: string, expectedDigest: string) {
+export async function readProviderPackArchive(root: string, expectedDigest: string, signal?: AbortSignal) {
+  signal?.throwIfAborted();
   let receipt: { schema?: string; encoding?: string; providerPackDigest?: string; sha256?: string; bytes?: number };
   try {
-    receipt = JSON.parse(await readFile(`${root}.archive.json`, "utf8"));
+    receipt = JSON.parse(await readFile(`${root}.archive.json`, { encoding: "utf8", signal }));
   } catch (error) {
+    signal?.throwIfAborted();
     if ((error as NodeJS.ErrnoException).code === "ENOENT") return null;
     throw invalidArchive("receipt_unreadable", error);
   }
   if (receipt.schema === "paperclip-runner/provider-pack-archive/v2") {
-    return readProviderPackLayers(root, expectedDigest, receipt);
+    return readProviderPackLayers(root, expectedDigest, receipt, signal);
   }
   if (receipt.schema !== "paperclip-runner/provider-pack-archive/v1"
     || receipt.encoding !== "tar+zstd"
@@ -26,7 +28,7 @@ export async function readProviderPackArchive(root: string, expectedDigest: stri
     throw invalidArchive(receipt.providerPackDigest !== expectedDigest ? "receipt_identity_mismatch" : "receipt_invalid");
   }
   const sourcePath = `${root}.tar.zst`;
-  await verifyArchive(sourcePath, { sha256: receipt.sha256!, bytes: receipt.bytes! });
+  await verifyArchive(sourcePath, { sha256: receipt.sha256!, bytes: receipt.bytes! }, signal);
   return { sourcePath, sha256: receipt.sha256!, bytes: receipt.bytes! };
 }
 

server/src/services/native-runtime/provider-pack-layers.ts

@@ -28,7 +28,8 @@ const validArchive = (value: unknown): value is Archive => {
 export const invalidArchive = (reason: string, cause?: unknown) =>
   new Error(`runner_remote_provider_archive_invalid: ${reason}`, cause === undefined ? undefined : { cause });
 
-export async function verifyArchive(path: string, expected: Archive) {
+export async function verifyArchive(path: string, expected: Archive, signal?: AbortSignal) {
+  signal?.throwIfAborted();
   let file;
   try { file = await open(path, constants.O_RDONLY | constants.O_NOFOLLOW); }
   catch (error) { throw invalidArchive("archive_unreadable", error); }
@@ -37,12 +38,13 @@ export async function verifyArchive(path: string, expected: Archive) {
     if (!metadata.isFile()) throw invalidArchive("archive_not_file");
     if (metadata.size !== expected.bytes) throw invalidArchive("size_mismatch");
     const hash = createHash("sha256");
-    for await (const chunk of file.createReadStream({ autoClose: false })) hash.update(chunk);
+    for await (const chunk of file.createReadStream({ autoClose: false, signal })) hash.update(chunk);
+    signal?.throwIfAborted();
     if (hash.digest("hex") !== expected.sha256) throw invalidArchive("sha256_mismatch");
   } finally { await file.close(); }
 }
 
-export async function readProviderPackLayers(root: string, expectedDigest: string, value: unknown): Promise<ProviderPackLayers> {
+export async function readProviderPackLayers(root: string, expectedDigest: string, value: unknown, signal?: AbortSignal): Promise<ProviderPackLayers> {
   const receipt = value as { encoding: string; providerPackDigest: string; overlay: Archive; files: FileLayer[] };
   if (receipt.providerPackDigest !== expectedDigest) throw invalidArchive("receipt_identity_mismatch");
   if (receipt.encoding !== "tar+zstd"
@@ -70,7 +72,7 @@ export async function readProviderPackLayers(root: string, expectedDigest: strin
   const sourceRoot = `${root}.archives`;
   const layers = await lstat(sourceRoot).catch(error => { throw invalidArchive("layers_unreadable", error); });
   if (!layers.isDirectory()) throw invalidArchive("layers_unreadable");
-  await verifyArchive(`${sourceRoot}/overlay.tar.zst`, receipt.overlay);
+  await verifyArchive(`${sourceRoot}/overlay.tar.zst`, receipt.overlay, signal);
   return { kind: "layers", sourceRoot, overlay: receipt.overlay, files: receipt.files };
 }
 
@@ -110,6 +112,11 @@ const reuseScript = `
 })().catch(()=>process.exitCode=1);
 `;
 
+export function providerPackLayersArgument(archive: ProviderPackLayers, targetPath: string, reuseRoots: readonly string[]) {
+  const spec = Buffer.from(JSON.stringify({ target: targetPath, roots: reuseRoots, files: archive.files })).toString("base64");
+  return Buffer.byteLength(spec) > 120_000 ? null : spec;
+}
+
 export async function stageProviderPackLayers(input: {
   archive: ProviderPackLayers;
   runner: CommandManagedRuntimeRunner;
@@ -120,9 +127,9 @@ export async function stageProviderPackLayers(input: {
   const { archive, runner } = input;
   if (!runner.syncIn) throw Error("runner_remote_provider_archive_transport_unavailable");
   const measure = <T>(name: string, fn: () => Promise<T>) => input.trace ? input.trace.measure(name, fn) : fn();
-  const spec = Buffer.from(JSON.stringify({ target: input.targetPath, roots: input.reuseRoots, files: archive.files })).toString("base64");
+  const spec = providerPackLayersArgument(archive, input.targetPath, input.reuseRoots);
   // Respect the per-argument Linux limit; unusual packs retain directory staging.
-  if (Buffer.byteLength(spec) > 120_000) return false;
+  if (spec === null) return false;
   const capability = await measure("provider_pack.archive.probe", () => runner.execute({
     command: "sh", args: ["-c", `${providerPackExtractionProbe} && node -e 'process.exit(Number(process.versions.node.split(".")[0])>=18?0:1)'`],
     bypassSession: true, timeoutMs: 10_000,

server/src/services/native-runtime/provider-pack-upload.ts

@@ -4,6 +4,7 @@ import { readProviderPackArchive, stageProviderPackArchive } from "./provider-pa
 import type { ProviderPackIdentity } from "./remote-provider-pack-verification.js";
 import { resolveProviderPackUploadSource } from "./retained-provider-pack-source.js";
 import { providerPackExtractionProbe } from "./provider-pack-extraction.js";
+import { providerPackLayersArgument } from "./provider-pack-layers.js";
 
 const quote = (value: string) => `'${value.replaceAll("'", "'\\''")}'`;
 
@@ -40,9 +41,12 @@ export async function uploadVerifiedProviderPack(input: {
   // Hash the directory only when it is actually the upload source. The
   // installed pack still passes full verification against the selected identity.
   const archive = await measure("provider_pack.archive.verify", () =>
-    readProviderPackArchive(source.root, input.expectedDigest));
+    readProviderPackArchive(source.root, input.expectedDigest, input.signal));
   if (source.archiveOnly) {
     if (!archive) throw Error("runner_remote_provider_source_invalid");
+    if ("kind" in archive && providerPackLayersArgument(archive, input.targetPath, input.reuseRoots) === null) {
+      throw Error("runner_remote_provider_archive_transport_unavailable");
+    }
     const capability = await measure("provider_pack.archive.probe", () => input.runner.execute({
       command: "sh", args: ["-c", `${providerPackExtractionProbe}${"kind" in archive ? " && node -e 'process.exit(Number(process.versions.node.split(\".\")[0])>=18?0:1)'" : ""}`],
       bypassSession: true, timeoutMs: 10_000,

server/src/services/native-runtime/retained-provider-pack-source.test.ts

@@ -11,6 +11,8 @@ import tar from "tar-stream";
 import { afterEach, expect, it, vi } from "vitest";
 import { canonicalProviderPackJson } from "./remote-provider-pack-verification.js";
 import { uploadVerifiedProviderPack } from "./provider-pack-upload.js";
+import * as archiveOwner from "./provider-pack-archive.js";
+import { resolveProviderPackUploadSource } from "./retained-provider-pack-source.js";
 const archiveScript = pathToFileURL(path.resolve(import.meta.dirname, "../../../../scripts/create-provider-pack-archive.mjs")).href;
 const { createProviderPackLayers } = await import(archiveScript);
 
@@ -40,8 +42,9 @@ async function fixture(options: { wrongCode?: boolean; unsafeOverlay?: boolean;
   const makePack = async (name: string, tool: string, source: string) => {
     const pack = path.join(root, name);
     await fs.mkdir(path.join(pack, "bin"), { recursive: true });
+    const artifact = { path: "bin/tool", sha256: `sha256:${hash(tool)}` };
     const payload = { pins: { opencode: "1.18.34" }, target: { platform: "linux", architecture: "x64" }, runnerSourceRevision: source,
-      contractDigest: `sha256:${hash(tool)}`, artifacts: { opencodeProxy: { path: "bin/tool", sha256: `sha256:${hash(tool)}` } } };
+      contractDigest: `sha256:${hash(tool)}`, artifacts: { nodeCommand: artifact, opencodeCommand: artifact, opencodeProxy: artifact } };
     const manifest = { schema: "paperclip-runner/remote-provider-pack/v1", payload, digest: `sha256:${hash(canonicalProviderPackJson(payload))}` };
     await fs.writeFile(path.join(pack, "provider-pack.json"), JSON.stringify(manifest));
     await fs.writeFile(path.join(pack, "bin/tool"), tool, { mode: 0o755 });
@@ -276,6 +279,58 @@ it("reuses a verified derived cache without registry reads and reconstructs it a
   expect(await fs.readFile(path.join(input.targetPath + "-empty", "bin/tool"), "utf8")).toBe("retained provider code");
 });
 
+it("shares warm-cache verification while a stopped caller leaves another waiter active", async () => {
+  const input = await fixture();
+  await upload(input);
+  const requests = input.requests.length;
+  const firstController = new AbortController(), secondController = new AbortController();
+  let enter!: () => void, release!: () => void, joined!: () => void, duplicate!: () => void;
+  const entered = new Promise<void>(done => { enter = done; });
+  const gate = new Promise<void>(done => { release = done; });
+  const secondJoined = new Promise<void>(done => { joined = done; });
+  const repeated = new Promise<void>(done => { duplicate = done; });
+  const originalRead = archiveOwner.readProviderPackArchive;
+  const verification = vi.spyOn(archiveOwner, "readProviderPackArchive").mockImplementation(async (...args) => {
+    enter();
+    if (verification.mock.calls.length > 1) duplicate();
+    await gate;
+    return originalRead(...args);
+  });
+  const originalAny = AbortSignal.any;
+  vi.spyOn(AbortSignal, "any").mockImplementation(signals => {
+    const combined = originalAny(signals);
+    if (signals.includes(secondController.signal)) {
+      const add = combined.addEventListener.bind(combined);
+      vi.spyOn(combined, "addEventListener").mockImplementation((type, listener, options) => {
+        const result = add(type, listener, options);
+        if (type === "abort") joined();
+        return result;
+      });
+    }
+    return combined;
+  });
+  const source = { root: input.current.pack, sourceDigest: input.current.manifest.digest, manifest: input.selected.manifest };
+  const first = resolveProviderPackUploadSource({ ...source, signal: firstController.signal });
+  const firstStopped = expect(first).rejects.toThrow("owned shared-cache waiter stopped");
+  await entered;
+  const second = resolveProviderPackUploadSource({ ...source, signal: secondController.signal });
+  try {
+    await Promise.race([secondJoined, repeated.then(() => { throw Error("warm-cache verification repeated"); })]);
+    expect(verification).toHaveBeenCalledOnce();
+    firstController.abort(new Error("owned shared-cache waiter stopped"));
+    await firstStopped;
+    expect(verification.mock.calls[0]?.[2]?.aborted).toBe(false);
+    release();
+    expect(await second).toMatchObject({ archiveOnly: true });
+    expect(input.requests).toHaveLength(requests);
+  } finally {
+    firstController.abort(new Error("owned shared-cache waiter stopped"));
+    secondController.abort(new Error("owned shared-cache fixture cleanup"));
+    release();
+    await Promise.allSettled([first, second, firstStopped]);
+  }
+});
+
 it("rejects a changed cached archive before any remote command or transfer", async () => {
   const input = await fixture();
   await upload(input);
@@ -312,3 +367,14 @@ it("leaves a target intact when its archive transport cannot install the verifie
   expect(input.runner.syncIn).not.toHaveBeenCalled();
   expect(await fs.readFile(path.join(input.targetPath, "marker"), "utf8")).toBe("untouched");
 });
+
+it("refuses an oversized historical layer specification before any target effects", async () => {
+  const input = await fixture();
+  await fs.mkdir(input.targetPath);
+  await fs.writeFile(path.join(input.targetPath, "marker"), "untouched");
+  const reuseRoots = Array.from({ length: 2000 }, (_, index) => `/qualified-image-${index}/` + "a".repeat(64));
+  await expect(upload(input, { reuseRoots })).rejects.toThrow("runner_remote_provider_archive_transport_unavailable");
+  expect(input.runner.execute).not.toHaveBeenCalled();
+  expect(input.runner.syncIn).not.toHaveBeenCalled();
+  expect(await fs.readFile(path.join(input.targetPath, "marker"), "utf8")).toBe("untouched");
+});

server/src/services/native-runtime/retained-provider-pack-source.ts

@@ -76,7 +76,7 @@ async function selectedArchive(root: string, manifest: ProviderPackIdentity, sig
   signal.throwIfAborted();
   const value = JSON.parse((await readStable(join(root, "provider-pack.json"))).toString()) as ProviderPackIdentity;
   if (canonicalProviderPackJson(value) !== canonicalProviderPackJson(manifest)) fail();
-  const archive = await readProviderPackArchive(root, manifest.digest);
+  const archive = await readProviderPackArchive(root, manifest.digest, signal);
   if (!archive || !("kind" in archive)) fail();
   const names = ["overlay.tar.zst", ...archive.files.map((_file, index) => `${index}.tar.zst`)];
   if (canonicalProviderPackJson((await readdir(archive.sourceRoot)).sort()) !== canonicalProviderPackJson(names.sort())) fail();
@@ -92,7 +92,7 @@ async function selectedArchive(root: string, manifest: ProviderPackIdentity, sig
 }
 
 async function extractManifest(root: string, manifest: ProviderPackIdentity, signal: AbortSignal) {
-  const source = await readProviderPackArchive(root, manifest.digest);
+  const source = await readProviderPackArchive(root, manifest.digest, signal);
   if (!source || !("kind" in source)) fail();
   const regular = new Set(source.files.map(file => file.path));
   const archive = tar.extract();
@@ -281,8 +281,6 @@ export async function resolveProviderPackUploadSource(input: {
   await privateDirectory(cache);
   const signal = input.signal ? AbortSignal.any([input.signal, AbortSignal.timeout(180_000)]) : AbortSignal.timeout(180_000);
   const pointer = join(cache, manifest.digest.slice(7) + ".json");
-  const retained = await cached(cache, pointer, manifest, signal);
-  if (retained) return retained;
   const key = `${cache}:${manifest.digest}`;
   let job = pending.get(key);
   if (!job || job.controller.signal.aborted) {
@@ -292,7 +290,7 @@ export async function resolveProviderPackUploadSource(input: {
     let created!: ImportJob;
     const operation = (async () => {
       const release = await importSlot(sharedSignal);
-      try { return await importSource(cache, pointer, manifest, sharedSignal); }
+      try { return await cached(cache, pointer, manifest, sharedSignal) ?? await importSource(cache, pointer, manifest, sharedSignal); }
       finally { release(); }
     })().finally(() => { if (pending.get(key) === created) pending.delete(key); });
     job = created = { operation, controller, waiters: 0 };