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 d48c93e2b55..887c4b3056b 100644 --- a/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md +++ b/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md @@ -230,3 +230,46 @@ source, export-failure cause or unknown recovery kind retains the existing blocker unless the importer already records that exact legacy recovery source. Live migration keeps its recovery exclusion. Recheck recovery evidence before publishing the saved workspace. + +### Online preparation for large provider histories + +Parallel offline transfer alone did not fit the copied staging dataset in the +maintenance window. A resumed attempt reached 1,389 of 1,953 sources. Keep the +offline deadline and move successful released history before draining the server. +The inventory contains 1,802 such backups, approximately 48.5 GB of the 54.8 GB +declared source bytes. This is preparatory publication under recorded runtime +authority, not permission to detach a volume. + +Run the finite operator entry from a normal released image: + +```sh +PAPERCLIP_NATIVE_HARNESS_PRESTAGE=true node --max-old-space-size=512 server/dist-runtime/native-harness-prestage.mjs +``` + +The entry loads the stored Cloud configuration and requires shared object storage. +It processes one source at a time, has a 90-minute deadline, and stops on shutdown. +It starts no scheduler, does not drain tasks, and always reports +`enrollmentReady: false`. Check serving memory headroom before running a separate +process on a tenant. Keep the deployment on normal releases. + +Only a successful, finished source with a released committed coordinator and an +accepted result can publish through this path. At both database fences, hold the +task share lock and recheck its checkout, related queued/running executions, +recorded session and Runner, exact backup-stamp lease, and unchanged local +manifest and suspended state. A concurrent turn, moved source, revoked stamp or +changed owner defers that source. Unrelated executions continue unchanged. +No lease, result, task or lifecycle state is rewritten to make a source eligible. + +The final offline migration still stops all writers, verifies every durable +object, imports deferred sources and all other data, and exports only its normal +complete eligibility receipt. Qualification starts from the saved pristine +database, records the separate preparation phase, and then proves the normal +bundled offline entry and empty-home restoration. Preparation alone cannot pass +enrollment or active-agent release acceptance. + +The online preparation command loads the configured app key before copying. It +requires a valid existing key and retains that key in its process environment +so a missing file cannot create a replacement during publication. Its database +connections have five-second lock waits and 30-second statement and idle +transaction limits. Cancellation closes the command's clients and has a +ten-second process-exit fallback for a dependency that does not settle. diff --git a/server/scripts/bundle-runtime.mjs b/server/scripts/bundle-runtime.mjs index e36a8c8749a..f3311e90a14 100644 --- a/server/scripts/bundle-runtime.mjs +++ b/server/scripts/bundle-runtime.mjs @@ -71,11 +71,14 @@ function externalizeThirdParty(isBundled, runtimeDir) { }; } -export async function bundleRuntime({ entry, outdir, maintenanceEntry, ephemeralMigrationEntry, isBundled = (specifier) => specifier.startsWith("@paperclipai/") }) { +/** @param {{ entry: string, outdir: string, maintenanceEntry?: string, ephemeralMigrationEntry?: string, + * nativeHarnessPrestageEntry?: string, isBundled?: (specifier: string) => boolean }} options */ +export async function bundleRuntime({ entry, outdir, maintenanceEntry, ephemeralMigrationEntry, nativeHarnessPrestageEntry, isBundled = (specifier) => specifier.startsWith("@paperclipai/") }) { await rm(outdir, { recursive: true, force: true }); const result = await build({ entryPoints: { index: "paperclip-runtime-entry", server: entry, ...(maintenanceEntry ? { "agent-files": maintenanceEntry } : {}), - ...(ephemeralMigrationEntry ? { "ephemeral-migration": ephemeralMigrationEntry } : {}) }, + ...(ephemeralMigrationEntry ? { "ephemeral-migration": ephemeralMigrationEntry } : {}), + ...(nativeHarnessPrestageEntry ? { "native-harness-prestage": nativeHarnessPrestageEntry } : {}) }, outdir, outExtension: { ".js": ".mjs" }, chunkNames: "chunk-[hash]", @@ -105,7 +108,8 @@ async function main() { const started = performance.now(); const summary = await bundleRuntime({ entry: join(serverDir, "dist", "index.js"), outdir: join(serverDir, "dist-runtime"), maintenanceEntry: join(serverDir, "dist", "maintenance", "agent-files.js"), - ephemeralMigrationEntry: join(serverDir, "dist", "maintenance", "ephemeral-migration.js") }); + ephemeralMigrationEntry: join(serverDir, "dist", "maintenance", "ephemeral-migration.js"), + nativeHarnessPrestageEntry: join(serverDir, "dist", "maintenance", "native-harness-prestage.js") }); console.log(`[bundle-runtime] ${summary.inputs} inputs -> ${summary.chunks} chunks in ${Math.round(performance.now() - started)} ms: ${summary.outfile}`); } diff --git a/server/scripts/bundle-runtime.test.mjs b/server/scripts/bundle-runtime.test.mjs index adacbca2871..f0d28cad9d7 100644 --- a/server/scripts/bundle-runtime.test.mjs +++ b/server/scripts/bundle-runtime.test.mjs @@ -89,6 +89,15 @@ describe("bundleRuntime", () => { expect(JSON.parse(stdout)).toEqual({ migrated: 42 }); }); + it("packages online harness preparation as a standalone plain-Node entry", async () => { + const nativeHarnessPrestageEntry = path.join(root, "server", "dist", "prestage.js"); + write(nativeHarnessPrestageEntry, 'import { values } from "@paperclipai/fixture"; console.log(JSON.stringify({ copied: values.doubled, enrollmentReady: false }));'); + const outdir = path.join(root, "server", "dist-runtime"); + await bundleRuntime({ entry: path.join(root, "server", "dist", "index.js"), outdir, nativeHarnessPrestageEntry }); + const { stdout } = await execFileAsync(process.execPath, [path.join(outdir, "native-harness-prestage.mjs")], { timeout: 30_000 }); + expect(JSON.parse(stdout)).toEqual({ copied: 42, enrollmentReady: false }); + }); + it("bundles workspace TypeScript and runs with plain node", async () => { const outdir = path.join(root, "server", "dist-runtime"); const summary = await bundleRuntime({ entry: path.join(root, "server", "dist", "index.js"), outdir }); diff --git a/server/src/maintenance/native-harness-prestage-runtime.test.ts b/server/src/maintenance/native-harness-prestage-runtime.test.ts new file mode 100644 index 00000000000..0712f162eca --- /dev/null +++ b/server/src/maintenance/native-harness-prestage-runtime.test.ts @@ -0,0 +1,78 @@ +import { randomBytes } from "node:crypto"; +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { closeRegisteredClients, createDb } from "@paperclipai/db"; +import { sql } from "drizzle-orm"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; +import { createLocalEncryptedCipher } from "../secrets/local-encrypted-provider.js"; +import { initializeNativeHarnessPrestageKey, nativeHarnessPrestageDatabaseUrl } from "./native-harness-prestage-runtime.js"; + +const roots: string[] = []; +afterEach(async () => { vi.unstubAllEnvs(); await Promise.all(roots.splice(0).map(root => fs.rm(root, { recursive: true, force: true }))); }); + +it("writes backups with the configured app key even after its file disappears", async () => { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "prestage-key-")); roots.push(root); + const keyFile = path.join(root, "custom-secrets.key"), key = randomBytes(32).toString("hex"); + await fs.writeFile(keyFile, key, { mode: 0o600 }); + vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", ""); vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY_FILE", ""); + await initializeNativeHarnessPrestageKey(keyFile); + await fs.unlink(keyFile); + const prepared = createLocalEncryptedCipher("write").encrypt("saved harness key"); + vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", ""); + await fs.writeFile(keyFile, key, { mode: 0o600 }); + expect(createLocalEncryptedCipher("read").decrypt(prepared)).toBe("saved harness key"); +}); + +it("does not create a replacement when the configured key is missing or invalid", async () => { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "prestage-key-missing-")); roots.push(root); + const keyFile = path.join(root, "missing.key"); + await expect(initializeNativeHarnessPrestageKey(keyFile, {})).rejects.toMatchObject({ code: "ENOENT" }); + expect(await fs.stat(keyFile).catch(() => null)).toBeNull(); + await fs.writeFile(keyFile, "invalid", { mode: 0o600 }); + await expect(initializeNativeHarnessPrestageKey(keyFile, {})).rejects.toThrow("native_harness_prestage_key_invalid"); + expect(await fs.readFile(keyFile, "utf8")).toBe("invalid"); +}); + +it("keeps explicit environment key and file precedence", async () => { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "prestage-key-env-")); roots.push(root); + const keyFile = path.join(root, "override.key"), configured = path.join(root, "configured.key"); + const key = randomBytes(32).toString("base64"); + await fs.writeFile(keyFile, key, { mode: 0o600 }); + const env = { PAPERCLIP_SECRETS_MASTER_KEY_FILE: keyFile }; + await initializeNativeHarnessPrestageKey(configured, env); + expect(env).toMatchObject({ PAPERCLIP_SECRETS_MASTER_KEY: key, PAPERCLIP_SECRETS_MASTER_KEY_FILE: keyFile }); + await fs.unlink(keyFile); + await initializeNativeHarnessPrestageKey(configured, env); + expect(await fs.stat(configured).catch(() => null)).toBeNull(); +}); + +const support = await getEmbeddedPostgresTestSupport(); +describe.skipIf(!support.supported)("prestage database waits", () => { + it("bounds a real blocked transaction on every connection", async () => { + const database = await startEmbeddedPostgresTestDatabase("prestage-wait-"); + const source = new URL(database.connectionString); source.searchParams.set("application_name", "prestage-test"); + const url = nativeHarnessPrestageDatabaseUrl(source.href), db = createDb(url, { maxConnections: 2 }); + let release!: () => void, held!: () => void; + const ready = new Promise(resolve => { held = resolve; }); + const lock = db.transaction(async tx => { await tx.execute(sql`select pg_advisory_xact_lock(939193)`); held(); + await new Promise(resolve => { release = resolve; }); }); + try { + await ready; + const [settings] = await db.execute(sql`select current_setting('lock_timeout') lock_timeout, + current_setting('statement_timeout') statement_timeout, current_setting('idle_in_transaction_session_timeout') idle_timeout, + current_setting('application_name') application_name`); + expect(settings).toEqual({ lock_timeout: "5s", statement_timeout: "30s", idle_timeout: "30s", application_name: "prestage-test" }); + const started = Date.now(); + await expect(db.transaction(tx => tx.execute(sql`select pg_advisory_xact_lock(939193)`))) + .rejects.toMatchObject({ cause: { code: "55P03" } }); + expect(Date.now() - started).toBeLessThan(10_000); + const waiting = db.execute(sql`select pg_sleep(20)`); + const rejected = expect(waiting).rejects.toBeTruthy(); + await new Promise(resolve => setTimeout(resolve, 100)); + await closeRegisteredClients(url, { scope: "database" }); + await rejected; + } finally { release(); await lock.catch(() => {}); await closeRegisteredClients(url, { scope: "database" }); await database.cleanup(); } + }, 90_000); +}); diff --git a/server/src/maintenance/native-harness-prestage-runtime.ts b/server/src/maintenance/native-harness-prestage-runtime.ts new file mode 100644 index 00000000000..9aae4ad1b60 --- /dev/null +++ b/server/src/maintenance/native-harness-prestage-runtime.ts @@ -0,0 +1,18 @@ +import fs from "node:fs/promises"; +import { isValidLocalEncryptedMasterKey } from "../secrets/local-encrypted-provider.js"; + +export async function initializeNativeHarnessPrestageKey(keyFilePath: string, env: NodeJS.ProcessEnv = process.env) { + const path = env.PAPERCLIP_SECRETS_MASTER_KEY_FILE?.trim() || keyFilePath; + const key = env.PAPERCLIP_SECRETS_MASTER_KEY?.trim() || await fs.readFile(path, "utf8"); + if (!isValidLocalEncryptedMasterKey(key)) throw new Error("native_harness_prestage_key_invalid"); + env.PAPERCLIP_SECRETS_MASTER_KEY_FILE = path; + env.PAPERCLIP_SECRETS_MASTER_KEY = key; +} + +export function nativeHarnessPrestageDatabaseUrl(source: string) { + const url = new URL(source); + url.searchParams.set("lock_timeout", "5000"); + url.searchParams.set("statement_timeout", "30000"); + url.searchParams.set("idle_in_transaction_session_timeout", "30000"); + return url.href; +} diff --git a/server/src/maintenance/native-harness-prestage.test.ts b/server/src/maintenance/native-harness-prestage.test.ts new file mode 100644 index 00000000000..f114f60b822 --- /dev/null +++ b/server/src/maintenance/native-harness-prestage.test.ts @@ -0,0 +1,113 @@ +import { afterEach, expect, it, vi } from "vitest"; +import { spawnSync } from "node:child_process"; +import { fileURLToPath } from "node:url"; +import { build } from "esbuild"; + +const mocks = vi.hoisted(() => ({ close: vi.fn(async () => {}), prestage: vi.fn(), key: vi.fn(async () => {}), obligations: vi.fn(async () => false) })); +vi.mock("@paperclipai/db", () => ({ createDb: () => ({}), closeRegisteredClients: mocks.close })); +vi.mock("../config.js", () => ({ loadConfig: () => ({ secretsMasterKeyFilePath: "/fixture/key" }) })); +vi.mock("../services/cloud-runtime-config.js", () => ({ cloudRuntimeConfigService: () => ({}) })); +vi.mock("../services/cloud-runtime-config-snapshot.js", () => ({ setCloudRuntimeConfigSnapshot: () => {} })); +vi.mock("../services/cloud-instance.js", () => ({ isCloudManagedInstance: () => false })); +vi.mock("../storage/index.js", () => ({ getStorageService: () => ({ provider: "s3" }) })); +vi.mock("../services/native-runtime/native-harness-backup-migration.js", () => ({ + prestageNativeHarnessBackups: mocks.prestage, hasLocalNativeHarnessObligations: mocks.obligations, +})); +vi.mock("./native-harness-prestage-runtime.js", async importOriginal => ({ + ...await importOriginal(), initializeNativeHarnessPrestageKey: mocks.key, +})); + +const initialExitCode = process.exitCode; +afterEach(() => { process.exitCode = initialExitCode; vi.useRealTimers(); vi.unstubAllEnvs(); vi.restoreAllMocks(); vi.resetModules(); mocks.close.mockClear(); mocks.obligations.mockReset().mockResolvedValue(false); }); + +it.each([true, false])("checks unfulfilled database obligations for an empty source scan (pending: %s)", async pending => { + vi.stubEnv("PAPERCLIP_NATIVE_HARNESS_PRESTAGE", "true"); + vi.stubEnv("DATABASE_URL", "postgres://fixture:fixture@localhost/fixture"); + const counts = { inspected: 0, published: 0, current: 0, deferred: {} }; + mocks.prestage.mockResolvedValue(counts); + mocks.obligations.mockResolvedValue(pending); + const success = vi.spyOn(console, "log").mockImplementation(() => {}); + const failure = vi.spyOn(console, "error").mockImplementation(() => {}); + await import("./native-harness-prestage.js"); + expect(mocks.close).toHaveBeenCalled(); + if (pending) { + expect(process.exitCode).toBe(1); + expect(success).not.toHaveBeenCalled(); + expect(failure).toHaveBeenCalledOnce(); + expect(JSON.parse(failure.mock.calls[0][0])).toEqual({ schema: "paperclip.native-harness-prestage.v1", + state: "blocked", failure: "native_harness_prestage_source_missing", enrollmentReady: false }); + } else { + expect(process.exitCode ?? 0).toBe(0); + expect(failure).not.toHaveBeenCalled(); + expect(success).toHaveBeenCalledWith(JSON.stringify({ schema: "paperclip.native-harness-prestage.v1", counts, enrollmentReady: false })); + } +}); + +it("keeps a nonempty deferred scan advisory without requiring all obligations to be copied", async () => { + vi.stubEnv("PAPERCLIP_NATIVE_HARNESS_PRESTAGE", "true"); + vi.stubEnv("DATABASE_URL", "postgres://fixture:fixture@localhost/fixture"); + const counts = { inspected: 1, published: 0, current: 0, deferred: { source_active: 1 } }; + mocks.prestage.mockResolvedValue(counts); mocks.obligations.mockResolvedValue(true); + const success = vi.spyOn(console, "log").mockImplementation(() => {}); + await import("./native-harness-prestage.js"); + expect(mocks.obligations).not.toHaveBeenCalled(); + expect(process.exitCode ?? 0).toBe(0); + expect(success).toHaveBeenCalledWith(JSON.stringify({ schema: "paperclip.native-harness-prestage.v1", counts, enrollmentReady: false })); +}); + +it("reports a failed deadline with real timers when the dependency holds no event-loop handle", async () => { + const bundled = await build({ + entryPoints: [fileURLToPath(new URL("./native-harness-prestage.ts", import.meta.url))], + bundle: true, platform: "node", format: "esm", write: false, + banner: { js: "const realTimeout = globalThis.setTimeout; globalThis.setTimeout = (fn, ms, ...args) => realTimeout(fn, ms === 5400000 ? 10 : ms === 10000 ? 50 : ms, ...args);" }, + plugins: [{ name: "prestage-dependencies", setup(builder) { + builder.onResolve({ filter: /.*/ }, args => args.importer ? { path: "dependencies", namespace: "fixture" } : undefined); + builder.onLoad({ filter: /.*/, namespace: "fixture" }, () => ({ contents: ` + export const createDb = () => ({}), closeRegisteredClients = async () => {}; + export const loadConfig = () => ({ secretsMasterKeyFilePath: '/fixture/key' }); + export const cloudRuntimeConfigService = () => ({}), setCloudRuntimeConfigSnapshot = () => {}; + export const isCloudManagedInstance = () => false, getStorageService = () => ({ provider: 's3' }); + export const initializeNativeHarnessPrestageKey = async () => {}, nativeHarnessPrestageDatabaseUrl = url => url; + export const prestageNativeHarnessBackups = () => new Promise(() => {}); + export const hasLocalNativeHarnessObligations = async () => false; + ` })); + } }], + }); + const child = spawnSync(process.execPath, ["--input-type=module"], { + input: bundled.outputFiles[0].text, encoding: "utf8", timeout: 5000, + env: { ...process.env, PAPERCLIP_NATIVE_HARNESS_PRESTAGE: "true", DATABASE_URL: "postgres://fixture:fixture@localhost/fixture" }, + }); + expect(child.error).toBeUndefined(); + expect(child.status).toBe(1); + expect(child.stdout).toBe(""); + expect(JSON.parse(child.stderr.trim())).toEqual({ schema: "paperclip.native-harness-prestage.v1", + state: "blocked", failure: "native_harness_prestage_deadline", enrollmentReady: false }); +}); + +it.each(["SIGTERM", "deadline"])("bounds shutdown if a dependency does not settle after %s", async mode => { + vi.useFakeTimers(); + vi.stubEnv("PAPERCLIP_NATIVE_HARNESS_PRESTAGE", "true"); + vi.stubEnv("DATABASE_URL", "postgres://fixture:fixture@localhost/fixture"); + const once = vi.spyOn(process, "once"), exit = vi.spyOn(process, "exit").mockImplementation(() => undefined as never); + vi.spyOn(console, "error").mockImplementation(() => {}); + const success = vi.spyOn(console, "log").mockImplementation(() => {}); + let settle!: () => void, started!: () => void; + const ready = new Promise(resolve => { started = resolve; }); + mocks.prestage.mockImplementation(async () => { started(); await new Promise(resolve => { settle = resolve; }); return { copied: 1 }; }); + const command = import("./native-harness-prestage.js"); + await ready; + try { + if (mode === "SIGTERM") { + const handler = once.mock.calls.find(([signal]) => signal === "SIGTERM")?.[1]; + expect(handler).toBeTypeOf("function"); handler!(); + } else await vi.advanceTimersByTimeAsync(90 * 60_000); + expect(mocks.key).toHaveBeenCalledWith("/fixture/key"); + expect(mocks.prestage.mock.calls.at(-1)?.[1].signal.aborted).toBe(true); + expect(mocks.close).toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(10_000); + expect(exit).toHaveBeenCalledWith(1); + } finally { settle(); await command; } + expect(success).not.toHaveBeenCalled(); + expect(process.exitCode).toBe(1); + expect(vi.getTimerCount()).toBe(0); +}); diff --git a/server/src/maintenance/native-harness-prestage.ts b/server/src/maintenance/native-harness-prestage.ts new file mode 100644 index 00000000000..fba4095a408 --- /dev/null +++ b/server/src/maintenance/native-harness-prestage.ts @@ -0,0 +1,59 @@ +import { closeRegisteredClients, createDb } from "@paperclipai/db"; +import { loadConfig } from "../config.js"; +import { cloudRuntimeConfigService } from "../services/cloud-runtime-config.js"; +import { setCloudRuntimeConfigSnapshot } from "../services/cloud-runtime-config-snapshot.js"; +import { isCloudManagedInstance } from "../services/cloud-instance.js"; +import { getStorageService } from "../storage/index.js"; +import { hasLocalNativeHarnessObligations, prestageNativeHarnessBackups } from "../services/native-runtime/native-harness-backup-migration.js"; +import { initializeNativeHarnessPrestageKey, nativeHarnessPrestageDatabaseUrl } from "./native-harness-prestage-runtime.js"; + +if (process.env.PAPERCLIP_NATIVE_HARNESS_PRESTAGE !== "true") throw new Error("Explicit harness pre-copy mode is required"); +const sourceUrl = process.env.DATABASE_URL; +if (!sourceUrl) throw new Error("Harness pre-copy requires an explicit database"); +const url = nativeHarnessPrestageDatabaseUrl(sourceUrl); +const cancellation = new AbortController(); +let shutdown: ReturnType | undefined; +let blocked = false; +const reportBlocked = (error: unknown) => { + process.exitCode = 1; + if (blocked) return; + blocked = true; + const message = error instanceof Error ? error.message : "native_harness_prestage_failed"; + const failure = /^[a-z_]{1,80}$/.test(message) ? message : "native_harness_prestage_failed"; + console.error(JSON.stringify({ schema: "paperclip.native-harness-prestage.v1", state: "blocked", failure, enrollmentReady: false })); +}; +const abort = (reason: string) => { + if (cancellation.signal.aborted) return; + reportBlocked(new Error(reason)); + cancellation.abort(new Error(reason)); + shutdown = setTimeout(() => process.exit(1), 10_000); + void closeRegisteredClients(url, { scope: "database" }); +}; +const stop = () => abort("native_harness_prestage_stopped"); +const deadline = setTimeout(() => abort("native_harness_prestage_deadline"), 90 * 60_000); +process.once("SIGINT", stop); process.once("SIGTERM", stop); +try { + const db = createDb(url, { maxConnections: 2, connectTimeoutSeconds: 10 }); + if (isCloudManagedInstance()) { + const config = await cloudRuntimeConfigService(db).inspectStored(); + if (!config) throw new Error("cloud_runtime_configuration_missing"); + setCloudRuntimeConfigSnapshot(config); + } + await initializeNativeHarnessPrestageKey(loadConfig().secretsMasterKeyFilePath); + const storage = getStorageService(); + if (storage.provider !== "s3") throw new Error("shared_object_storage_required"); + const counts = await prestageNativeHarnessBackups(db, { storage, signal: cancellation.signal }); + cancellation.signal.throwIfAborted(); + if (counts.inspected === 0) { + const obligations = await hasLocalNativeHarnessObligations(db); + cancellation.signal.throwIfAborted(); + if (obligations) throw new Error("native_harness_prestage_source_missing"); + } + console.log(JSON.stringify({ schema: "paperclip.native-harness-prestage.v1", counts, enrollmentReady: false })); +} catch (error) { + reportBlocked(error); +} finally { + clearTimeout(deadline); process.off("SIGINT", stop); process.off("SIGTERM", stop); + await closeRegisteredClients(url, { scope: "database" }); + clearTimeout(shutdown); +} 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 388a516eeea..b36d7ab07b4 100644 --- a/server/src/services/native-runtime/native-harness-backup-archive.ts +++ b/server/src/services/native-runtime/native-harness-backup-archive.ts @@ -4,6 +4,7 @@ import { join, posix } from "node:path"; import { Transform } from "node:stream"; import { pipeline } from "node:stream/promises"; import tar from "tar-stream"; +import { readNativeHarnessBackupSourceInput } from "./native-harness-backup-source.js"; export const MAX_NATIVE_HARNESS_BYTES = 4 * 1024 * 1024 * 1024; export const MAX_NATIVE_HARNESS_OBJECT_BYTES = MAX_NATIVE_HARNESS_BYTES + 768 * 1024 * 1024; @@ -45,7 +46,7 @@ const boundedArchive = (limit = MAX_NATIVE_HARNESS_OBJECT_BYTES) => { export async function captureNativeHarnessArchive(root: string, target: string, signal?: AbortSignal, maxArchiveBytes = MAX_NATIVE_HARNESS_OBJECT_BYTES) { if (!Number.isSafeInteger(maxArchiveBytes) || maxArchiveBytes <= 0 || maxArchiveBytes > MAX_NATIVE_HARNESS_OBJECT_BYTES) throw fail(); - if (!(await lstat(root)).isDirectory()) throw fail(); + if (!(await readNativeHarnessBackupSourceInput(() => lstat(root))).isDirectory()) throw fail(); const archive = tar.pack(); let outputFailure: unknown; const output = pipeline(archive, boundedArchive(maxArchiveBytes), createWriteStream(target, { flags: "wx", mode: 0o600 }), { signal }); @@ -53,11 +54,11 @@ export async function captureNativeHarnessArchive(root: string, target: string, let count = 0, bytes = 0; const links = new Map(); const visit = async (relative: string): Promise => { - for (const name of (await readdir(join(root, relative))).sort()) { + for (const name of (await readNativeHarnessBackupSourceInput(() => readdir(join(root, relative)))).sort()) { signal?.throwIfAborted(); const entry = relative ? `${relative}/${name}` : name; if (!safeName(entry) || ++count > MAX_NATIVE_HARNESS_ENTRIES) throw fail(); - const file = join(root, entry), stat = await lstat(file); + const file = join(root, entry), stat = await readNativeHarnessBackupSourceInput(() => lstat(file)); if (stat.isDirectory()) { await new Promise((resolve, reject) => archive.entry({ name: entry, type: "directory", mode: stat.mode & 0o777 }, error => error ? reject(error) : resolve())); @@ -65,17 +66,17 @@ export async function captureNativeHarnessArchive(root: string, target: string, } else if (stat.isFile()) { bytes += stat.size; if (bytes > MAX_NATIVE_HARNESS_BYTES) throw fail(); - const handle = await open(file, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK); + const handle = await readNativeHarnessBackupSourceInput(() => open(file, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK)); try { - const before = await handle.stat(); + const before = await readNativeHarnessBackupSourceInput(() => handle.stat()); if (!before.isFile() || before.ino !== stat.ino || before.dev !== stat.dev || before.size !== stat.size) throw fail(); const stream = archive.entry({ name: entry, size: stat.size, type: "file", mode: stat.mode & 0o777 }); await pipeline(handle.createReadStream({ autoClose: false }), stream, { signal }); - const after = await handle.stat(); + const after = await readNativeHarnessBackupSourceInput(() => handle.stat()); if (before.size !== after.size || before.mtimeMs !== after.mtimeMs || before.ctimeMs !== after.ctimeMs) throw fail(); } finally { await handle.close(); } } else if (stat.isSymbolicLink()) { - const linkname = await readlink(file); + const linkname = await readNativeHarnessBackupSourceInput(() => readlink(file)); if (!safeLink(entry, linkname)) throw fail(); links.set(entry, linkname); await new Promise((resolve, reject) => archive.entry({ name: entry, type: "symlink", linkname }, 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 3cb5050b962..12d55d49757 100644 --- a/server/src/services/native-runtime/native-harness-backup-migration.ts +++ b/server/src/services/native-runtime/native-harness-backup-migration.ts @@ -1,13 +1,14 @@ import { createHash } from "node:crypto"; -import { lstat, readFile, readdir } from "node:fs/promises"; +import { readdir } from "node:fs/promises"; import { join } from "node:path"; -import { and, eq, sql } from "drizzle-orm"; -import { environmentLeases, heartbeatRuns, nativeHarnessBackups, type Db } from "@paperclipai/db"; +import { and, eq, inArray, or, sql } from "drizzle-orm"; +import { environmentLeases, heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations, type Db } from "@paperclipai/db"; 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 { nativeHarnessBackupOwner, 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"; +import { NativeHarnessBackupSourceError, readNativeHarnessBackupSource } from "./native-harness-backup-source.js"; export async function hasLocalNativeHarnessObligations(db: Db): Promise { const rows = await db.execute(sql`select 1 from agent_task_sessions s @@ -29,14 +30,9 @@ 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")); + const files = await readNativeHarnessBackupSource(root); + if (!files) return null; + const { manifestBytes, manifest, state } = files; 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(); @@ -61,16 +57,34 @@ async function prepareSource(base: string, scopeDigest: string) { type Source = NonNullable>>; -async function migrateSource(db: Db, input: { storage: StorageService; signal: AbortSignal }, source: Source) { - input.signal.throwIfAborted(); - const { root, scopeDigest, manifestSha256 } = source; +async function sourceRun(db: Db, source: Source) { const [run] = await db.select({ companyId: heartbeatRuns.companyId, issueId: heartbeatRuns.nativeIssueId, - sessionId: heartbeatRuns.nativeSessionId, runnerId: heartbeatRuns.runnerInstanceId }).from(heartbeatRuns) + sessionId: heartbeatRuns.nativeSessionId, runnerId: heartbeatRuns.runnerInstanceId, + status: heartbeatRuns.status, finishedAt: heartbeatRuns.finishedAt }).from(heartbeatRuns) .where(eq(heartbeatRuns.id, source.runId)); - if (!run) return 0; - if (!run.issueId || run.sessionId !== source.sessionId || run.runnerId !== source.runnerId) { + if (!run) return null; + if (!run.issueId || !run.sessionId || !run.runnerId || run.sessionId !== source.sessionId || run.runnerId !== source.runnerId) { throw new Error("native_harness_migration_binding_mismatch"); } + return { ...run, issueId: run.issueId, sessionId: run.sessionId, runnerId: run.runnerId }; +} + +async function verifiedSourceLease(db: Db, source: Source, run: NonNullable>>, signal: AbortSignal) { + 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' = ${source.manifestSha256}`)); + for (const lease of leases) { + const stamp = readNativeHarnessBackupStamp(lease.stamp, lease.providerLeaseId ?? ""); + if (stamp?.sessionScopeSha256 === source.scopeDigest && await verifyNativeHarnessBackupStampAsync(stamp, lease.providerLeaseId!, signal)) return lease.providerLeaseId!; + } + return null; +} + +async function migrateSource(db: Db, input: { storage: StorageService; signal: AbortSignal }, source: Source) { + input.signal.throwIfAborted(); + const { root, scopeDigest, manifestSha256 } = source, run = await sourceRun(db, source); + if (!run) return 0; const [current] = await db.select().from(nativeHarnessBackups).where(and(eq(nativeHarnessBackups.companyId, run.companyId), eq(nativeHarnessBackups.scopeDigest, scopeDigest), eq(nativeHarnessBackups.status, "current"))); if (current) { @@ -79,57 +93,128 @@ async function migrateSource(db: Db, input: { storage: StorageService; signal: A 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"); + if (!await verifiedSourceLease(db, source, run, input.signal)) 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; } +async function mapSources(input: { signal: AbortSignal; concurrency?: number; + unavailableSource?: (error: unknown, scopeDigest: string) => T | undefined }, visit: (source: Source, signal: AbortSignal) => Promise) { + const base = nativeHarnessStateBase(); + const directories = await readdir(base).catch((error: NodeJS.ErrnoException) => { + if (error.code === "ENOENT") return [] as string[]; throw error; + }); + let reserved = 0, wave: Source[] = []; + const results: T[] = []; + 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 visit(source, signal); } + 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; + results.push(result.value); + } + wave = []; reserved = 0; + }; + for (const scopeDigest of directories.filter(name => /^[a-f0-9]{64}$/.test(name)).sort()) { + input.signal.throwIfAborted(); + let source; + try { source = await prepareSource(base, scopeDigest); } + catch (error) { + input.signal.throwIfAborted(); + const deferred = input.unavailableSource?.(error, scopeDigest); + if (deferred === undefined) throw error; + results.push(deferred); continue; + } + if (!source) continue; + const reservation = 2 * source.archiveLimit; + if (wave.length && (wave.length === (input.concurrency ?? MAX_PARALLEL_BACKUPS) || reserved + reservation > MAX_SCRATCH_BYTES)) await flush(); + wave.push(source); reserved += reservation; + } + if (wave.length) await flush(); + return results; +} + export async function migrateNativeHarnessBackups(db: Db, input: { storage: StorageService; writersStopped: true; signal: AbortSignal; }): Promise { if (!input.writersStopped) throw new Error("native_harness_migration_requires_offline_writers"); const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(offlineMigrationExecutionWhere()).limit(1); if (active) throw new Error("native_harness_migration_requires_offline_writers"); - const base = nativeHarnessStateBase(); - const directories = await readdir(base).catch((error: NodeJS.ErrnoException) => { - if (error.code === "ENOENT") return [] as string[]; throw error; + const counts = await mapSources(input, (source, signal) => migrateSource(db, { storage: input.storage, signal }, source)); + return counts.reduce((total, count) => total + count, 0); +} + +export async function prestageNativeHarnessBackups(db: Db, input: { storage: StorageService; signal: AbortSignal }) { + const deferSource = (scopeDigest: string, reason: string, outcome: string) => { + console.warn(JSON.stringify({ schema: "paperclip.native-harness-prestage-source.v1", scopeDigest, reason })); + return outcome; + }; + const outcomes = await mapSources({ ...input, concurrency: 1, unavailableSource: (error, scopeDigest) => { + if (error instanceof NativeHarnessBackupSourceError || error instanceof Error && error.message === "native_harness_migration_source_unsettled") { + return deferSource(scopeDigest, error.message, "source_unavailable"); + } + if (["ENOENT", "ENOTDIR", "EISDIR", "ELOOP", "EACCES", "EPERM"].includes((error as NodeJS.ErrnoException)?.code ?? "")) { + return deferSource(scopeDigest, "native_harness_migration_source_unavailable", "source_unavailable"); + } + return undefined; + } }, async (source, signal) => { + signal.throwIfAborted(); + const run = await sourceRun(db, source).catch(error => { + if (error instanceof Error && error.message === "native_harness_migration_binding_mismatch") return null; + throw error; + }); + if (!run) return "missing_run"; + const [current] = await db.select().from(nativeHarnessBackups).where(and(eq(nativeHarnessBackups.companyId, run.companyId), + eq(nativeHarnessBackups.scopeDigest, source.scopeDigest), eq(nativeHarnessBackups.status, "current"))); + if (current) { + if (current.manifestSha256 !== source.manifestSha256) return "durable_source_changed"; + await verifyDurableNativeHarnessBackup({ db, storage: input.storage, companyId: run.companyId, + scopeDigest: source.scopeDigest, manifestSha256: source.manifestSha256, signal }); + return "current"; + } + if (run.status !== "succeeded" || !run.finishedAt) return "source_not_successful"; + const [coordinator] = await db.select().from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.companyId, run.companyId), eq(nativeRunFinalizations.issueId, run.issueId), + eq(nativeRunFinalizations.runId, source.runId))); + if (coordinator?.phase !== "committed" || coordinator.leaseOwner || !coordinator.resultId || !coordinator.controllerBootId) return "owner_not_released"; + const [task] = await db.select({ executionRunId: issues.executionRunId, checkoutRunId: issues.checkoutRunId }).from(issues) + .where(and(eq(issues.companyId, run.companyId), eq(issues.id, run.issueId))); + const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, run.companyId), inArray(heartbeatRuns.status, ["queued", "running"]), + or(eq(heartbeatRuns.nativeIssueId, run.issueId), eq(heartbeatRuns.issueId, run.issueId), eq(heartbeatRuns.nativeSessionId, run.sessionId)))).limit(1); + if (!task || task.executionRunId || task.checkoutRunId || active) return "source_active"; + const providerLeaseId = await verifiedSourceLease(db, source, run, signal); + if (!providerLeaseId) return "source_unverified"; + try { + const owner = await nativeHarnessBackupOwner(db, { companyId: run.companyId, issueId: run.issueId, + runId: source.runId, bootId: coordinator.controllerBootId }); + await persistNativeHarnessBackup({ db, storage: input.storage, root: source.root, scopeDigest: source.scopeDigest, + owner, signal, maxArchiveBytes: source.archiveLimit, + inactiveSource: { normalizedSessionId: run.sessionId, runnerInstanceId: run.runnerId, providerLeaseId } }); + return "published"; + } catch (error) { + signal.throwIfAborted(); + if (error instanceof NativeHarnessBackupSourceError || error instanceof Error && ["native_runner_checkpoint_owner_lost", "native_harness_storage_owner_missing", + "native_harness_storage_prestage_source_active", "native_harness_storage_prestage_source_changed", + "native_harness_storage_owner_superseded", "native_harness_archive_invalid"].includes(error.message)) { + return deferSource(source.scopeDigest, error.message, "source_changed"); + } + throw error; + } }); - 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 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; + const counts = { inspected: outcomes.length, published: 0, current: 0, deferred: {} as Record }; + for (const outcome of outcomes) { + if (outcome === "published" || outcome === "current") counts[outcome]++; + else counts.deferred[outcome] = (counts.deferred[outcome] ?? 0) + 1; } - if (wave.length) await flush(); - return count; + return counts; } diff --git a/server/src/services/native-runtime/native-harness-backup-source.ts b/server/src/services/native-runtime/native-harness-backup-source.ts new file mode 100644 index 00000000000..5a8500bf921 --- /dev/null +++ b/server/src/services/native-runtime/native-harness-backup-source.ts @@ -0,0 +1,35 @@ +import { lstat, readFile } from "node:fs/promises"; +import { join } from "node:path"; + +export class NativeHarnessBackupSourceError extends Error {} + +export async function readNativeHarnessBackupSourceInput(read: () => Promise): Promise { + try { return await read(); } + catch (error) { + if (["ENOENT", "ENOTDIR", "EISDIR", "ELOOP", "EACCES", "EPERM"].includes((error as NodeJS.ErrnoException)?.code ?? "")) { + throw new NativeHarnessBackupSourceError("native_harness_migration_source_unavailable", { cause: error }); + } + throw error; + } +} + +export async function readNativeHarnessBackupSource(root: string) { + try { + const stat = await readNativeHarnessBackupSourceInput(() => lstat(root).catch((error: NodeJS.ErrnoException) => { + if (error.code === "ENOENT") return null; throw error; + })); + if (!stat) return null; + if (!stat.isDirectory() || stat.isSymbolicLink()) throw new NativeHarnessBackupSourceError("native_harness_migration_unsafe_source"); + const manifestBytes = await readNativeHarnessBackupSourceInput(() => readFile(join(root, "manifest.json"))); + const manifest: unknown = JSON.parse(manifestBytes.toString("utf8")); + const state: unknown = JSON.parse(await readNativeHarnessBackupSourceInput(() => readFile(join(root, "runner", "runner-state.json"), "utf8"))); + if (!manifest || typeof manifest !== "object" || Array.isArray(manifest) + || !state || typeof state !== "object" || Array.isArray(state)) { + throw new NativeHarnessBackupSourceError("native_harness_migration_source_unsettled"); + } + return { manifestBytes, manifest: manifest as Record, state: state as Record }; + } catch (error) { + if (error instanceof SyntaxError) throw new NativeHarnessBackupSourceError("native_harness_migration_source_invalid_json", { cause: error }); + throw error; + } +} 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 c66b503e2dd..ab53c6e1b5a 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 @@ -12,9 +12,10 @@ import { nativeHarnessBackupOwner, persistNativeHarnessBackup, restoreDurableNat import { removeNativeHarnessBackup } from "./native-harness-backup-cleanup.js"; import { buildNativeHarnessBackupManifest } from "./native-session-executor.js"; import { createNativeHarnessBackupStamp, verifyNativeHarnessBackupStamp, verifyNativeHarnessBackupStampAsync } from "./native-harness-backup-stamp.js"; -import { hasLocalNativeHarnessObligations, migrateNativeHarnessBackups } from "./native-harness-backup-migration.js"; +import { hasLocalNativeHarnessObligations, migrateNativeHarnessBackups, prestageNativeHarnessBackups } from "./native-harness-backup-migration.js"; import { nativeHarnessBackupExpirationService } from "./native-harness-backup-expiration.js"; import { legacyControllerBootId } from "../legacy-controller-lease.js"; +import * as backupArchive from "./native-harness-backup-archive.js"; describeEmbeddedPostgres("durable native provider backups", () => { const environment = nativeRestartFixture(); @@ -67,7 +68,7 @@ describeEmbeddedPostgres("durable native provider backups", () => { 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 }; + return { ...f, home, root, runnerInstanceId, providerLeaseId }; } async function offlineFixtures(count: number) { const base = join(environment.current.runtimeRoot, randomUUID()); ownedHomes.push(base); await mkdir(base); @@ -179,6 +180,164 @@ describeEmbeddedPostgres("durable native provider backups", () => { leaseExpiresAt: null }).where(eq(nativeRunFinalizations.runId, f.runId)); return result; } + async function inactiveFixture() { + const [f] = await offlineFixtures(1); + await commitWarmRun(f); + await f.db.update(issues).set({ executionRunId: null, checkoutRunId: null }).where(eq(issues.id, f.issueId)); + const owner = await nativeHarnessBackupOwner(f.db, { companyId: f.companyId, issueId: f.issueId, + runId: f.runId, bootId: environment.current.successor.bootId }); + return { ...f, owner, inactiveSource: { normalizedSessionId: f.identity.normalizedSessionId, + runnerInstanceId: f.runnerInstanceId, providerLeaseId: f.providerLeaseId } }; + } + it("pre-copies released history and restores its bytes while an unrelated execution stays active", async () => { + const f = await inactiveFixture(), active = await environment.seedRun(randomUUID()); + const input = { storage: f.storage, signal: AbortSignal.timeout(15_000) }; + expect(await prestageNativeHarnessBackups(f.db, input)).toEqual({ inspected: 1, published: 1, current: 0, deferred: {} }); + expect(await prestageNativeHarnessBackups(f.db, input)).toEqual({ inspected: 1, published: 0, current: 1, deferred: {} }); + await expect(migrateNativeHarnessBackups(f.db, { ...input, writersStopped: true })).rejects.toThrow("requires_offline_writers"); + expect((await active.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, active.runId)))[0].status).toBe("running"); + await removeNativeHarnessBackup(f.home); + const replacement = join(environment.current.runtimeRoot, randomUUID()); ownedHomes.push(replacement); + expect(await restoreDurableNativeHarnessBackup({ ...f, backupRoot: replacement })).toBe(true); + expect(JSON.parse(await readFile(join(replacement, "current", "codex-home", "sessions", "history.json"), "utf8"))).toEqual(f.identity); + }); + it.each(["queued", "running"] as const)("defers pre-copy when the same task has a %s execution", async status => { + const f = await inactiveFixture(), put = vi.spyOn(f.storage, "putFile"); + await f.db.insert(heartbeatRuns).values({ id: randomUUID(), companyId: f.companyId, agentId: environment.current.agentId, + issueId: f.issueId, nativeIssueId: f.issueId, nativeSessionId: f.identity.normalizedSessionId, status, runtimeMode: "native" }); + expect(await prestageNativeHarnessBackups(f.db, { storage: f.storage, signal: AbortSignal.timeout(15_000) })) + .toEqual({ inspected: 1, published: 0, current: 0, deferred: { source_active: 1 } }); + await expect(persistNativeHarnessBackup(f)).rejects.toThrow("native_harness_storage_prestage_source_active"); + expect(put).not.toHaveBeenCalled(); + expect(await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest))).toEqual([]); + }); + it("defers failed history without treating pre-copy as offline authority", async () => { + const [f] = await offlineFixtures(1); + expect(await prestageNativeHarnessBackups(f.db, { storage: f.storage, signal: AbortSignal.timeout(15_000) })) + .toEqual({ inspected: 1, published: 0, current: 0, deferred: { source_not_successful: 1 } }); + expect(await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest))).toEqual([]); + }); + it("defers a source being replaced online while offline migration still refuses it", async () => { + const f = await inactiveFixture(); + await writeFile(join(f.root, "runner", "runner-state.json"), JSON.stringify({ runId: f.runId, lifecycle: "running" })); + expect(await prestageNativeHarnessBackups(f.db, { storage: f.storage, signal: AbortSignal.timeout(15_000) })) + .toEqual({ inspected: 1, published: 0, current: 0, deferred: { source_unavailable: 1 } }); + await expect(migrateNativeHarnessBackups(f.db, { storage: f.storage, writersStopped: true, signal: AbortSignal.timeout(15_000) })) + .rejects.toThrow("native_harness_migration_source_unsettled"); + }); + async function releasedFixtures() { + const fixtures = await offlineFixtures(2); + for (const f of fixtures) { + await commitWarmRun(f); + await f.db.update(issues).set({ executionRunId: null, checkoutRunId: null }).where(eq(issues.id, f.issueId)); + } + return fixtures; + } + async function damageSource(f: Awaited>, kind: string) { + if (kind === "unsafe") { + await rename(f.root, f.root + "-retained"); + await symlink(f.root + "-retained", f.root); + } else await writeFile(join(f.root, kind === "manifest" ? "manifest.json" : "runner/runner-state.json"), "{"); + } + it.each(["source", "output"])("distinguishes %s ENOENT during archive capture after source preparation", async kind => { + const [bad, good] = await releasedFixtures(), capture = backupArchive.captureNativeHarnessArchive; + const warning = vi.spyOn(console, "warn").mockImplementation(() => {}); + const archive = vi.spyOn(backupArchive, "captureNativeHarnessArchive").mockImplementation(async (root, target, signal, limit) => { + if (root === bad.root) { + if (kind === "source") await rename(root, root + "-retained"); + else target = join(target, "missing-parent", "archive"); + } + return capture(root, target, signal, limit); + }); + try { + const preparation = prestageNativeHarnessBackups(bad.db, { storage: bad.storage, signal: AbortSignal.timeout(15_000) }); + if (kind === "source") { + expect(await preparation).toEqual({ inspected: 2, published: 1, current: 0, deferred: { source_changed: 1 } }); + expect(JSON.parse(warning.mock.calls[0][0])).toMatchObject({ scopeDigest: bad.scopeDigest, + reason: "native_harness_migration_source_unavailable" }); + expect((await good.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, good.scopeDigest)))[0].status).toBe("current"); + } else { + await expect(preparation).rejects.toMatchObject({ code: "ENOENT" }); + expect(warning).not.toHaveBeenCalled(); + expect(await good.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, good.scopeDigest))).toHaveLength(0); + } + expect(await bad.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, bad.scopeDigest))).toHaveLength(0); + } finally { archive.mockRestore(); warning.mockRestore(); } + }); + it.each(["manifest", "runner", "unsafe"])("defers a damaged %s source and still copies the next eligible backup", async kind => { + const [bad, good] = await releasedFixtures(), warning = vi.spyOn(console, "warn").mockImplementation(() => {}); + try { + await damageSource(bad, kind); + expect(await prestageNativeHarnessBackups(bad.db, { storage: bad.storage, signal: AbortSignal.timeout(15_000) })) + .toEqual({ inspected: 2, published: 1, current: 0, deferred: { source_unavailable: 1 } }); + expect(warning).toHaveBeenCalledWith(expect.stringContaining(bad.scopeDigest)); + expect(JSON.parse(warning.mock.calls[0][0])).toMatchObject({ scopeDigest: bad.scopeDigest, + reason: kind === "unsafe" ? "native_harness_migration_unsafe_source" : "native_harness_migration_source_invalid_json" }); + expect(await bad.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, bad.scopeDigest))).toHaveLength(0); + expect((await good.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, good.scopeDigest)))[0].status).toBe("current"); + await expect(migrateNativeHarnessBackups(bad.db, { storage: bad.storage, writersStopped: true, signal: AbortSignal.timeout(15_000) })) + .rejects.toThrow(kind === "unsafe" ? "native_harness_migration_unsafe_source" : /JSON|json/); + } finally { warning.mockRestore(); } + }); + it.each(["manifest", "runner", "unsafe"])("defers %s damage at the publication fence and copies the next backup", async kind => { + const [bad, good] = await releasedFixtures(), put = bad.storage.putFile.bind(bad.storage); + const warning = vi.spyOn(console, "warn").mockImplementation(() => {}); + const upload = vi.spyOn(bad.storage, "putFile").mockImplementation(async input => { + const result = await put(input); + if (input.objectKey?.includes(bad.scopeDigest)) await damageSource(bad, kind); + return result; + }); + try { + expect(await prestageNativeHarnessBackups(bad.db, { storage: bad.storage, signal: AbortSignal.timeout(15_000) })) + .toEqual({ inspected: 2, published: 1, current: 0, deferred: { source_changed: 1 } }); + expect(JSON.parse(warning.mock.calls[0][0])).toMatchObject({ scopeDigest: bad.scopeDigest, + reason: kind === "unsafe" ? "native_harness_migration_unsafe_source" : "native_harness_migration_source_invalid_json" }); + expect((await bad.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, bad.scopeDigest))).some(row => row.status === "current")).toBe(false); + expect((await good.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, good.scopeDigest)))[0].status).toBe("current"); + } finally { upload.mockRestore(); warning.mockRestore(); } + }); + it("keeps a database SyntaxError fatal instead of deferring it as source damage", async () => { + const fixtures = await releasedFixtures(), f = fixtures[0], failure = new SyntaxError("database result decoding failed"); + const transaction = vi.spyOn(f.db, "transaction").mockRejectedValueOnce(failure); + try { + await expect(prestageNativeHarnessBackups(f.db, { storage: f.storage, signal: AbortSignal.timeout(15_000) })).rejects.toBe(failure); + for (const source of fixtures) expect(await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, source.scopeDigest))).toHaveLength(0); + } finally { transaction.mockRestore(); } + }); + it("keeps cancellation fatal when a source also changes during upload", async () => { + const [f] = await releasedFixtures(), cancellation = new AbortController(), failure = new Error("operator stopped"); + const put = f.storage.putFile.bind(f.storage); + const upload = vi.spyOn(f.storage, "putFile").mockImplementation(async input => { + const result = await put(input); await damageSource(f, "manifest"); cancellation.abort(failure); return result; + }); + try { + await expect(prestageNativeHarnessBackups(f.db, { storage: f.storage, signal: cancellation.signal })).rejects.toBe(failure); + expect((await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest))).some(row => row.status === "current")).toBe(false); + } finally { upload.mockRestore(); } + }); + it.each(["source_advanced", "queued_turn", "lease_revoked", "owner_changed", "run_failed"])("keeps the published source when %s happens during pre-copy upload", async change => { + const f = await inactiveFixture(); await persistNativeHarnessBackup(f); + const [before] = await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest)); + const put = f.storage.putFile.bind(f.storage); + vi.spyOn(f.storage, "putFile").mockImplementation(async input => { + const result = await put(input); + if (change === "source_advanced") await writeFile(join(f.root, "manifest.json"), (await readFile(join(f.root, "manifest.json"), "utf8")) + "\n"); + if (change === "queued_turn") await f.db.insert(heartbeatRuns).values({ id: randomUUID(), companyId: f.companyId, + agentId: environment.current.agentId, issueId: f.issueId, status: "queued" }); + if (change === "lease_revoked") await f.db.update(environmentLeases).set({ metadata: {} }) + .where(eq(environmentLeases.providerLeaseId, f.providerLeaseId)); + if (change === "owner_changed") await f.db.update(nativeRunFinalizations).set({ controllerGeneration: 2 }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + if (change === "run_failed") await f.db.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, f.runId)); + return result; + }); + await expect(persistNativeHarnessBackup(f)).rejects.toThrow(change === "owner_changed" ? "native_runner_checkpoint_owner_lost" + : change === "queued_turn" ? "native_harness_storage_prestage_source_active" : "native_harness_storage_prestage_source_changed"); + const rows = await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest)); + expect(rows.filter(row => row.status === "current").map(row => row.id)).toEqual([before.id]); + expect(rows.filter(row => row.status === "pending")).toHaveLength(1); + expect(await readFile(join(f.root, "manifest.json"), "utf8")).toContain("paperclip.native-harness-backup.v1"); + }); it.each(["idle", "preparing", "native"])("preserves a released committed warm session with a %s successor", async state => { const f = await fixture(); const result = await commitWarmRun(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 8a9880f8785..ae7ee97d9af 100644 --- a/server/src/services/native-runtime/native-harness-backup-storage.ts +++ b/server/src/services/native-runtime/native-harness-backup-storage.ts @@ -6,7 +6,7 @@ import { join } from "node:path"; import os from "node:os"; import { pipeline } from "node:stream/promises"; import { Transform } from "node:stream"; -import { and, eq, inArray, sql } from "drizzle-orm"; +import { and, eq, inArray, isNotNull, or, sql } from "drizzle-orm"; import { environmentLeases, heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations, type Db } from "@paperclipai/db"; import { createLocalEncryptedCipher } from "../../secrets/local-encrypted-provider.js"; import type { StorageService } from "../../storage/types.js"; @@ -18,6 +18,7 @@ import { captureNativeHarnessArchive, restoreNativeHarnessArchive, MAX_NATIVE_HA import { removeNativeHarnessBackup } from "./native-harness-backup-cleanup.js"; import { hasRemoteTerminationReceipt, resolveRemoteTerminationLease } from "../remote-execution-termination.js"; import { offlineMigrationExecutionWhere } from "../offline-migration-execution.js"; +import { readNativeHarnessBackupSource } from "./native-harness-backup-source.js"; const SCHEMA = "paperclip.native-harness-object.v1"; type Binding = { companyId: string; scopeDigest: string; id: string }; @@ -28,7 +29,9 @@ type Reference = Binding & { type Row = typeof nativeHarnessBackups.$inferSelect; type OfflineOwner = { kind: "offline_migration"; companyId: string; issueId: string; runId: string; writersStopped: true; terminalFailureOnly?: true }; -const fail = (reason: string): never => { throw new Error(`native_harness_storage_${reason}`); }; +type InactiveSource = { root: string; manifestSha256: string; scopeDigest: string; + normalizedSessionId: string; runnerInstanceId: string; providerLeaseId: string }; +function fail(reason: string): never { throw new Error(`native_harness_storage_${reason}`); } const aad = (binding: Binding) => Buffer.from(JSON.stringify([SCHEMA, binding.companyId, binding.scopeDigest, binding.id])); const scope = (binding: Pick) => and( eq(nativeHarnessBackups.companyId, binding.companyId), eq(nativeHarnessBackups.scopeDigest, binding.scopeDigest)); @@ -89,9 +92,11 @@ export async function nativeHarnessBackupOwner(db: Db, input: { companyId: strin if (!row || !row.leaseOwner && (row.phase !== "committed" || !row.resultId)) fail("owner_missing"); return { ...input, authorityRunId: input.runId, leaseOwner: row!.leaseOwner, generation: row!.controllerGeneration } satisfies NativeRunnerCheckpointFenceOwner; } -async function fence(tx: Parameters[0]>[0], owner: NativeRunnerCheckpointFenceOwner | OfflineOwner) { +async function fence(tx: Parameters[0]>[0], owner: NativeRunnerCheckpointFenceOwner | OfflineOwner, + inactiveSource?: InactiveSource) { + if (inactiveSource && "kind" in owner) fail("prestage_requires_runtime_owner"); const checkpointOwner = "kind" in owner ? null : await lockNativeRunnerCheckpointOwner(tx, owner); - const [task] = await tx.select({ executionRunId: issues.executionRunId }).from(issues) + const [task] = await tx.select({ executionRunId: issues.executionRunId, checkoutRunId: issues.checkoutRunId }).from(issues) .where(and(eq(issues.companyId, owner.companyId), eq(issues.id, owner.issueId))).for("share"); const authorityRunId = "kind" in owner ? owner.runId : owner.authorityRunId; if (!task) fail("owner_superseded"); @@ -138,15 +143,48 @@ async function fence(tx: Parameters[0]>[0], owner: if (owner.leaseOwner === null && !committed) fail("owner_missing"); if (task.executionRunId && task.executionRunId !== authorityRunId && !committed) fail("owner_superseded"); await validateNativeRunnerCheckpointOwner(tx, owner, owned, false); + if (inactiveSource) { + if (!committed || task.executionRunId || task.checkoutRunId) fail("prestage_source_active"); + const [active] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, owner.companyId), inArray(heartbeatRuns.status, ["queued", "running"]), + or(eq(heartbeatRuns.nativeIssueId, owner.issueId), eq(heartbeatRuns.issueId, owner.issueId), + eq(heartbeatRuns.nativeSessionId, inactiveSource.normalizedSessionId)))).limit(1); + if (active) fail("prestage_source_active"); + const [binding] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, owner.authorityRunId), eq(heartbeatRuns.companyId, owner.companyId), + eq(heartbeatRuns.nativeIssueId, owner.issueId), eq(heartbeatRuns.nativeSessionId, inactiveSource.normalizedSessionId), + eq(heartbeatRuns.runnerInstanceId, inactiveSource.runnerInstanceId), eq(heartbeatRuns.status, "succeeded"), + isNotNull(heartbeatRuns.finishedAt))); + const [lease] = await tx.select({ stamp: sql`${environmentLeases.metadata}->'nativeHarnessBackup'` }) + .from(environmentLeases).where(and(eq(environmentLeases.companyId, owner.companyId), + eq(environmentLeases.issueId, owner.issueId), eq(environmentLeases.providerLeaseId, inactiveSource.providerLeaseId), + sql`${environmentLeases.metadata}->'nativeHarnessBackup'->>'manifestSha256' = ${inactiveSource.manifestSha256}`)).limit(1); + const stamp = readNativeHarnessBackupStamp(lease?.stamp, inactiveSource.providerLeaseId); + if (!binding || !stamp || stamp.sessionScopeSha256 !== inactiveSource.scopeDigest + || stamp.normalizedSessionId !== inactiveSource.normalizedSessionId || stamp.runnerInstanceId !== inactiveSource.runnerInstanceId) { + fail("prestage_source_changed"); + } + const source = await readNativeHarnessBackupSource(inactiveSource.root); + if (!source) fail("prestage_source_changed"); + const { manifestBytes: manifest, manifest: identity, state } = source; + if (`sha256:${createHash("sha256").update(manifest).digest("hex")}` !== inactiveSource.manifestSha256 + || identity.normalizedSessionId !== stamp.normalizedSessionId || identity.runnerInstanceId !== stamp.runnerInstanceId + || identity.sourceProviderLeaseId !== stamp.sourceProviderLeaseId || identity.completedAt !== stamp.completedAt + || state.runId !== owner.authorityRunId || state.lifecycle !== "suspended") fail("prestage_source_changed"); + } } export async function persistNativeHarnessBackup(input: { db: Db; storage: StorageService; owner: NativeRunnerCheckpointFenceOwner | OfflineOwner; scopeDigest: string; root: string; signal?: AbortSignal; maxArchiveBytes?: number; + inactiveSource?: { normalizedSessionId: string; runnerInstanceId: string; providerLeaseId: string }; }): Promise { if (!/^[a-f0-9]{64}$/.test(input.scopeDigest)) fail("scope_invalid"); - const manifestBytes = await readFile(join(input.root, "manifest.json")); + const source = input.inactiveSource ? await readNativeHarnessBackupSource(input.root) : undefined; + if (source === null) fail("prestage_source_changed"); + const manifestBytes = source?.manifestBytes ?? await readFile(join(input.root, "manifest.json")); const manifestSha256 = `sha256:${createHash("sha256").update(manifestBytes).digest("hex")}`; + const inactiveSource = input.inactiveSource ? { ...input.inactiveSource, root: input.root, manifestSha256, scopeDigest: input.scopeDigest } : undefined; const binding = { companyId: input.owner.companyId, scopeDigest: input.scopeDigest, id: randomUUID() }; const scratch = await mkdtemp(join(os.tmpdir(), "pc-harness-upload-")); const key = randomBytes(32); @@ -163,7 +201,7 @@ export async function persistNativeHarnessBackup(input: { key: key.toString("base64"), iv: iv.toString("base64"), tag: cipher.getAuthTag().toString("base64"), plaintext })) }; if (ref.byteSize > MAX_NATIVE_HARNESS_OBJECT_BYTES) fail("size_limit_exceeded"); await input.db.transaction(async tx => { - await fence(tx, input.owner); + await fence(tx, input.owner, inactiveSource); await tx.insert(nativeHarnessBackups).values({ ...binding, issueId: input.owner.issueId, runId: input.owner.runId, manifestSha256, reference: ref }); }); @@ -172,7 +210,7 @@ export async function persistNativeHarnessBackup(input: { byteSize: ref.byteSize, sha256: ref.sha256, signal: input.signal }); await receive(input.storage, ref, undefined, input.signal); await input.db.transaction(async tx => { - await fence(tx, input.owner); + await fence(tx, input.owner, inactiveSource); await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${`${ref.companyId}:${ref.scopeDigest}`}, 0))`); const [pending] = await tx.select({ status: nativeHarnessBackups.status }).from(nativeHarnessBackups) .where(eq(nativeHarnessBackups.id, ref.id)).for("update"); @@ -182,7 +220,7 @@ export async function persistNativeHarnessBackup(input: { eq(nativeHarnessBackups.status, "current"), sql`${nativeHarnessBackups.runId} <> ${input.owner.runId}::uuid`, sql`${heartbeatRuns.createdAt} >= (select created_at from heartbeat_runs where id = ${input.owner.runId}::uuid)`)).limit(1); if (successor) fail("owner_superseded"); - await fence(tx, input.owner); + await fence(tx, input.owner, inactiveSource); await tx.update(nativeHarnessBackups).set({ status: "superseded", updatedAt: new Date() }) .where(and(scope(ref), eq(nativeHarnessBackups.status, "current"))); await tx.update(nativeHarnessBackups).set({ status: "current", updatedAt: new Date() })