From 7d46532c6d26b094c2e3d220e79a78cee576b982 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 4 Nov 2026 18:16:36 -0600 Subject: [PATCH] fix(runtime): bound parallel offline harness backup transfers (#19108) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - An ephemeral web server must retain provider history outside its local disk. > - Offline enrollment moves verified local harness backups to object storage. > - A cold rehearsal reached 699 current backup rows while processing 1,953 sources before its 25-minute deadline. > - The serial transfer loop prevents this existing dataset from enrolling. > - This pull request overlaps at most four transfers within the existing total scratch limit. > - The benefit is faster preservation with the same ownership and publication checks. ## Linked Issues or Issue Description Refs #19038, #18943, #18932. **What happened?** Offline enrollment processed 1,953 local provider backups one at a time. The normal Cloud image reached 699 current backup rows (including nine pre-existing rows) before maintenance expired. The serving instance remained unchanged during the isolated test. **Expected behavior** Enrollment must preserve every source within bounded time and disk use. It must stop all in-flight transfers before it reports failure. **Steps to reproduce** 1. Prepare a settled instance with multiple verified native harness backups. 2. Run offline enrollment against shared object storage. 3. Observe that the serial owner waits for each archive, upload, readback and publication before starting the next source. ## What Changed - Process groups of up to four verified offline harness backups. - Reserve space for the archive and encrypted copy before each group starts. Keep the total scratch limit equal to one maximum-size serial transfer. - Derive a smaller archive limit from a closed manifest directory set. Preserve unlisted files with an exclusive maximum-size reservation. - Enforce the reservation while streaming. Keep the existing maximum for ordinary runtime backups. - Cancel and await peer transfers after failure. Keep completed publications valid for retry. - Preserve the original archive-stream failure when the tar writer also fails after destruction. - Document the migration bound and add integration regressions for overlap, restoration, cancellation, retry, extra files and archive limits. ## Verification - Serial baseline: the new overlap regression fails with one active transfer when four are required. - Candidate: 59 targeted tests pass with real PostgreSQL, S3, filesystem archives and restored provider bytes. The bundled-entry test is skipped locally because this source worktree has no runtime bundle. The full normal image rehearsal remains required. - Follow-up: 48 archive/storage checks pass after correcting the PAX-header allowance. The real encoder produces a 972,801,024-byte full-entry projection, above the old 805,306,368-byte allowance and below the new 1,024,001,024-byte bound. The existing global object limit remains in force. The old formula fails the new regression. Exact-size restoration and one-byte-over-limit rejection pass. - The previous revision passed full CI after one setup-route timeout retry. That failure remains recorded; the corrected current revision f7c2be1289 passed fresh full CI (41937700826) and Greptile 5/5. Its review thread is resolved. - Local server typecheck found four test-only Promise library errors; those are fixed. The remaining 26 diagnostics come from stale linked dependency declarations outside the changed files. Clean full CI is required before merge. - The resumed isolated staging-copy experiment also exceeded its 25-minute deadline. It reached 1,389 current backups. This improves transfer throughput but does not by itself qualify enrollment; online preparation is tracked in #19137. - Full fresh normal-image migration, empty-home restoration and hosted release handoffs remain deployment acceptance gates. ## Risks - Parallel work increases concurrent object-storage and database requests to at most four transfers. The scratch reservation still bounds archive and ciphertext files together. - A malformed or changed source can fail its streaming limit. The migration retains its source and stops before export. - Partial completed publications survive failure. Retry verifies them under the existing hash and ownership rules. - No serving configuration, deployment pin, database contract or provider lifecycle policy changes in this pull request. ## Model Used OpenAI GPT-6.1 Codex, exact model ID `gpt-6.1-sol`. Used for reasoning, code editing, shell tools and test execution. The runtime does not expose its context-window size. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- ...-existing-workspace-ephemeral-migration.md | 27 ++++ .../native-harness-backup-archive.test.ts | 30 +++- .../native-harness-backup-archive.ts | 32 +++- .../native-harness-backup-migration.ts | 138 ++++++++++++------ ...harness-backup-storage.integration.test.ts | 98 +++++++++++++ .../native-harness-backup-storage.ts | 4 +- 6 files changed, 277 insertions(+), 52 deletions(-) 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 });