diff --git a/doc/opencode-cold-recovery.md b/doc/opencode-cold-recovery.md index e30a8aa7e2a..dd26b43c0ad 100644 --- a/doc/opencode-cold-recovery.md +++ b/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 diff --git a/server/src/services/native-runtime/provider-pack-archive.test.ts b/server/src/services/native-runtime/provider-pack-archive.test.ts index e602f887646..ec1b1ec1916 100644 --- a/server/src/services/native-runtime/provider-pack-archive.test.ts +++ b/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 }); diff --git a/server/src/services/native-runtime/provider-pack-archive.ts b/server/src/services/native-runtime/provider-pack-archive.ts index 0afc4187896..87e1938e432 100644 --- a/server/src/services/native-runtime/provider-pack-archive.ts +++ b/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! }; } diff --git a/server/src/services/native-runtime/provider-pack-layers.ts b/server/src/services/native-runtime/provider-pack-layers.ts index 158c1ca9103..c2b8382e15b 100644 --- a/server/src/services/native-runtime/provider-pack-layers.ts +++ b/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 { +export async function readProviderPackLayers(root: string, expectedDigest: string, value: unknown, signal?: AbortSignal): Promise { 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 = (name: string, fn: () => Promise) => 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, diff --git a/server/src/services/native-runtime/provider-pack-upload.ts b/server/src/services/native-runtime/provider-pack-upload.ts index 1a8606c268e..3826036f20f 100644 --- a/server/src/services/native-runtime/provider-pack-upload.ts +++ b/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, diff --git a/server/src/services/native-runtime/retained-provider-pack-source.test.ts b/server/src/services/native-runtime/retained-provider-pack-source.test.ts index 38b0ffe2639..1897c08f17d 100644 --- a/server/src/services/native-runtime/retained-provider-pack-source.test.ts +++ b/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(done => { enter = done; }); + const gate = new Promise(done => { release = done; }); + const secondJoined = new Promise(done => { joined = done; }); + const repeated = new Promise(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"); +}); diff --git a/server/src/services/native-runtime/retained-provider-pack-source.ts b/server/src/services/native-runtime/retained-provider-pack-source.ts index 0e8c1b130b8..f3a1f293a31 100644 --- a/server/src/services/native-runtime/retained-provider-pack-source.ts +++ b/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 };