From 4b5e7d0db21fe70717b72a86f047667cde6345e2 Mon Sep 17 00:00:00 2001 From: Dotta Date: Wed, 4 Nov 2026 21:57:08 -0600 Subject: [PATCH] fix: report aborted harness preparation and defer damaged sources Emit the blocked receipt and failure status before waiting for cancellation cleanup. Keep the fallback referenced and cover its real Node event-loop behavior. Distinguish filesystem source damage from database failures at preparation and publication fences. Continue online copying with a source/reason diagnostic while preserving strict offline migration and fatal cancellation. Co-Authored-By: Paperclip --- .../native-harness-prestage.test.ts | 33 ++++++++++ .../maintenance/native-harness-prestage.ts | 17 +++-- .../native-harness-backup-migration.ts | 41 +++++++----- .../native-harness-backup-source.ts | 30 +++++++++ ...harness-backup-storage.integration.test.ts | 65 +++++++++++++++++++ .../native-harness-backup-storage.ts | 11 ++-- 6 files changed, 172 insertions(+), 25 deletions(-) create mode 100644 server/src/services/native-runtime/native-harness-backup-source.ts diff --git a/server/src/maintenance/native-harness-prestage.test.ts b/server/src/maintenance/native-harness-prestage.test.ts index 366123532a0..7d0f8be17f2 100644 --- a/server/src/maintenance/native-harness-prestage.test.ts +++ b/server/src/maintenance/native-harness-prestage.test.ts @@ -1,4 +1,7 @@ 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 () => {}) })); vi.mock("@paperclipai/db", () => ({ createDb: () => ({}), closeRegisteredClients: mocks.close })); @@ -15,6 +18,36 @@ vi.mock("./native-harness-prestage-runtime.js", async importOriginal => ({ const initialExitCode = process.exitCode; afterEach(() => { process.exitCode = initialExitCode; vi.useRealTimers(); vi.unstubAllEnvs(); vi.restoreAllMocks(); vi.resetModules(); mocks.close.mockClear(); }); +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, + // Shorten only the command's deadlines. These remain real Node timers in + // a separate process, so an unreferenced fallback cannot be advanced by us. + 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(() => {}); + ` })); + } }], + }); + 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"); diff --git a/server/src/maintenance/native-harness-prestage.ts b/server/src/maintenance/native-harness-prestage.ts index 7e8674ee460..4d835213c6e 100644 --- a/server/src/maintenance/native-harness-prestage.ts +++ b/server/src/maintenance/native-harness-prestage.ts @@ -15,13 +15,23 @@ 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; + // Report failure before awaiting cleanup: a dependency may never settle. + reportBlocked(new Error(reason)); cancellation.abort(new Error(reason)); // Closing this command's clients interrupts blocked queries. A bounded // fallback also covers an unresponsive storage or filesystem dependency. shutdown = setTimeout(() => process.exit(1), 10_000); - shutdown.unref(); void closeRegisteredClients(url, { scope: "database" }); }; const stop = () => abort("native_harness_prestage_stopped"); @@ -41,10 +51,7 @@ try { cancellation.signal.throwIfAborted(); console.log(JSON.stringify({ schema: "paperclip.native-harness-prestage.v1", counts, enrollmentReady: false })); } catch (error) { - 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 })); - process.exitCode = 1; + reportBlocked(error); } finally { clearTimeout(deadline); process.off("SIGINT", stop); process.off("SIGTERM", stop); await closeRegisteredClients(url, { scope: "database" }); 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 5d7ee216fd4..a3f9f84fbba 100644 --- a/server/src/services/native-runtime/native-harness-backup-migration.ts +++ b/server/src/services/native-runtime/native-harness-backup-migration.ts @@ -1,5 +1,5 @@ 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, inArray, or, sql } from "drizzle-orm"; import { environmentLeases, heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations, type Db } from "@paperclipai/db"; @@ -8,6 +8,7 @@ import { nativeHarnessStateBase, readNativeHarnessBackupStamp, verifyNativeHarne 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(); @@ -105,7 +101,7 @@ async function migrateSource(db: Db, input: { storage: StorageService; signal: A } async function mapSources(input: { signal: AbortSignal; concurrency?: number; - unavailableSource?: (error: unknown) => T | undefined }, visit: (source: Source, signal: AbortSignal) => Promise) { + 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; @@ -134,7 +130,7 @@ async function mapSources(input: { signal: AbortSignal; concurrency?: number; try { source = await prepareSource(base, scopeDigest); } catch (error) { input.signal.throwIfAborted(); - const deferred = input.unavailableSource?.(error); + const deferred = input.unavailableSource?.(error, scopeDigest); if (deferred === undefined) throw error; results.push(deferred); continue; } @@ -162,9 +158,20 @@ export async function migrateNativeHarnessBackups(db: Db, input: { export async function prestageNativeHarnessBackups(db: Db, input: { storage: StorageService; signal: AbortSignal }) { // Keep preparatory work below offline throughput so the serving process can // continue handling real work. This finite operation has no idle schedule. - const outcomes = await mapSources({ ...input, concurrency: 1, unavailableSource: error => - (error as NodeJS.ErrnoException)?.code === "ENOENT" || error instanceof Error && error.message === "native_harness_migration_source_unsettled" - ? "source_unavailable" : undefined }, async (source, signal) => { + 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"); + } + // This callback only receives preparation filesystem failures, never DB errors. + 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; @@ -201,9 +208,11 @@ export async function prestageNativeHarnessBackups(db: Db, input: { storage: Sto return "published"; } catch (error) { signal.throwIfAborted(); - if ((error as NodeJS.ErrnoException)?.code === "ENOENT" || error instanceof Error && ["native_runner_checkpoint_owner_lost", "native_harness_storage_owner_missing", + 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 "source_changed"; + "native_harness_storage_owner_superseded", "native_harness_archive_invalid"].includes(error.message)) { + return deferSource(source.scopeDigest, error.message, "source_changed"); + } throw error; } }); 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..9479474920f --- /dev/null +++ b/server/src/services/native-runtime/native-harness-backup-source.ts @@ -0,0 +1,30 @@ +import { lstat, readFile } from "node:fs/promises"; +import { join } from "node:path"; + +/** Only filesystem inspection can produce this error. Database failures must + * retain their identity, even if their class or code resembles a source error. */ +export class NativeHarnessBackupSourceError extends Error {} + +export async function readNativeHarnessBackupSource(root: string) { + try { + 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 NativeHarnessBackupSourceError("native_harness_migration_unsafe_source"); + const manifestBytes = await readFile(join(root, "manifest.json")); + const manifest: unknown = JSON.parse(manifestBytes.toString("utf8")); + const state: unknown = JSON.parse(await 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 }); + 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; + } +} 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 fde6df5306c..d2ee1d52295 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 @@ -224,6 +224,71 @@ describeEmbeddedPostgres("durable native provider backups", () => { 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(["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)); 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 eff1a75f618..5c5fb935836 100644 --- a/server/src/services/native-runtime/native-harness-backup-storage.ts +++ b/server/src/services/native-runtime/native-harness-backup-storage.ts @@ -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 }; @@ -165,9 +166,9 @@ async function fence(tx: Parameters[0]>[0], owner: } // The task share lock serializes publication with execution checkout. A // turn that ran during upload also changes the immutable backup manifest. - const manifest = await readFile(join(inactiveSource.root, "manifest.json")); - const identity = JSON.parse(manifest.toString("utf8")); - const state = JSON.parse(await readFile(join(inactiveSource.root, "runner", "runner-state.json"), "utf8")); + 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 @@ -181,7 +182,9 @@ export async function persistNativeHarnessBackup(input: { 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() };