diff --git a/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md b/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md index 098f1ecb34f..a739c0bf59f 100644 --- a/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md +++ b/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md @@ -170,3 +170,30 @@ keeps its supervisor and ownership requirements. For a retired reusable sandbox lease, the existing remote-termination resolver may prove destruction through its exact successor handoff chain. A stopped successor, broken chain or competing tip does not grant archival authority. + +### Bounded offline harness transfers + +The full bundled cold rehearsal reached only 699 of 1,953 local harness backups +before the 25-minute maintenance deadline. Keep every source and its existing +ownership and hash checks. Migrate up to four sources together. Reserve space for +both each tar archive and its encrypted copy before starting a group, with the +same total scratch limit as one maximum-size serial transfer. Use the verified +manifest sizes for a closed directory set; give sources with extra files an +exclusive maximum-size reservation so those files are preserved. Enforce the +reservation while streaming the archive. Ordinary runtime backups keep their +existing maximum archive limit. + +Derive tar overhead from the maximum entry count and both maximum path lengths. +PAX can encode a path and symlink target for every entry. The difference between +the global object and content limits is not a safe per-source overhead estimate. +Check the bound against real maximum-path PAX encoding at the full entry budget, +and check streaming at an exact byte limit and one byte beyond it. + +On failure, cancel and await every transfer in the group before returning. A +completed publication remains valid for the next attempt. Tests must exercise +real PostgreSQL publication and byte restoration, the four-transfer bound, +failure while another storage call is unresolved, retry, extra-file preservation, +and rejection at the archive limit before any object or row is published. This +finite offline operation adds no idle background work. A resumed rehearsal can +measure the fix but cannot replace fresh full-image migration and empty-home +recovery proof. diff --git a/server/src/services/native-runtime/native-harness-backup-archive.test.ts b/server/src/services/native-runtime/native-harness-backup-archive.test.ts index a08ca880f2c..24b6170cc57 100644 --- a/server/src/services/native-runtime/native-harness-backup-archive.test.ts +++ b/server/src/services/native-runtime/native-harness-backup-archive.test.ts @@ -5,7 +5,9 @@ import { tmpdir } from "node:os"; import { pipeline } from "node:stream/promises"; import tar from "tar-stream"; import { afterEach, expect, it } from "vitest"; -import { captureNativeHarnessArchive, restoreNativeHarnessArchive, MAX_NATIVE_HARNESS_BYTES } from "./native-harness-backup-archive.js"; +import { captureNativeHarnessArchive, restoreNativeHarnessArchive, MAX_NATIVE_HARNESS_BYTES, + MAX_NATIVE_HARNESS_ENTRIES, MAX_NATIVE_HARNESS_OBJECT_BYTES, MAX_NATIVE_HARNESS_PATH_BYTES, + nativeHarnessArchiveSizeLimit } from "./native-harness-backup-archive.js"; const roots: string[] = []; afterEach(async () => { for (const root of roots.splice(0)) await rm(root, { recursive: true, force: true }); }); @@ -16,6 +18,32 @@ async function maliciousArchive(root: string, headers: Array) { for (const header of headers) archive.entry(header, Buffer.alloc(0)); archive.finalize(); await written; return file; } +it("reserves the full entry budget using real maximum-path PAX encoding", async () => { + const root = await fixture(), name = "a/".repeat(MAX_NATIVE_HARNESS_PATH_BYTES / 2 - 1) + "aa"; + const file = await maliciousArchive(root, [{ name, type: "symlink", linkname: "b".repeat(MAX_NATIVE_HARNESS_PATH_BYTES) }]); + // One real encoded record gives the repeated-entry size without creating a + // gigabyte fixture. Each hypothetical entry includes its own PAX records. + const entryBytes = (await stat(file)).size - 1024; + const denseArchiveBytes = entryBytes * MAX_NATIVE_HARNESS_ENTRIES + 1024; + const bound = nativeHarnessArchiveSizeLimit(0); + expect(denseArchiveBytes).toBeGreaterThan(MAX_NATIVE_HARNESS_OBJECT_BYTES - MAX_NATIVE_HARNESS_BYTES); + expect(denseArchiveBytes).toBeLessThanOrEqual(bound); + expect(denseArchiveBytes).toBeGreaterThan(bound * 0.9); + expect(nativeHarnessArchiveSizeLimit(MAX_NATIVE_HARNESS_BYTES)).toBe(MAX_NATIVE_HARNESS_OBJECT_BYTES); +}); +it("captures at the exact archive byte limit and refuses one byte beyond it", async () => { + const root = await fixture(), source = join(root, "source"); await mkdir(source); + await writeFile(join(source, "manifest.json"), "{}"); await writeFile(join(source, "state"), "retained provider bytes"); + const original = join(root, "original"); await captureNativeHarnessArchive(source, original); + const bytes = (await stat(original)).size, exact = join(root, "exact"); + await captureNativeHarnessArchive(source, exact, undefined, bytes); + expect((await stat(exact)).size).toBe(bytes); + await restoreNativeHarnessArchive(exact, join(root, "restored-exact")); + expect(await readFile(join(root, "restored-exact", "state"), "utf8")).toBe("retained provider bytes"); + const exceeded = join(root, "exceeded"); + await expect(captureNativeHarnessArchive(source, exceeded, undefined, bytes - 1)).rejects.toThrow("native_harness_archive_invalid"); + expect((await stat(exceeded)).size).toBeLessThan(bytes); +}); it.each([ { name: "../escaped", type: "file" as const }, { name: "/absolute", type: "file" as const }, diff --git a/server/src/services/native-runtime/native-harness-backup-archive.ts b/server/src/services/native-runtime/native-harness-backup-archive.ts index f6249d5f118..388a516eeea 100644 --- a/server/src/services/native-runtime/native-harness-backup-archive.ts +++ b/server/src/services/native-runtime/native-harness-backup-archive.ts @@ -8,12 +8,22 @@ import tar from "tar-stream"; export const MAX_NATIVE_HARNESS_BYTES = 4 * 1024 * 1024 * 1024; export const MAX_NATIVE_HARNESS_OBJECT_BYTES = MAX_NATIVE_HARNESS_BYTES + 768 * 1024 * 1024; export const MAX_NATIVE_HARNESS_ENTRIES = 100_000; +export const MAX_NATIVE_HARNESS_PATH_BYTES = 4096; +// PAX can store both a full path and a full symlink target for each entry. +// Two KiB also cover PAX framing, both headers and payload padding. The final +// two zero blocks terminate the archive. Keep this separate from content size. +export const MAX_NATIVE_HARNESS_TAR_OVERHEAD_BYTES = MAX_NATIVE_HARNESS_ENTRIES + * (2 * MAX_NATIVE_HARNESS_PATH_BYTES + 2048) + 1024; const fail = () => new Error("native_harness_archive_invalid"); -const safeName = (name: string) => name.length > 0 && Buffer.byteLength(name) <= 4096 && !name.includes("\\") +export function nativeHarnessArchiveSizeLimit(contentBytes: number) { + if (!Number.isSafeInteger(contentBytes) || contentBytes < 0 || contentBytes > MAX_NATIVE_HARNESS_BYTES) throw fail(); + return Math.min(MAX_NATIVE_HARNESS_OBJECT_BYTES, contentBytes + MAX_NATIVE_HARNESS_TAR_OVERHEAD_BYTES); +} +const safeName = (name: string) => name.length > 0 && Buffer.byteLength(name) <= MAX_NATIVE_HARNESS_PATH_BYTES && !name.includes("\\") && !/[\u0000-\u001f\u007f]/u.test(name) && !posix.isAbsolute(name) && name.split("/").every(part => part !== ".." && part !== "." && part !== ""); -const safeLink = (name: string, target: string) => target.length > 0 && Buffer.byteLength(target) <= 4096 +const safeLink = (name: string, target: string) => target.length > 0 && Buffer.byteLength(target) <= MAX_NATIVE_HARNESS_PATH_BYTES && !target.includes("\\") && !/[\u0000-\u001f\u007f]/u.test(target) && !posix.isAbsolute(target) && !["..", "."].includes(posix.normalize(posix.join(posix.dirname(name), target))) && !posix.normalize(posix.join(posix.dirname(name), target)).startsWith("../"); @@ -25,19 +35,21 @@ const validateLinks = (links: Map) => { } } }; -const boundedArchive = () => { +const boundedArchive = (limit = MAX_NATIVE_HARNESS_OBJECT_BYTES) => { let bytes = 0; return new Transform({ transform(chunk, _encoding, callback) { bytes += chunk.length; - callback(bytes > MAX_NATIVE_HARNESS_OBJECT_BYTES ? fail() : null, chunk); + callback(bytes > limit ? fail() : null, chunk); } }); }; -export async function captureNativeHarnessArchive(root: string, target: string, signal?: AbortSignal) { +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(); const archive = tar.pack(); - const output = pipeline(archive, boundedArchive(), createWriteStream(target, { flags: "wx", mode: 0o600 }), { signal }); - void output.catch(() => undefined); + let outputFailure: unknown; + const output = pipeline(archive, boundedArchive(maxArchiveBytes), createWriteStream(target, { flags: "wx", mode: 0o600 }), { signal }); + void output.catch(error => { outputFailure = error; }); let count = 0, bytes = 0; const links = new Map(); const visit = async (relative: string): Promise => { @@ -72,7 +84,11 @@ export async function captureNativeHarnessArchive(root: string, target: string, } }; try { await visit(""); validateLinks(links); archive.finalize(); await output; } - catch (error) { archive.destroy(error as Error); await output.catch(() => undefined); throw error; } + catch (error) { + archive.destroy((outputFailure ?? error) as Error); + await output.catch(() => undefined); + throw outputFailure ?? error; + } } export async function restoreNativeHarnessArchive(source: string, target: string, signal?: AbortSignal) { diff --git a/server/src/services/native-runtime/native-harness-backup-migration.ts b/server/src/services/native-runtime/native-harness-backup-migration.ts index d2d7aa0d6c3..3cb5050b962 100644 --- a/server/src/services/native-runtime/native-harness-backup-migration.ts +++ b/server/src/services/native-runtime/native-harness-backup-migration.ts @@ -6,6 +6,7 @@ import { environmentLeases, heartbeatRuns, nativeHarnessBackups, type Db } from import type { StorageService } from "../../storage/types.js"; import { nativeHarnessStateBase, readNativeHarnessBackupStamp, verifyNativeHarnessBackupStampAsync } from "./native-harness-backup-stamp.js"; import { persistNativeHarnessBackup, verifyDurableNativeHarnessBackup } from "./native-harness-backup-storage.js"; +import { MAX_NATIVE_HARNESS_BYTES, MAX_NATIVE_HARNESS_OBJECT_BYTES, nativeHarnessArchiveSizeLimit } from "./native-harness-backup-archive.js"; import { offlineMigrationExecutionWhere } from "../offline-migration-execution.js"; export async function hasLocalNativeHarnessObligations(db: Db): Promise { @@ -21,6 +22,79 @@ export async function hasLocalNativeHarnessObligations(db: Db): Promise return rows.length > 0; } +const MAX_PARALLEL_BACKUPS = 4; +// An archive and its encrypted copy coexist. Keep the same worst-case scratch +// bound as one maximum-size serial transfer, including tar header overhead. +const MAX_SCRATCH_BYTES = 2 * MAX_NATIVE_HARNESS_OBJECT_BYTES; + +async function prepareSource(base: string, scopeDigest: string) { + const root = join(base, scopeDigest, "failover-backups", "current"); + const stat = await 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 Error("native_harness_migration_unsafe_source"); + const manifestBytes = await readFile(join(root, "manifest.json")); + const manifest = JSON.parse(manifestBytes.toString("utf8")) as Record; + const state = JSON.parse(await readFile(join(root, "runner", "runner-state.json"), "utf8")); + if (state.lifecycle !== "suspended" || typeof state.runId !== "string" + || manifest.schema !== "paperclip.native-harness-backup.v1") throw new Error("native_harness_migration_source_unsettled"); + const names = new Set(); + let bytes = manifestBytes.length, closed = Array.isArray(manifest.directories) && manifest.directories.length > 0; + for (const value of Array.isArray(manifest.directories) ? manifest.directories : []) { + const entry = value as { name?: unknown; bytes?: unknown } | null; + if (!entry || typeof entry.name !== "string" || !/^[A-Za-z0-9._-]+$/.test(entry.name) + || [".", ".."].includes(entry.name) || !Number.isSafeInteger(entry.bytes) || Number(entry.bytes) < 0) { closed = false; break; } + names.add(entry.name); bytes += Number(entry.bytes); + } + const entries = await readdir(root, { withFileTypes: true }); + closed &&= entries.length === names.size + 1 && entries.every(entry => entry.name === "manifest.json" + ? entry.isFile() : names.has(entry.name) && entry.isDirectory() && !entry.isSymbolicLink()); + // Unknown extra files are preserved with an exclusive full-size reservation. + const archiveLimit = closed && Number.isSafeInteger(bytes) && bytes <= MAX_NATIVE_HARNESS_BYTES + ? nativeHarnessArchiveSizeLimit(bytes) + : MAX_NATIVE_HARNESS_OBJECT_BYTES; + return { root, scopeDigest, runId: state.runId as string, archiveLimit, + sessionId: manifest.normalizedSessionId, runnerId: manifest.runnerInstanceId, + manifestSha256: `sha256:${createHash("sha256").update(manifestBytes).digest("hex")}` }; +} + +type Source = NonNullable>>; + +async function migrateSource(db: Db, input: { storage: StorageService; signal: AbortSignal }, source: Source) { + input.signal.throwIfAborted(); + const { root, scopeDigest, manifestSha256 } = source; + const [run] = await db.select({ companyId: heartbeatRuns.companyId, issueId: heartbeatRuns.nativeIssueId, + sessionId: heartbeatRuns.nativeSessionId, runnerId: heartbeatRuns.runnerInstanceId }).from(heartbeatRuns) + .where(eq(heartbeatRuns.id, source.runId)); + if (!run) return 0; + if (!run.issueId || run.sessionId !== source.sessionId || run.runnerId !== source.runnerId) { + throw new Error("native_harness_migration_binding_mismatch"); + } + const [current] = await db.select().from(nativeHarnessBackups).where(and(eq(nativeHarnessBackups.companyId, run.companyId), + eq(nativeHarnessBackups.scopeDigest, scopeDigest), eq(nativeHarnessBackups.status, "current"))); + if (current) { + if (current.manifestSha256 !== manifestSha256) throw new Error("native_harness_migration_durable_source_conflict"); + await verifyDurableNativeHarnessBackup({ db, storage: input.storage, companyId: run.companyId, + scopeDigest, manifestSha256: current.manifestSha256, signal: input.signal }); + return 1; + } + const leases = await db.select({ stamp: sql`${environmentLeases.metadata}->'nativeHarnessBackup'`, + providerLeaseId: environmentLeases.providerLeaseId }).from(environmentLeases).where(and( + eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.issueId, run.issueId), + sql`${environmentLeases.metadata}->'nativeHarnessBackup'->>'manifestSha256' = ${manifestSha256}`)); + let valid = false; + for (const lease of leases) { + const stamp = readNativeHarnessBackupStamp(lease.stamp, lease.providerLeaseId ?? ""); + if (stamp?.sessionScopeSha256 === scopeDigest && await verifyNativeHarnessBackupStampAsync(stamp, lease.providerLeaseId!, input.signal)) { valid = true; break; } + } + if (!valid) throw new Error("native_harness_migration_backup_unverified"); + await persistNativeHarnessBackup({ db, storage: input.storage, root, scopeDigest, signal: input.signal, + maxArchiveBytes: source.archiveLimit, + owner: { kind: "offline_migration", writersStopped: true, companyId: run.companyId, issueId: run.issueId, runId: source.runId } }); + return 1; +} + export async function migrateNativeHarnessBackups(db: Db, input: { storage: StorageService; writersStopped: true; signal: AbortSignal; }): Promise { @@ -31,49 +105,31 @@ export async function migrateNativeHarnessBackups(db: Db, input: { const directories = await readdir(base).catch((error: NodeJS.ErrnoException) => { if (error.code === "ENOENT") return [] as string[]; throw error; }); - let count = 0; + let count = 0, reserved = 0, wave: Source[] = []; + const flush = async () => { + const cancellation = new AbortController(); + const signal = AbortSignal.any([input.signal, cancellation.signal]); + const settled = await Promise.allSettled(wave.map(async source => { + try { return await migrateSource(db, { ...input, signal }, source); } + catch (error) { cancellation.abort(error); throw error; } + })); + // Never return failure while a peer can still upload or publish. Partial + // completed transfers remain valid for the next offline attempt. + if (cancellation.signal.aborted) throw cancellation.signal.reason; + for (const result of settled) { + if (result.status === "rejected") throw result.reason; + count += result.value; + } + wave = []; reserved = 0; + }; for (const scopeDigest of directories.filter(name => /^[a-f0-9]{64}$/.test(name)).sort()) { input.signal.throwIfAborted(); - const root = join(base, scopeDigest, "failover-backups", "current"); - const stat = await lstat(root).catch((error: NodeJS.ErrnoException) => { - if (error.code === "ENOENT") return null; throw error; - }); - if (!stat) continue; - if (!stat.isDirectory() || stat.isSymbolicLink()) throw new Error("native_harness_migration_unsafe_source"); - const manifestBytes = await readFile(join(root, "manifest.json")); - const manifest = JSON.parse(manifestBytes.toString("utf8")); - const state = JSON.parse(await readFile(join(root, "runner", "runner-state.json"), "utf8")); - if (state.lifecycle !== "suspended" || typeof state.runId !== "string" - || manifest.schema !== "paperclip.native-harness-backup.v1") throw new Error("native_harness_migration_source_unsettled"); - const [run] = await db.select({ companyId: heartbeatRuns.companyId, issueId: heartbeatRuns.nativeIssueId, - sessionId: heartbeatRuns.nativeSessionId, runnerId: heartbeatRuns.runnerInstanceId }).from(heartbeatRuns) - .where(eq(heartbeatRuns.id, state.runId)); - if (!run) continue; - if (!run.issueId || run.sessionId !== manifest.normalizedSessionId || run.runnerId !== manifest.runnerInstanceId) { - throw new Error("native_harness_migration_binding_mismatch"); - } - const [current] = await db.select().from(nativeHarnessBackups).where(and(eq(nativeHarnessBackups.companyId, run.companyId), - eq(nativeHarnessBackups.scopeDigest, scopeDigest), eq(nativeHarnessBackups.status, "current"))); - const manifestSha256 = `sha256:${createHash("sha256").update(manifestBytes).digest("hex")}`; - if (current) { - if (current.manifestSha256 !== manifestSha256) throw new Error("native_harness_migration_durable_source_conflict"); - await verifyDurableNativeHarnessBackup({ db, storage: input.storage, companyId: run.companyId, - scopeDigest, manifestSha256: current.manifestSha256, signal: input.signal }); - count++; continue; - } - const leases = await db.select({ stamp: sql`${environmentLeases.metadata}->'nativeHarnessBackup'`, - providerLeaseId: environmentLeases.providerLeaseId }).from(environmentLeases).where(and( - eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.issueId, run.issueId), - sql`${environmentLeases.metadata}->'nativeHarnessBackup'->>'manifestSha256' = ${manifestSha256}`)); - let valid = false; - for (const lease of leases) { - const stamp = readNativeHarnessBackupStamp(lease.stamp, lease.providerLeaseId ?? ""); - if (stamp?.sessionScopeSha256 === scopeDigest && await verifyNativeHarnessBackupStampAsync(stamp, lease.providerLeaseId!, input.signal)) { valid = true; break; } - } - if (!valid) throw new Error("native_harness_migration_backup_unverified"); - await persistNativeHarnessBackup({ db, storage: input.storage, root, scopeDigest, signal: input.signal, - owner: { kind: "offline_migration", writersStopped: true, companyId: run.companyId, issueId: run.issueId, runId: state.runId } }); - count++; + const source = await prepareSource(base, scopeDigest); + if (!source) continue; + const reservation = 2 * source.archiveLimit; + if (wave.length && (wave.length === MAX_PARALLEL_BACKUPS || reserved + reservation > MAX_SCRATCH_BYTES)) await flush(); + wave.push(source); reserved += reservation; } + if (wave.length) await flush(); return count; } diff --git a/server/src/services/native-runtime/native-harness-backup-storage.integration.test.ts b/server/src/services/native-runtime/native-harness-backup-storage.integration.test.ts index 4bcda4b168d..c66b503e2dd 100644 --- a/server/src/services/native-runtime/native-harness-backup-storage.integration.test.ts +++ b/server/src/services/native-runtime/native-harness-backup-storage.integration.test.ts @@ -19,6 +19,11 @@ import { legacyControllerBootId } from "../legacy-controller-lease.js"; describeEmbeddedPostgres("durable native provider backups", () => { const environment = nativeRestartFixture(); const ownedHomes: string[] = []; + function deferred() { + let resolve!: () => void; + const promise = new Promise(done => { resolve = done; }); + return { promise, resolve }; + } afterEach(async () => { for (const home of ownedHomes.splice(0)) await removeNativeHarnessBackup(home); vi.unstubAllEnvs(); @@ -45,6 +50,99 @@ describeEmbeddedPostgres("durable native provider backups", () => { return { ...f, home, root, companyId, owner, scopeDigest, sessionScopeId, storage, identity, manifestSha256: `sha256:${createHash("sha256").update(manifest).digest("hex")}` }; } + async function offlineFixture(base: string) { + const f = await fixture("codex-home"), home = join(base, f.scopeDigest); + await rename(f.home, home); + const root = join(home, "failover-backups", "current"), runnerInstanceId = randomUUID(), providerLeaseId = randomUUID(); + const execution = { provider: { kind: "codex" }, binding: { companyId: f.companyId, issueId: f.issueId, + runId: f.runId, agentId: environment.current.agentId, executionWorkspaceId: randomUUID() }, + workspace: { cwd: "/workspace", repoUrl: null, repoRef: null, branchName: null }, + session: { normalizedSessionId: f.identity.normalizedSessionId, driverKind: "codex_app_server", + protocolVersion: 1, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } } } as never; + const manifest = buildNativeHarnessBackupManifest({ backupRoot: root, execution, runnerInstanceId, + providerSessionIdentity: { providerSessionId: f.identity.providerSessionId }, sourceProviderLeaseId: providerLeaseId }); + await writeFile(join(root, "manifest.json"), JSON.stringify(manifest)); + const stamp = createNativeHarnessBackupStamp({ manifestPath: join(root, "manifest.json"), sessionScopeId: f.sessionScopeId, + authorizedProviderLeaseId: providerLeaseId, runnerInstanceId, normalizedSessionId: f.identity.normalizedSessionId, completedAt: manifest.completedAt }); + await f.db.update(heartbeatRuns).set({ nativeSessionId: f.identity.normalizedSessionId, runnerInstanceId }).where(eq(heartbeatRuns.id, f.runId)); + await f.db.insert(environmentLeases).values({ companyId: f.companyId, issueId: f.issueId, heartbeatRunId: f.runId, + providerLeaseId, status: "expired", metadata: { nativeHarnessBackup: stamp } }); + return { ...f, home, root }; + } + async function offlineFixtures(count: number) { + const base = join(environment.current.runtimeRoot, randomUUID()); ownedHomes.push(base); await mkdir(base); + vi.stubEnv("PAPERCLIP_RUNNER_STATE_DIR", base); + const fixtures = []; + for (let i = 0; i < count; i++) fixtures.push(await offlineFixture(base)); + const runs = await fixtures[0].db.select({ id: heartbeatRuns.id, issueId: heartbeatRuns.nativeIssueId }).from(heartbeatRuns) + .where(eq(heartbeatRuns.companyId, fixtures[0].companyId)); + for (const run of runs) await settle({ ...fixtures[0], runId: run.id, issueId: run.issueId! }, new Date()); + return fixtures.sort((a, b) => a.scopeDigest.localeCompare(b.scopeDigest)); + } + it("overlaps verified offline backups within a four-transfer bound and restores their bytes", async () => { + const fixtures = await offlineFixtures(6), f = fixtures[0], gate = deferred(); + const put = f.storage.putFile.bind(f.storage); let active = 0, peak = 0, entered = 0; + vi.spyOn(f.storage, "putFile").mockImplementation(async input => { + active++; entered++; peak = Math.max(peak, active); + try { await gate.promise; return await put(input); } finally { active--; } + }); + const migration = migrateNativeHarnessBackups(f.db, { storage: f.storage, writersStopped: true, signal: AbortSignal.timeout(15_000) }); + try { await vi.waitFor(() => expect(active).toBe(4), { timeout: 3000 }); } + finally { gate.resolve(); await migration.catch(() => undefined); } + expect(await migration).toBe(6); expect(peak).toBe(4); expect(entered).toBe(6); expect(active).toBe(0); + for (const source of fixtures) { + await removeNativeHarnessBackup(source.home); + const replacement = join(environment.current.runtimeRoot, randomUUID()); ownedHomes.push(replacement); + expect(await restoreDurableNativeHarnessBackup({ ...source, storage: f.storage, backupRoot: replacement })).toBe(true); + expect(JSON.parse(await readFile(join(replacement, "current", "codex-home", "sessions", "history.json"), "utf8"))).toEqual(source.identity); + } + }); + it("waits for cancelled peer transfers before returning failure and permits a complete retry", async () => { + const fixtures = await offlineFixtures(2), f = fixtures[0], peerEntered = deferred(), release = deferred(); + const failed = deferred(), put = f.storage.putFile.bind(f.storage); let entered = 0, active = 0, settled = false; + const spy = vi.spyOn(f.storage, "putFile").mockImplementation(async input => { + const ordinal = ++entered; active++; + try { + if (ordinal === 1) { await peerEntered.promise; failed.resolve(); throw new Error("storage unavailable"); } + peerEntered.resolve(); await release.promise; return await put(input); + } finally { active--; } + }); + const migration = migrateNativeHarnessBackups(f.db, { storage: f.storage, writersStopped: true, signal: AbortSignal.timeout(15_000) }); + const completed = migration.then(() => { settled = true; }, () => { settled = true; }); + try { + await vi.waitFor(() => expect(entered).toBe(2), { timeout: 3000 }); await failed.promise; + await new Promise(resolve => setImmediate(resolve)); expect(settled).toBe(false); + } finally { peerEntered.resolve(); release.resolve(); await completed; } + await expect(migration).rejects.toThrow("storage unavailable"); expect(active).toBe(0); spy.mockRestore(); + for (const source of fixtures) expect(await readFile(join(source.root, "manifest.json"), "utf8")).toContain("paperclip.native-harness-backup.v1"); + expect(await migrateNativeHarnessBackups(f.db, { storage: f.storage, writersStopped: true, signal: AbortSignal.timeout(15_000) })).toBe(2); + for (const source of fixtures) { + const rows = await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, source.scopeDigest)); + expect(rows.filter(row => row.status === "current")).toHaveLength(1); + } + }); + it("preserves unlisted files with an exclusive scratch reservation", async () => { + const fixtures = await offlineFixtures(3), f = fixtures[0], extra = fixtures[1]; + await writeFile(join(extra.root, "retained-extra.txt"), "unlisted surviving bytes"); + const put = f.storage.putFile.bind(f.storage); let active = 0, peak = 0; + vi.spyOn(f.storage, "putFile").mockImplementation(async input => { + active++; peak = Math.max(peak, active); + try { await new Promise(resolve => setImmediate(resolve)); return await put(input); } finally { active--; } + }); + expect(await migrateNativeHarnessBackups(f.db, { storage: f.storage, writersStopped: true, signal: AbortSignal.timeout(15_000) })).toBe(3); + expect(peak).toBe(1); + await removeNativeHarnessBackup(extra.home); + const replacement = join(environment.current.runtimeRoot, randomUUID()); ownedHomes.push(replacement); + expect(await restoreDurableNativeHarnessBackup({ ...extra, storage: f.storage, backupRoot: replacement })).toBe(true); + expect(await readFile(join(replacement, "current", "retained-extra.txt"), "utf8")).toBe("unlisted surviving bytes"); + }); + it("enforces the reserved archive size while streaming before publishing any object", async () => { + const f = await fixture(), put = vi.spyOn(f.storage, "putFile"); + await expect(persistNativeHarnessBackup({ ...f, maxArchiveBytes: 512 })).rejects.toThrow("native_harness_archive_invalid"); + expect(put).not.toHaveBeenCalled(); + expect(await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest))).toHaveLength(0); + expect(JSON.parse(await readFile(join(f.root, "codex", "sessions", "history.json"), "utf8"))).toEqual(f.identity); + }); it.each(["codex", "claude", "opencode", "grok", "new-provider"])("restores %s session bytes after deleting the entire controller home", async provider => { const f = await fixture(provider); await persistNativeHarnessBackup(f); diff --git a/server/src/services/native-runtime/native-harness-backup-storage.ts b/server/src/services/native-runtime/native-harness-backup-storage.ts index b6de9ffa865..8a9880f8785 100644 --- a/server/src/services/native-runtime/native-harness-backup-storage.ts +++ b/server/src/services/native-runtime/native-harness-backup-storage.ts @@ -142,7 +142,7 @@ async function fence(tx: Parameters[0]>[0], owner: export async function persistNativeHarnessBackup(input: { db: Db; storage: StorageService; owner: NativeRunnerCheckpointFenceOwner | OfflineOwner; - scopeDigest: string; root: string; signal?: AbortSignal; + scopeDigest: string; root: string; signal?: AbortSignal; maxArchiveBytes?: number; }): Promise { if (!/^[a-f0-9]{64}$/.test(input.scopeDigest)) fail("scope_invalid"); const manifestBytes = await readFile(join(input.root, "manifest.json")); @@ -152,7 +152,7 @@ export async function persistNativeHarnessBackup(input: { const key = randomBytes(32); try { const archive = join(scratch, "archive"), encrypted = join(scratch, "encrypted"); - await captureNativeHarnessArchive(input.root, archive, input.signal); + await captureNativeHarnessArchive(input.root, archive, input.signal, input.maxArchiveBytes); const plaintext = await digest(archive, input.signal), iv = randomBytes(12); const cipher = createCipheriv("aes-256-gcm", key, iv); cipher.setAAD(aad(binding)); await pipeline(createReadStream(archive), cipher, createWriteStream(encrypted, { flags: "wx", mode: 0o600 }), { signal: input.signal });