diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 95d333ac571..b3a453458e8 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1275,7 +1275,12 @@ Another task can hold the same routine execution slot even when this task's lock is free. If that unique claim fails, restart recovery rolls back the whole candidate claim and waits through the existing recovery deadline. It preserves the current routine owner, checkpoint, and provider attempt, and continues the -startup batch. Other database failures still fail the recovery pass. +startup batch. Any other failure while claiming one run is handled the same +way with reason `restart_recovery_failed`: the run's claim rolls back, the +error is logged, a warning transition is recorded on the run, and the run waits +for the next deadline. A failed transition write is logged and does not stop +the claims that already committed. Only a failure of the candidate scan itself +fails the recovery pass. Automatic productivity reviews are retired. Run counts, missing comments, and elapsed task time do not create review tasks or impose continuation holds. Bounded continuation, provider recovery, budget limits, explicit blockers, and normal review/approval stages remain in force. Existing productivity-review tasks, comments, assignments, and dependencies remain unchanged and readable; their historical origins still identify them as recovery work for recursion suppression. diff --git a/server/src/services/heartbeat/recovery.test.ts b/server/src/services/heartbeat/recovery.test.ts index d109fcec13c..3c92e5ed84b 100644 --- a/server/src/services/heartbeat/recovery.test.ts +++ b/server/src/services/heartbeat/recovery.test.ts @@ -13,7 +13,7 @@ import { createHeartbeatRunState } from "./run-state.js"; import type { HotRestartIntent } from "../hot-restart.js"; import type { NativeRestartRecoveryClaim, NativeRestartRecoveryDisposition } from "../native-runtime/index.js"; import { NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE } from "../native-runtime/native-runner-ownership.js"; -import { claimNativeRestartRecoveries, currentNativeControllerIdentity } from "../native-runtime/native-restart-recovery.js"; +import { claimNativeRestartRecoveries, currentNativeControllerIdentity, NATIVE_RESTART_RECOVERY_FAILED } from "../native-runtime/native-restart-recovery.js"; import { startNativeSessionExecutionLeaseRenewal } from "../native-runtime/native-execution-lease.js"; import { allowLegacyShutdownWorkspaceCleanup, beginLegacyShutdownWorkspaceSettlement, finishLegacyShutdownWorkspaceSettlement, maintainLegacyShutdownWorkspaceSettlement, legacyShutdownWorkspaceResourceProtected, sweepLegacyShutdownWorkspaceSettlements } from "../legacy-shutdown-workspace-settlement.js"; import { adapterExecutionControls, createAdapterExecutionControl } from "../adapter-execution-control.js"; @@ -1203,6 +1203,29 @@ describePostgres("heartbeat recovery module ownership", () => { } }); + it.each(["getRun", "appendRunEvent"] as const)("dispatches committed claims when recording a transition fails in %s", async failing => { + const deps = callbacks(db); + const deferred = { kind: "awaiting_evidence" as const, runId: randomUUID(), reason: NATIVE_RESTART_RECOVERY_FAILED, + retryAt: new Date(Date.now() + 30_000) }; + const claim: NativeRestartRecoveryClaim = { kind: "bootstrap_incomplete", runId: randomUUID(), leaseOwner: "claimed-owner", + controllerGeneration: 1, providerAttempt: 1, restartKind: "hard", recoveryRequestId: null }; + native.claims.mockResolvedValueOnce([deferred, claim]); + if (failing === "getRun") deps.getRun.mockRejectedValue(new Error("run read failed")); + else { + deps.getRun.mockImplementation(async id => ({ id }) as unknown as Awaited>); + deps.appendRunEvent.mockRejectedValue(new Error("event write failed")); + } + const result = await createHeartbeatRecovery(db, deps).recoverNativeRunsAfterRestart(); + expect(result.claims).toEqual([claim]); + expect(result.awaitingEvidenceRunIds).toEqual([deferred.runId]); + expect(deps.executeRun).toHaveBeenCalledExactlyOnceWith(claim.runId, { nativeLeaseOwner: claim.leaseOwner, nativeRestartRecovery: claim }); + if (failing === "appendRunEvent") { + expect(deps.appendRunEvent).toHaveBeenCalledWith(expect.objectContaining({ id: deferred.runId }), expect.objectContaining({ + level: "warn", payload: expect.objectContaining({ reason: NATIVE_RESTART_RECOVERY_FAILED }) })); + } + await Promise.all([...deps.activeRunExecutionPromises]); + }); + it("does not claim or execute recovery when persisted finalization fails", async () => { const deps = callbacks(db); native.finalizations.mockRejectedValueOnce(new Error("result settlement failed")); diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 739a7584444..bf701c8d55d 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -67,6 +67,7 @@ import { dispatchNativeSessionResumptions, detachNativeSessionsForRestart, isNativeSessionId, + NATIVE_RESTART_RECOVERY_FAILED, readNativeWorkspaceSyncReference, type NativeRestartRecoveryClaim, reconcileNativeFinalizations, @@ -1018,7 +1019,13 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD // No ownership decision was made for a locked row. Writing an event // would wait on that same row and defeat the non-blocking startup scan. if (disposition.kind === "awaiting_evidence" && disposition.reason === "recovery_rows_busy") continue; - const run = await getRun(disposition.runId); + // Claims are already committed and must still dispatch, so one failed + // audit write cannot stop the rest of the batch. + const auditFailed = (error: unknown) => { + logger.warn({ err: error, runId: disposition.runId }, "failed to record native restart recovery transition"); + return null; + }; + const run = await getRun(disposition.runId).catch(auditFailed); if (run) { const isClaim = disposition.kind === "reattach_existing_runner" || @@ -1028,7 +1035,8 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD await appendRunEvent(run, { eventType: "native.recovery.transition", stream: "system", - level: disposition.kind === "blocked" ? "warn" : "info", + level: disposition.kind === "blocked" || (disposition.kind === "awaiting_evidence" + && disposition.reason === NATIVE_RESTART_RECOVERY_FAILED) ? "warn" : "info", message: disposition.kind === "reattach_existing_runner" || disposition.kind === "reattach_remote_runner" @@ -1037,6 +1045,8 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD ? "Resuming the durable native provider session after runner process loss" : disposition.kind === "bootstrap_incomplete" ? "Restarting an incomplete native runner bootstrap on the same heartbeat run" + : disposition.kind === "awaiting_evidence" && disposition.reason === NATIVE_RESTART_RECOVERY_FAILED + ? "Native restart recovery failed for this run and will retry" : disposition.kind === "awaiting_evidence" ? "Native restart recovery is waiting for safe ownership evidence" : disposition.kind === "already_finalized" @@ -1064,7 +1074,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD } : {}), }, - }); + }).catch(auditFailed); } } for (const claim of claims) { diff --git a/server/src/services/native-runtime/native-failed-workspace-recovery.test.ts b/server/src/services/native-runtime/native-failed-workspace-recovery.test.ts index eaf445caa85..6f94d461e24 100644 --- a/server/src/services/native-runtime/native-failed-workspace-recovery.test.ts +++ b/server/src/services/native-runtime/native-failed-workspace-recovery.test.ts @@ -13,7 +13,7 @@ import { getExecutionBlocker } from "../execution-blocker.js"; import { markExecutionReconciliation, settleUnrecoverableExecutions, validateExecutionReconciliation } from "../execution-recovery-resolution.js"; import { prepareSandboxStopAndRetain, settleStopOnlyCleanup } from "../sandbox-stop-and-retain.js"; import { preserveCancelledNativeWorkspaces, preserveNativeFailedWorkspace, preserveOrphanedNativeWorkspace, registerCancelledWorkspacePreservationWork } from "./native-failed-workspace-recovery.js"; -import { claimNativeRestartRecoveries, currentNativeControllerIdentity, type NativeRestartRecoveryClaim } from "./native-restart-recovery.js"; +import { claimNativeRestartRecoveries, currentNativeControllerIdentity, NATIVE_RESTART_RECOVERY_FAILED, type NativeRestartRecoveryClaim } from "./native-restart-recovery.js"; import { NativeWorkspaceFinalizationBusyError } from "./native-workspace-finalization-ownership.js"; import { preserveWorkspaceRestoreRecoveryMetadataSql } from "../legacy-workspace-restore-recovery.js"; import { claimNativeSessionResumptions } from "./native-finalization-reconciler.js"; @@ -119,7 +119,10 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes $$`); await db.execute(sql`create trigger ${name} before update on heartbeat_runs for each row execute function ${name}()`); try { - await expect(claimNativeRestartRecoveries({ db, restartKind: "hard", runIds: [f.runId] })).rejects.toThrow(); + // The run's whole claim rolls back and waits for the next deadline; the batch continues. + await expect(claimNativeRestartRecoveries({ db, restartKind: "hard", runIds: [f.runId] })).resolves.toEqual([ + { kind: "awaiting_evidence", runId: f.runId, reason: NATIVE_RESTART_RECOVERY_FAILED, retryAt: expect.any(Date) }, + ]); expect(await readRun(f.runId)).toMatchObject({ status: "running", nativePhase: "observed", resultJson: { originalDiagnostic: true } }); expect(await readLease(f.lease.id)).toMatchObject({ status: "active", leasePolicy: "ephemeral" }); expect(await actions(f.issueId)).toHaveLength(0); diff --git a/server/src/services/native-runtime/native-restart-recovery-isolation.test.ts b/server/src/services/native-runtime/native-restart-recovery-isolation.test.ts new file mode 100644 index 00000000000..22cd7b5d89c --- /dev/null +++ b/server/src/services/native-runtime/native-restart-recovery-isolation.test.ts @@ -0,0 +1,149 @@ +import { randomUUID } from "node:crypto"; +import { and, eq, inArray, sql } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { + agents, applyPendingMigrations, companies, createDb, environmentLeases, environments, heartbeatRunEvents, + heartbeatRuns, issues, nativeRunFinalizations, +} from "@paperclipai/db"; +import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { heartbeatService } from "../heartbeat.js"; +import { claimNativeRestartRecoveries, NATIVE_RESTART_RECOVERY_FAILED } from "./native-restart-recovery.js"; + +const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_TEST_DATABASE_URL"); +const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); + +// An hourly routine has two open tasks. The older task's remote run failed with +// a scheduled resume, and its execution lock was released. The current hour's +// task holds the routine's execution slot. Startup recovery also meets a run +// whose claim fails for an unrelated reason. Neither may stop the batch. +(support.supported ? describe : describe.skip)("native restart recovery isolates one run's failure", () => { + let db: ReturnType; + let temporary: Awaited> | undefined; + const companyId = randomUUID(); + const agentId = randomUUID(); + beforeAll(async () => { + if (externalUrl) { + await applyPendingMigrations(externalUrl); + db = createDb(externalUrl); + } else { + temporary = await startEmbeddedPostgresTestDatabase("native-restart-recovery-isolation-"); + db = createDb(temporary.connectionString); + } + await db.insert(companies).values({ id: companyId, name: "Restart isolation", issuePrefix: companyId.slice(0, 8) }); + await db.insert(agents).values({ id: agentId, companyId, name: "Engineer", adapterType: "paperclip_runner" }); + }, 30_000); + afterAll(async () => { await db?.$client.end(); await temporary?.cleanup(); }); + + async function seedScheduledResume(options: { lock: "self" | "none"; routine?: { id: string; fingerprint: string } }) { + const issueId = randomUUID(), runId = randomUUID(), sessionId = randomUUID(); + const failedAt = new Date(Date.now() - 5 * 60_000); + await db.insert(issues).values({ id: issueId, companyId, title: "Hourly report", status: "in_progress", + assigneeAgentId: agentId, + ...(options.routine ? { originKind: "routine_execution", originId: options.routine.id, + originFingerprint: options.routine.fingerprint } : {}) }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, issueId, nativeIssueId: issueId, + runtimeMode: "native", status: "failed", nativePhase: "retryable_failure", nativeSessionId: sessionId, + processPid: 4_242, processStartedAt: failedAt, startedAt: failedAt, finishedAt: new Date(), + errorCode: "native_session_interrupted", contextSnapshot: { issueId } }); + const [environment] = await db.insert(environments).values({ name: `Restart isolation ${runId}`, driver: "sandbox" }).returning(); + const [lease] = await db.insert(environmentLeases).values({ companyId, issueId, heartbeatRunId: runId, + environmentId: environment.id, provider: "daytona", providerLeaseId: randomUUID(), status: "active", + leasePolicy: "ephemeral", metadata: { driver: "sandbox", remoteCwd: "/work" } }).returning(); + await db.update(heartbeatRuns).set({ runnerProfileJson: { + nativeWorkspaceSync: { + schema: "paperclip.native-workspace-sync/v3", state: "prepared", descriptorSha256: "a".repeat(64), + baselineSha256: "b".repeat(64), finalHostSha256: null, workspaceId: runId, + leaseId: lease.id, providerLeaseId: lease.providerLeaseId, remoteCwd: "/work", resourceDisposition: "destroy", + }, + sessionCheckpoint: { sessionId: "driver-session", providerSessionId: "provider-session", + identity: { runId, companyId, issueId, agentId, sessionId } }, + } }).where(eq(heartbeatRuns.id, runId)); + if (options.lock === "self") { + await db.update(issues).set({ executionRunId: runId, checkoutRunId: runId, executionLockedAt: failedAt }) + .where(eq(issues.id, issueId)); + } + await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, phase: "retryable_failure", attempt: 1, + nextAttemptAt: new Date(Date.now() - 60_000), recoveryState: "resuming_session", + failureCode: "native_session_interrupted" }); + return { issueId, runId }; + } + + async function seedRoutineSlotHolder(routine: { id: string; fingerprint: string }) { + const issueId = randomUUID(), runId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Hourly report", status: "in_progress", + assigneeAgentId: agentId, originKind: "routine_execution", originId: routine.id, + originFingerprint: routine.fingerprint }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, issueId, status: "running", + startedAt: new Date(), contextSnapshot: { issueId } }); + await db.update(issues).set({ executionRunId: runId, executionLockedAt: new Date() }).where(eq(issues.id, issueId)); + return { issueId, runId }; + } + + async function rejectClaimsFor(runId: string) { + const name = sql.raw(`reject_recovery_claim_${runId.replaceAll("-", "")}`); + await db.execute(sql`create function ${name}() returns trigger language plpgsql as ${sql.raw(`$$ + begin + if new.run_id = '${runId}' and new.phase = 'observed' then + raise exception 'injected recovery claim failure'; + end if; + return new; + end $$`)}`); + await db.execute(sql`create trigger ${name} before update on native_run_finalizations + for each row execute function ${name}()`); + return async () => { + await db.execute(sql`drop trigger ${name} on native_run_finalizations`); + await db.execute(sql`drop function ${name}()`); + }; + } + + it("defers the routine collision and the failing run, claims the healthy run, and completes", async () => { + const routine = { id: randomUUID(), fingerprint: "scheduled" }; + const older = await seedScheduledResume({ lock: "none", routine }); + const newer = await seedRoutineSlotHolder(routine); + const failing = await seedScheduledResume({ lock: "self" }); + const healthy = await seedScheduledResume({ lock: "self" }); + const coordinator = async (runId: string) => + (await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)))[0]!; + const task = async (issueId: string) => (await db.select().from(issues).where(eq(issues.id, issueId)))[0]!; + const restore = await rejectClaimsFor(failing.runId); + const heartbeat = heartbeatService(db); + try { + const now = new Date(); + const dispositions = await claimNativeRestartRecoveries({ db, restartKind: "hard", now, + runIds: [older.runId, failing.runId, healthy.runId] }); + const byRun = new Map(dispositions.map(disposition => [disposition.runId, disposition])); + expect(byRun.get(older.runId)).toEqual({ kind: "awaiting_evidence", runId: older.runId, + reason: "routine_execution_busy", retryAt: new Date(now.getTime() + 30_000) }); + expect(byRun.get(failing.runId)).toEqual({ kind: "awaiting_evidence", runId: failing.runId, + reason: NATIVE_RESTART_RECOVERY_FAILED, retryAt: new Date(now.getTime() + 30_000) }); + expect(byRun.get(healthy.runId)).toMatchObject({ kind: "reattach_remote_runner", runId: healthy.runId }); + + // The losing transactions rolled back: no owner, lock, or attempt changed. + expect(await coordinator(older.runId)).toMatchObject({ phase: "retryable_failure", attempt: 1, leaseOwner: null }); + expect(await coordinator(failing.runId)).toMatchObject({ phase: "retryable_failure", attempt: 1, leaseOwner: null }); + expect(await task(older.issueId)).toMatchObject({ executionRunId: null }); + expect(await task(newer.issueId)).toMatchObject({ executionRunId: newer.runId }); + expect(await task(failing.issueId)).toMatchObject({ executionRunId: failing.runId }); + + // The startup entry point completes and records each deferral on its run. + await expect(heartbeat.recoverNativeRunsAfterRestart(now, [older.runId, failing.runId])) + .resolves.toMatchObject({ claims: [], awaitingEvidenceRunIds: expect.arrayContaining([older.runId, failing.runId]) }); + const events = await db.select({ runId: heartbeatRunEvents.runId, level: heartbeatRunEvents.level, + payload: heartbeatRunEvents.payload }).from(heartbeatRunEvents).where(and( + inArray(heartbeatRunEvents.runId, [older.runId, failing.runId]), + eq(heartbeatRunEvents.eventType, "native.recovery.transition"))); + expect(events).toEqual(expect.arrayContaining([ + expect.objectContaining({ runId: older.runId, level: "info", + payload: expect.objectContaining({ reason: "routine_execution_busy" }) }), + expect.objectContaining({ runId: failing.runId, level: "warn", + payload: expect.objectContaining({ reason: NATIVE_RESTART_RECOVERY_FAILED }) }), + ])); + expect(await coordinator(older.runId)).toMatchObject({ phase: "retryable_failure", attempt: 1 }); + expect(await task(newer.issueId)).toMatchObject({ executionRunId: newer.runId }); + } finally { + await heartbeat.drainActiveRunExecutions(); + await restore(); + } + }); +}); diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index 25613267f68..ba6df0e4e02 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -35,6 +35,7 @@ import { readProcessStartedAt } from "../hot-restart.js"; import { getServerProcessStartedAt } from "../../server-info.js"; import { redactSensitiveText } from "../../redaction.js"; import { isUniqueViolation } from "../../db-errors.js"; +import { logger } from "../../middleware/logger.js"; import { issueRecoveryActionService } from "../issue-recovery-actions.js"; import { isNewTerminalRunFailure, reportRunFailure } from "../run-failure-report.js"; import { isNativeRunnerOwnershipHeld } from "./native-runner-ownership.js"; @@ -52,6 +53,9 @@ export type NativeControllerIdentity = { export type NativeRestartKind = "hot" | "hard" | "graceful"; +/** A recovery claim for one run failed unexpectedly and was deferred. */ +export const NATIVE_RESTART_RECOVERY_FAILED = "restart_recovery_failed"; + export type NativeRestartRecoveryClaim = | { kind: "reattach_existing_runner"; @@ -1238,11 +1242,19 @@ export async function claimNativeRestartRecoveries(input: { ...common, } satisfies NativeRestartRecoveryClaim; }).catch((error: unknown) => { - if (!isUniqueViolation(error, "issues_open_routine_execution_uq")) throw error; + // The whole candidate transaction rolled back, so this run's state is + // unchanged and nothing was claimed. Keep the obligation on the existing + // recovery deadline and continue the batch: one run must not stop + // startup recovery. terminalRunToReport = null; claimScope = undefined; + const routineBusy = isUniqueViolation(error, "issues_open_routine_execution_uq"); + if (!routineBusy) { + logger.error({ err: error, runId: candidate.runId }, "native restart recovery failed for one run; deferring it"); + } return { kind: "awaiting_evidence" as const, runId: candidate.runId, - reason: "routine_execution_busy", retryAt: new Date(now.getTime() + 30_000) }; + reason: routineBusy ? "routine_execution_busy" : NATIVE_RESTART_RECOVERY_FAILED, + retryAt: new Date(now.getTime() + 30_000) }; }); if (terminalRunToReport) void reportRunFailure(input.db, terminalRunToReport); diff --git a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts index 4083815d215..b7668e5cd08 100644 --- a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts +++ b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts @@ -82,7 +82,7 @@ vi.mock("../../sentry.js", async () => { captureRunFailure: mockCaptureRunFailure, }; }); -import { claimNativeRestartRecoveries, currentNativeControllerIdentity, type NativeControllerIdentity } from "./native-restart-recovery.js"; +import { claimNativeRestartRecoveries, currentNativeControllerIdentity, NATIVE_RESTART_RECOVERY_FAILED, type NativeControllerIdentity } from "./native-restart-recovery.js"; // Exact v5 wire prompt from before the task-monitor upgrade (#15446). const historicalPrompt = { @@ -228,7 +228,7 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( eq(heartbeatRunEvents.eventType, "native.recovery.execution_reclaimed")))).toHaveLength(1); }); - it("does not turn another database constraint failure into a routine wait", async () => { + it("defers another database constraint failure without calling it a routine wait", async () => { const fixture = await seedRun("UNRELATED-DATABASE-FAILURE"); const db = new Proxy(fixture.db, { get(target, key) { if (key === "transaction") return () => target.transaction(async tx => { @@ -241,7 +241,8 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( } }); const onClaim = vi.fn(); await expect(claimNativeRestartRecoveries({ db, controller: successor, - restartKind: "hard", runIds: [fixture.runId], onClaim })).rejects.toThrow(/recovery_constraint_control/); + restartKind: "hard", runIds: [fixture.runId], onClaim })).resolves.toEqual([{ kind: "awaiting_evidence", + runId: fixture.runId, reason: NATIVE_RESTART_RECOVERY_FAILED, retryAt: expect.any(Date) }]); expect(onClaim).not.toHaveBeenCalled(); const [coordinator] = await fixture.db.select().from(nativeRunFinalizations) .where(eq(nativeRunFinalizations.runId, fixture.runId));