diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index fb9b1717564..a5e128b9af7 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -4070,6 +4070,13 @@ Environment overrides: negative, empty, or non-numeric value falls back to the default. `PAPERCLIP_NATIVE_HARNESS_BACKUP_EXPIRATION_ENABLED=false` turns expiry off, and `PAPERCLIP_NATIVE_HARNESS_BACKUP_EXPIRATION_DRY_RUN=true` only reports candidates. + A resumed session restores its failover backup from durable storage before the + provider starts. Each read has a five-minute idle limit instead of a short total + limit: a large or slow backup keeps restoring while bytes flow. A read that makes + no progress for five minutes fails with `native_harness_storage_download_stalled` + and publishes no partial backup. The whole restore, including local decrypt, + digest and extraction, has a 30-minute outer limit + (`native_harness_storage_restore_deadline_exceeded`). Stop ends the restore. - `PAPERCLIP_NATIVE_RUNNER_EVIDENCE_RETENTION_DAYS=` sets how long native session continuity-break archives and quarantined session state are kept (see `doc/recoverable-disk-cleanup.md`). The default is `7` and the minimum is `1`. A diff --git a/server/src/services/native-runtime/native-harness-backup-download-deadline.integration.test.ts b/server/src/services/native-runtime/native-harness-backup-download-deadline.integration.test.ts new file mode 100644 index 00000000000..e0edbce71bf --- /dev/null +++ b/server/src/services/native-runtime/native-harness-backup-download-deadline.integration.test.ts @@ -0,0 +1,173 @@ +import fs from "node:fs/promises"; +import { createHash, randomUUID } from "node:crypto"; +import { mkdir, readFile, stat, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { Readable } from "node:stream"; +import { and, eq } from "drizzle-orm"; +import { heartbeatRuns, nativeHarnessBackups, nativeRunFinalizations } from "@paperclipai/db"; +import { afterEach, expect, it, vi } from "vitest"; +import { describeEmbeddedPostgres, nativeRestartFixture } from "../../__tests__/helpers/native-restart.js"; +import { createLocalDiskStorageProvider } from "../../storage/local-disk-provider.js"; +import { createStorageService } from "../../storage/service.js"; +import type { StorageService } from "../../storage/types.js"; +import { nativeHarnessBackupOwner, persistNativeHarnessBackup, restoreDurableNativeHarnessBackup, + verifyDurableNativeHarnessBackup } from "./native-harness-backup-storage.js"; +import { removeNativeHarnessBackup } from "./native-harness-backup-cleanup.js"; +import { canonicalProviderPackJson } from "./remote-provider-pack-verification.js"; + +vi.mock("node:fs/promises", async importOriginal => { + const original = await importOriginal(); + const files = { ...original }; + return { ...original, default: files, cp: (...args: Parameters) => files.cp(...args) }; +}); + +/** Durable reads that deliver `chunkBytes` every `delayMs`, optionally stopping after `stallAfter` chunks. */ +function throttledStorage(storage: StorageService, delayMs: number, chunkBytes: number, stallAfter?: number) { + let sent = 0; + const reads: string[] = []; + const throttled = new Proxy(storage, { get(target, property) { + if (property !== "getObject") { + const value = Reflect.get(target, property, target) as unknown; + return typeof value === "function" ? (value as (...args: unknown[]) => unknown).bind(target) : value; + } + return async (companyId: string, objectKey: string, options?: { signal?: AbortSignal }) => { + reads.push(objectKey); + const object = await target.getObject(companyId, objectKey, options); + async function* slow() { + for await (const part of object.stream as AsyncIterable) { + for (let offset = 0; offset < part.length; offset += chunkBytes) { + if (stallAfter !== undefined && sent >= stallAfter) { + await new Promise(resolve => options?.signal?.addEventListener("abort", () => resolve(), { once: true })); + options?.signal?.throwIfAborted(); + } + await new Promise(resolve => setTimeout(resolve, delayMs)); + options?.signal?.throwIfAborted(); + sent++; + yield part.subarray(offset, offset + chunkBytes); + } + } + } + return { ...object, stream: Readable.from(slow()) }; + }; + } }); + return { storage: throttled, reads }; +} + +describeEmbeddedPostgres("durable native backup download deadlines", () => { + const environment = nativeRestartFixture(); + const ownedHomes: string[] = []; + afterEach(async () => { + for (const home of ownedHomes.splice(0)) await removeNativeHarnessBackup(home); + vi.restoreAllMocks(); + vi.unstubAllEnvs(); + }); + /** An OpenCode backup with provider artifacts, so restores make two separate durable reads. */ + async function fixture() { + vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "51".repeat(32)); + const f = await environment.seedRun(randomUUID()); + const { companyId, runtimeRoot, successor, agentId } = environment.current; + await f.db.update(nativeRunFinalizations).set({ controllerBootId: successor.bootId, controllerGeneration: 1, + leaseOwner: "backup-writer", leaseExpiresAt: new Date(Date.now() + 120_000) }).where(eq(nativeRunFinalizations.runId, f.runId)); + const owner = await nativeHarnessBackupOwner(f.db, { companyId, issueId: f.issueId, runId: f.runId, bootId: successor.bootId }); + const home = join(runtimeRoot, randomUUID()), root = join(home, "failover-backups", "current"); + ownedHomes.push(home); + await mkdir(join(root, "opencode", "sessions"), { recursive: true }); + const identity = { normalizedSessionId: randomUUID(), providerSessionId: randomUUID(), provider: "opencode" }; + // Enough bytes for many throttled chunks, so a restore outlasts the idle window. + await writeFile(join(root, "opencode", "sessions", "history.json"), JSON.stringify({ ...identity, turns: "x".repeat(4096) })); + const manifest = JSON.stringify({ schema: "paperclip.native-harness-backup.v1", ...identity, directories: ["opencode"] }); + await writeFile(join(root, "manifest.json"), manifest, { mode: 0o600 }); + const sourceRoot = join(home, "qualified-provider-pack"); + await mkdir(sourceRoot); + const original = { nodeCommand: "qualified Node executable", opencodeExecutable: "qualified OpenCode executable", + opencodeProxy: `qualified proxy bytes ${randomUUID()}` }; + const artifacts = Object.fromEntries(await Promise.all(Object.entries(original).map(async ([name, bytes]) => { + await writeFile(join(sourceRoot, name), bytes, { mode: 0o500 }); + return [name, { path: name, sha256: `sha256:${createHash("sha256").update(bytes).digest("hex")}` }]; + }))); + const payload = { runnerSourceRevision: "a".repeat(40), target: { platform: "linux", architecture: "x64" }, + pins: { opencode: "1.18.34" }, artifacts: { ...artifacts, opencodeCommand: artifacts.opencodeExecutable } }; + const packManifest = { schema: "paperclip-runner/remote-provider-pack/v1", payload, + digest: `sha256:${createHash("sha256").update(canonicalProviderPackJson(payload)).digest("hex")}` }; + const launchRoot = `/home/daytona/.paperclip-runtime/paperclip-runner/provider-artifacts/${packManifest.digest.slice(7)}`; + await f.db.update(heartbeatRuns).set({ nativeSessionId: identity.normalizedSessionId, driverKind: "opencode_server", + runnerProfileJson: { ...f.preparedProfile, nativeProviderPackAdmissionState: { + schema: "paperclip.native-provider-pack-admission-state.v1", state: "verified", admissionEpoch: randomUUID() }, + nativeProviderPackAdmission: { schema: "paperclip.native-provider-pack-admission.v1", stage: "session_admitted", + authority: "core_verified_selection", companyId, agentId, runId: f.runId, sessionId: identity.normalizedSessionId, + runnerInstanceId: "deadline-fixture-runner", runnerEnvironmentLeaseId: "deadline-fixture-lease", + providerPackDigest: packManifest.digest, opencodeProxySha256: artifacts.opencodeProxy.sha256, + sourceRevision: payload.runnerSourceRevision } }, + }).where(eq(heartbeatRuns.id, f.runId)); + const scopeDigest = createHash("sha256").update(randomUUID()).digest("hex"); + const storage = createStorageService(createLocalDiskStorageProvider(join(runtimeRoot, "durable-objects"))); + await persistNativeHarnessBackup({ db: f.db, root, owner, scopeDigest, storage, + providerArtifacts: { sourceRoot, manifest: packManifest, launchRoot, accountUid: 1000 } }); + const [row] = await f.db.select().from(nativeHarnessBackups).where(and(eq(nativeHarnessBackups.scopeDigest, scopeDigest), + eq(nativeHarnessBackups.status, "current"))); + return { db: f.db, companyId, scopeDigest, storage, identity, original, manifestSha256: row.manifestSha256 }; + } + function replacementRoot() { + const root = join(environment.current.runtimeRoot, randomUUID(), "failover-backups"); + ownedHomes.push(root); + return root; + } + + it("keeps restoring a backup whose downloads outlast the idle window while bytes keep arriving", async () => { + const f = await fixture(), idle = 250, slow = throttledStorage(f.storage, 15, 64); + // The former total deadline aborted this same progressing transfer with + // the generic "The operation was aborted" seen on resumed sessions. + const aborted = replacementRoot(); + await expect(restoreDurableNativeHarnessBackup({ ...f, storage: slow.storage, backupRoot: aborted, signal: AbortSignal.timeout(idle) })) + .rejects.toThrow("The operation was aborted"); + await expect(stat(join(aborted, "current"))).rejects.toMatchObject({ code: "ENOENT" }); + const replacement = replacementRoot(), startedAt = Date.now(); + slow.reads.splice(0); + expect(await restoreDurableNativeHarnessBackup({ ...f, storage: slow.storage, backupRoot: replacement, + downloadIdleTimeoutMs: idle })).toBe(true); + expect(Date.now() - startedAt).toBeGreaterThan(idle * 2); + // Both the backup object and the provider artifacts were read through the deadline. + expect(slow.reads.some(key => key.includes("/native-harness-backups/"))).toBe(true); + expect(slow.reads.some(key => !key.includes("/native-harness-backups/"))).toBe(true); + expect(JSON.parse(await readFile(join(replacement, "current/opencode/sessions/history.json"), "utf8"))) + .toMatchObject(f.identity); + expect(await readFile(join(replacement, "current/provider-artifacts/opencodeProxy"), "utf8")).toBe(f.original.opencodeProxy); + expect(await verifyDurableNativeHarnessBackup({ ...f, storage: slow.storage, downloadIdleTimeoutMs: idle })).toBe(true); + }); + + it("names a download that stops sending bytes and publishes no partial backup", { timeout: 10_000 }, async () => { + const f = await fixture(), replacement = replacementRoot(); + await expect(restoreDurableNativeHarnessBackup({ ...f, storage: throttledStorage(f.storage, 5, 64, 3).storage, + backupRoot: replacement, downloadIdleTimeoutMs: 250 })).rejects.toThrow("native_harness_storage_download_stalled"); + await expect(stat(join(replacement, "current"))).rejects.toMatchObject({ code: "ENOENT" }); + await expect(verifyDurableNativeHarnessBackup({ ...f, storage: throttledStorage(f.storage, 5, 64, 3).storage, + downloadIdleTimeoutMs: 250 })).rejects.toThrow("native_harness_storage_download_stalled"); + // The saved backup is untouched and restores once the store responds again. + expect(await restoreDurableNativeHarnessBackup({ ...f, backupRoot: replacement, downloadIdleTimeoutMs: 250 })).toBe(true); + }); + + it("stops at the outer bound during local copy and publishes nothing", async () => { + const f = await fixture(), replacement = replacementRoot(), copy = fs.cp.bind(fs); + vi.spyOn(fs, "cp").mockImplementation(async (source, destination, options) => { + if (String(destination).includes(".durable-")) await new Promise(resolve => setTimeout(resolve, 400)); + return copy(source, destination, options); + }); + await expect(restoreDurableNativeHarnessBackup({ ...f, backupRoot: replacement, downloadIdleTimeoutMs: 5_000, + restoreMaxMs: 200 })).rejects.toThrow("native_harness_storage_restore_deadline_exceeded"); + await expect(stat(join(replacement, "current"))).rejects.toMatchObject({ code: "ENOENT" }); + }); + + it("lets Stop end a progressing download with its own reason, including a Stop before it starts", async () => { + const f = await fixture(), replacement = replacementRoot(), stop = new AbortController(); + setTimeout(() => stop.abort(new Error("caller_stopped")), 50); + const failure = await restoreDurableNativeHarnessBackup({ ...f, storage: throttledStorage(f.storage, 15, 64).storage, + backupRoot: replacement, signal: stop.signal, downloadIdleTimeoutMs: 5_000 }).then(() => null, (error: unknown) => error); + const reason = failure as Error & { cause?: unknown }; + expect(failure).toBeInstanceOf(Error); + expect([reason.message, (reason.cause as Error | undefined)?.message]).toContain("caller_stopped"); + await expect(stat(join(replacement, "current"))).rejects.toMatchObject({ code: "ENOENT" }); + await expect(restoreDurableNativeHarnessBackup({ ...f, backupRoot: replacement, signal: stop.signal, + downloadIdleTimeoutMs: 5_000 })).rejects.toThrow(); + await expect(stat(join(replacement, "current"))).rejects.toMatchObject({ code: "ENOENT" }); + }); +}); 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 d285e211b07..993952c7a9a 100644 --- a/server/src/services/native-runtime/native-harness-backup-storage.ts +++ b/server/src/services/native-runtime/native-harness-backup-storage.ts @@ -57,7 +57,7 @@ export async function verifyRecoverableNativeHarnessBackup(input: { const storage = durableNativeHarnessStorage(); if (!storage) return verifyNativeHarnessBackupStamp(stamp, input.providerLeaseId); const durable = await verifyDurableNativeHarnessBackup({ db: input.db, storage, companyId: input.companyId, - scopeDigest: stamp.sessionScopeSha256, manifestSha256: stamp.manifestSha256, signal: AbortSignal.timeout(300_000) }); + scopeDigest: stamp.sessionScopeSha256, manifestSha256: stamp.manifestSha256, downloadIdleTimeoutMs: NATIVE_HARNESS_DOWNLOAD_IDLE_TIMEOUT_MS }); return durable || process.env.PAPERCLIP_EPHEMERAL_SERVER !== "true" && verifyNativeHarnessBackupStamp(stamp, input.providerLeaseId); } function reference(row: Row): Reference { @@ -287,7 +287,82 @@ export async function persistNativeHarnessBackup(input: { } finally { key.fill(0); await removeNativeHarnessBackup(scratch); } } +/** How long a durable read may go without progress. */ +export const NATIVE_HARNESS_DOWNLOAD_IDLE_TIMEOUT_MS = 300_000; +/** Outer bound for one whole restore or verification, including local steps. */ +export const NATIVE_HARNESS_RESTORE_MAX_MS = 30 * 60_000; +export const NATIVE_HARNESS_DOWNLOAD_STALLED = "native_harness_storage_download_stalled"; +export const NATIVE_HARNESS_RESTORE_DEADLINE_EXCEEDED = "native_harness_storage_restore_deadline_exceeded"; + +/** + * Bound each durable read by inactivity, not by a short total time. A large or + * slow backup keeps restoring while bytes flow through the read; a read that + * makes no progress for the idle window aborts the whole restore with a named + * error. A generous outer bound keeps a hung local step diagnosable. Caller + * cancellation (Stop) keeps its own reason. + */ +async function withDownloadProgressDeadline( + input: { storage: StorageService; signal?: AbortSignal; downloadIdleTimeoutMs?: number; restoreMaxMs?: number }, + run: (storage: StorageService, signal: AbortSignal | undefined) => Promise, +): Promise { + const idleMs = input.downloadIdleTimeoutMs; + if (idleMs === undefined) return run(input.storage, input.signal); + if (!Number.isSafeInteger(idleMs) || idleMs <= 0) throw new Error("native_harness_storage_idle_timeout_invalid"); + const stall = new AbortController(); + const outer = setTimeout(() => stall.abort(new Error(NATIVE_HARNESS_RESTORE_DEADLINE_EXCEEDED)), + input.restoreMaxMs ?? Math.max(idleMs, NATIVE_HARNESS_RESTORE_MAX_MS)); + outer.unref?.(); + const signal = input.signal ? AbortSignal.any([input.signal, stall.signal]) : stall.signal; + const base = input.storage; + const getObject: StorageService["getObject"] = async (companyId, objectKey, options) => { + let timer: NodeJS.Timeout | undefined; + const disarm = () => { if (timer) clearTimeout(timer); timer = undefined; }; + const arm = () => { + disarm(); + timer = setTimeout(() => stall.abort(new Error(NATIVE_HARNESS_DOWNLOAD_STALLED)), idleMs); + timer.unref?.(); + }; + arm(); + let object: Awaited>; + try { + object = await base.getObject(companyId, objectKey, { + ...options, signal: options?.signal ? AbortSignal.any([options.signal, signal]) : signal }); + } catch (error) { disarm(); throw error; } + arm(); + // Chunks pass through as the consumer reads them, so a read is idle when + // neither the store nor the local writer makes progress. + const observed = new Transform({ + transform(bytes, _encoding, callback) { arm(); callback(null, bytes); }, + flush(callback) { disarm(); callback(); }, + }); + observed.once("close", disarm); + pipeline(object.stream, observed, { signal }).catch(() => disarm()); + return { ...object, stream: observed }; + }; + // Forward every other member to the real service so its own state stays intact. + const storage = new Proxy(base, { + get(target, property) { + if (property === "getObject") return getObject; + const value = Reflect.get(target, property, target) as unknown; + return typeof value === "function" ? (value as (...args: unknown[]) => unknown).bind(target) : value; + }, + }); + try { + return await run(storage, signal); + } catch (error) { + if (stall.signal.aborted && !input.signal?.aborted) throw stall.signal.reason; + throw error; + } finally { clearTimeout(outer); } +} + export async function restoreDurableNativeHarnessBackup(input: { + db: Db; storage: StorageService; companyId: string; scopeDigest: string; + backupRoot: string; signal?: AbortSignal; downloadIdleTimeoutMs?: number; restoreMaxMs?: number; +}): Promise { + return withDownloadProgressDeadline(input, (storage, signal) => restoreCurrentNativeHarnessBackup({ ...input, storage, signal })); +} + +async function restoreCurrentNativeHarnessBackup(input: { db: Db; storage: StorageService; companyId: string; scopeDigest: string; backupRoot: string; signal?: AbortSignal; }): Promise { @@ -332,6 +407,8 @@ export async function restoreDurableNativeHarnessBackup(input: { const pending = join(input.backupRoot, `.durable-${randomUUID()}`); try { await cp(restored, pending, { recursive: true, preserveTimestamps: true, verbatimSymlinks: true, errorOnExist: true, force: false }); + // A deadline or Stop that fired during the local copy must not publish. + input.signal?.throwIfAborted(); await publishNativeHarnessBackup({ db: input.db, companyId: input.companyId, scopeDigest: input.scopeDigest, backupRoot: input.backupRoot, pendingRoot: pending }); } finally { await removeNativeHarnessBackup(pending); } @@ -340,13 +417,14 @@ export async function restoreDurableNativeHarnessBackup(input: { } export async function verifyDurableNativeHarnessBackup(input: { - db: Db; storage: StorageService; companyId: string; scopeDigest: string; manifestSha256: string; backupId?: string; signal?: AbortSignal; + db: Db; storage: StorageService; companyId: string; scopeDigest: string; manifestSha256: string; backupId?: string; + signal?: AbortSignal; downloadIdleTimeoutMs?: number; }): Promise { const [row] = await input.db.select().from(nativeHarnessBackups).where(and(scope(input), eq(nativeHarnessBackups.manifestSha256, input.manifestSha256), input.backupId ? eq(nativeHarnessBackups.id, input.backupId) : undefined, inArray(nativeHarnessBackups.status, ["current", "superseded"]))).limit(1); if (!row) return false; - await receive(input.storage, reference(row), undefined, input.signal); + await withDownloadProgressDeadline(input, (storage, signal) => receive(storage, reference(row), undefined, signal)); return true; } diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 331748456eb..9d0c3b7f7a2 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -13,7 +13,8 @@ import { readProviderFailure } from "@paperclipai/shared"; import { nativeCancellationHasUnsettledOwner, notifyNativeCancellation, waitForNativeCancellation } from "./native-cancellation-delivery.js"; import { readClaimedNativeStopRecovery } from "./native-cancellation-request.js"; import { publishNativeHarnessBackup } from "./native-harness-backup-publication.js"; -import { assertNativeHarnessBackupOwner, durableNativeHarnessStorage, nativeHarnessBackupOwner, persistNativeHarnessBackup, restoreDurableNativeHarnessBackup } from "./native-harness-backup-storage.js"; +import { assertNativeHarnessBackupOwner, durableNativeHarnessStorage, NATIVE_HARNESS_DOWNLOAD_IDLE_TIMEOUT_MS, nativeHarnessBackupOwner, persistNativeHarnessBackup, + restoreDurableNativeHarnessBackup } from "./native-harness-backup-storage.js"; import { restoreNativeProviderArtifacts, retainNativeProviderArtifacts, retainExistingNativeProviderArtifacts, type NativeProviderArtifactReceipt } from "./native-provider-artifacts.js"; import { prepareRemoteOpenCodeArtifacts, type OpenCodeLaunchArtifacts } from "./remote-opencode-artifacts.js"; @@ -7275,9 +7276,16 @@ function harnessBackupRoot(root: string): string { async function restoreDurableHarnessForExecution(db: Db, execution: NativeExecutionInput, root: string) { const storage = durableNativeHarnessStorage(); if (!storage) return; + // Stop or lease loss during startup aborts this source, including before this + // restore begins. Only a completed startup's signal is not reused. + const startupSignal = nativeSessionStartups.get(execution.binding.runId)?.providerPackSourceAbort.signal; + const completed = startupSignal?.aborted && startupSignal.reason instanceof Error + && startupSignal.reason.message === "native_execution_startup_completed"; + const stop = completed ? undefined : startupSignal; await restoreDurableNativeHarnessBackup({ db, storage, companyId: execution.binding.companyId, scopeDigest: createHash("sha256").update(nativeSessionScopeKey(execution)).digest("hex"), - backupRoot: harnessBackupRoot(root), signal: AbortSignal.timeout(300_000) }); + backupRoot: harnessBackupRoot(root), signal: stop, + downloadIdleTimeoutMs: NATIVE_HARNESS_DOWNLOAD_IDLE_TIMEOUT_MS }); } function harnessBackupCandidates(root: string): string[] {