diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index b788e06ff84..952d74c857e 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -3829,6 +3829,16 @@ input does not keep the worker active; a committed change starts a fresh pass once the current pass reaches its tail. A pass is never restarted midway, so on a busy instance, where changes commit continuously, every candidate is still visited: the due re-arms and fallbacks of older exhausted holds included. +Discovery is instance-wide and draws from separate lanes, each paged by its own +cursor: unfinished saves, exhausted holds whose re-arm or fallback is due, and +this pass. Each lane with work takes one free slot before any lane takes a +second, and the due lane and the pass alternate when unfinished work is +present. A save that another server owns, or hundreds of holds whose saved +input waits, therefore cannot starve a due hold: it is reached within a few +runs of the worker. The due lane wraps at its tail, so a hold that becomes due +by time alone (a lease expiry or a re-arm backoff) is reached even after the +pass completed with no committed change. Lane positions are kept in memory; a +restart starts each lane from its first candidate, with no migration. Shutdown joins this worker's exports before closing the database. Repeated Retry during physical preservation binds the canonical intent without changing the attempt's generation or revision. Progress writes preserve diff --git a/server/src/services/workspace-recovery-passive-fairness.test.ts b/server/src/services/workspace-recovery-passive-fairness.test.ts new file mode 100644 index 00000000000..1a11fe87761 --- /dev/null +++ b/server/src/services/workspace-recovery-passive-fairness.test.ts @@ -0,0 +1,155 @@ +import { randomUUID } from "node:crypto"; +import { eq, sql } from "drizzle-orm"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { agentWakeupRequests, createDb, issueRecoveryActions, nativeRunFinalizations } from "@paperclipai/db"; +import type { AdapterSandboxExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; +import { nativeWorkspaceFinalizationLockKey } from "./native-runtime/native-workspace-finalization-ownership.js"; +import { readSystemRecovery } from "./system-recovery-state.js"; +import { adoptSystemRecovery } from "./system-recovery.js"; +import { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { registerWorkspaceRecoveryWork } from "./workspace-recovery-work.js"; +import { reconcileRetainedWorkspaces, retainedWorkspaceCandidates, type PassiveSweep } from "./workspace-recovery.js"; +import { acceptedRemoteExport, action, connectionString, db, run, seed, useWorkspaceRecoveryFixture } from "./__tests__/workspace-recovery-fixture.js"; + +const targets = vi.hoisted(() => new Map()); +vi.mock("./execution-control-deadline.js", async importOriginal => ({ + ...await importOriginal(), EXECUTION_RECONCILIATION_INTERVAL_MS: 200 })); +vi.mock("./environment-execution-target.js", async importOriginal => { + const actual = await importOriginal(); + return { ...actual, resolveEnvironmentExecutionTarget: async (input: Parameters[0]) => + (input.leaseId ? targets.get(input.leaseId) : undefined) ?? actual.resolveEnvironmentExecutionTarget(input) }; +}); + +/** + * Staging (Nov 11): 36 minutes after #20680 shipped, the + * worker had still re-armed none of the company's 29 exhausted holds, not even + * one whose id sorts in the first percent. It did run manual Retries, which + * its unfinished lane selects. Discovery is instance-wide: unfinished work + * had strict priority over every obligation, and the obligation pass mixed due + * holds with hundreds of settled holds whose saved user input waits, two per + * run. Either kind of load kept the due holds from being reached. + */ +describe("the workspace recovery worker reaches every due hold whatever its other load", () => { + useWorkspaceRecoveryFixture(targets); + beforeEach(() => { + vi.stubEnv("DATABASE_URL", connectionString); + vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "63".repeat(32)); + }); + afterEach(() => { vi.unstubAllEnvs(); }); + + /** An export hold of Nov 4, exhausted waiting for stop proof: the exact evidence that release saved. */ + async function stopProofHold(id: string) { + const f = await acceptedRemoteExport("retained"); + const firstFailureAt = Date.parse("2026-11-04T23:58:14Z"); + await db.update(issueRecoveryActions).set({ id, ownerType: "system", evidence: { ...(await action(f.actionId)).evidence, + systemRecovery: { ...readSystemRecovery((await action(f.actionId)).evidence)!, revision: 3, phase: "verify_execution", state: "exhausted", + attempts: 1, retriesUsed: 0, attemptId: randomUUID(), firstFailureAt, deadlineAt: firstFailureAt + 30 * 60_000, nextAttemptAt: null, + failure: { code: "agent_stop_unconfirmed", retryClass: "recheck_evidence", evidenceFingerprint: "9615".repeat(16) } }, + workspaceRecovery: { schema: "paperclip.workspace-recovery.v1", runId: f.runId, status: "blocked", stage: "verifying_stop", + attempt: 1, attemptLimit: 6, retryable: true, requestIds: [] } } }).where(eq(issueRecoveryActions.id, f.actionId)); + return { ...f, actionId: id }; + } + + /** A board-owned save of Nov 10, exhausted after 6 attempts under capture policy 1 (no version saved). */ + async function capturePolicyHold(id: string) { + const f = await acceptedRemoteExport("retained"); + const prior = await action(f.actionId); + const { capturePolicyVersion: _omitted, ...saved } = readSystemRecovery(prior.evidence)! as ReturnType & { capturePolicyVersion?: number }; + const firstFailureAt = Date.parse("2026-11-10T13:58:11Z"); + await db.update(issueRecoveryActions).set({ id, ownerType: "board", evidence: { ...prior.evidence, + systemRecovery: { ...saved, revision: 18, phase: "preserve_files", state: "exhausted", attempts: 6, retriesUsed: 5, + attemptId: randomUUID(), firstFailureAt, deadlineAt: firstFailureAt + 30 * 60_000, nextAttemptAt: null, + failure: { code: "export_interrupted", retryClass: "retry_transient", evidenceFingerprint: "9483".repeat(16), + cause: { code: null, label: "provider_error", status: null } } }, + workspaceRecovery: { schema: "paperclip.workspace-recovery.v1", runId: f.runId, status: "blocked", stage: "restoring_files", + attempt: 6, attemptLimit: 6, retryable: true, requestIds: [] } } }).where(eq(issueRecoveryActions.id, f.actionId)); + return { ...f, actionId: id }; + } + + /** A settled no-replay hold (active or resolved-but-blocked) whose task has saved user input waiting: selectable on every pass. */ + async function waitingSettledHold(id: string, status: "active" | "resolved") { + const f = await seed("native", true); + await adoptSystemRecovery(db, f); + const prior = await action(f.actionId); + await db.update(issueRecoveryActions).set({ id, status, outcome: status === "resolved" ? "blocked" : null, ownerType: "system", + evidence: { ...prior.evidence, + systemRecovery: { ...readSystemRecovery(prior.evidence)!, state: "complete", phase: "reconcile_effects", attempts: 1 }, + workspaceRecovery: { schema: "paperclip.workspace-recovery.v1", runId: f.runId, status: "recovered", stage: "saving_receipts", + attempt: 1, attemptLimit: 6, retryable: false, requestIds: [] }, + automaticRecovery: { runId: f.runId, policy: "preserve_without_replay_v1", replay: "blocked" } } }) + .where(eq(issueRecoveryActions.id, f.actionId)); + await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: (await run(f.runId)).agentId, source: "on_demand", + reason: "issue_commented", status: "deferred_issue_execution", requestedByActorType: "user", requestedByActorId: "board", + payload: { issueId: f.issueId } }); + return f; + } + + const rearmed = async (id: string) => { + const evidence = (await action(id)).evidence; + return Boolean(evidence.stopProofRecovery || evidence.capturePolicyRecovery); + }; + + it("re-arms both staging holds while a save that another server owns stays unfinished", async () => { + // A scheduled save whose run another server is finalizing: every attempt + // here finds the run's finalization lock busy and leaves it scheduled. + const busy = await seed("native", true); + await adoptSystemRecovery(db, busy); + expect(readSystemRecovery((await action(busy.actionId)).evidence)?.state).toBe("scheduled"); + const owner = createDb(connectionString); + await owner.execute(sql`select pg_advisory_lock(hashtextextended(${nativeWorkspaceFinalizationLockKey(busy.companyId, busy.runId)}, 0))`); + const capture = await capturePolicyHold(randomUUID()); + const stop = await stopProofHold(randomUUID()); + const onError = vi.fn(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, canReconcile: () => true, onError }); + const worker = registerWorkspaceRecoveryWork(db, coordinator, stop.environmentRuntime, async () => undefined); + try { + await worker.ready; + await vi.waitFor(async () => { + expect(await rearmed(capture.actionId)).toBe(true); + expect(await rearmed(stop.actionId)).toBe(true); + }, { timeout: 20_000, interval: 250 }); + // The busy save is still retried; it is not dropped to make room. + expect(readSystemRecovery((await action(busy.actionId)).evidence)?.state).toBe("scheduled"); + expect(onError).not.toHaveBeenCalled(); + } finally { + await worker.stop(); + await coordinator.stop(); + await owner.$client.end(); + } + }, 90_000); + + it("reaches due holds within two runs however many settled holds with waiting input sort before them", async () => { + for (let index = 1; index <= 10; index += 1) + await waitingSettledHold(`00000000-0000-4000-8000-${String(index).padStart(12, "0")}`, index % 3 ? "resolved" : "active"); + const capture = await capturePolicyHold("fffffffe-0000-4000-8000-000000000000"); + const stop = await stopProofHold("ffffffff-0000-4000-8000-000000000000"); + const passive = (await retainedWorkspaceCandidates(db, undefined, false, true)).map(row => row.action.id); + expect(passive.slice(0, 5).every(id => id.startsWith("00000000-"))).toBe(true); + + const sweep: PassiveSweep = {}; + for (let pass = 0; pass < 2; pass += 1) + await Promise.all(await reconcileRetainedWorkspaces(db, stop.environmentRuntime, async () => undefined, true, undefined, sweep) ?? []); + expect(await rearmed(capture.actionId)).toBe(true); + expect(await rearmed(stop.actionId)).toBe(true); + // The waiting input is still visited: the pass keeps advancing. + expect(sweep.cursor).toMatch(/^00000000-/); + }, 90_000); + + it("re-arms a hold that becomes due after its pass completed, with no committed change", async () => { + const stop = await stopProofHold("00000000-0000-4000-8000-000000000001"); + await waitingSettledHold("ffffffff-0000-4000-8000-000000000000", "resolved"); + // The finished coordinator's owner holds a live lease: not due yet. + await db.update(nativeRunFinalizations).set({ leaseOwner: `native-finalizer:${randomUUID()}`, leaseExpiresAt: new Date(Date.now() + 60_000) }) + .where(eq(nativeRunFinalizations.runId, stop.runId)); + const sweep: PassiveSweep = {}; + for (let pass = 0; pass < 3 && !sweep.complete; pass += 1) + await Promise.all(await reconcileRetainedWorkspaces(db, stop.environmentRuntime, async () => undefined, true, undefined, sweep) ?? []); + expect(sweep.complete).toBe(true); + expect(await rearmed(stop.actionId)).toBe(false); + + // The lease lapses. Time alone makes the re-arm due: nothing commits. + await db.update(nativeRunFinalizations).set({ leaseExpiresAt: new Date(Date.now() - 1_000) }).where(eq(nativeRunFinalizations.runId, stop.runId)); + await Promise.all(await reconcileRetainedWorkspaces(db, stop.environmentRuntime, async () => undefined, true, undefined, sweep) ?? []); + expect(await rearmed(stop.actionId)).toBe(true); + }, 90_000); +}); diff --git a/server/src/services/workspace-recovery.ts b/server/src/services/workspace-recovery.ts index 6c8712a031a..3b2264cbde2 100644 --- a/server/src/services/workspace-recovery.ts +++ b/server/src/services/workspace-recovery.ts @@ -587,7 +587,11 @@ export async function scheduleRecoveredWorkspaceCleanup(db: Db, run: typeof hear sql`${environmentLeases.metadata} = ${JSON.stringify(lease.metadata)}::jsonb`, sql`${environmentLeases.metadata}->>'workspaceRecoveryCleanup' is null`)), { companyId: run.companyId, runId: run.id }); } -export function retainedWorkspaceCandidates(db: Db, cursor?: string, unfinishedOnly = false, obligationsOnly = false, inFlightRunIds: string[] = []) { +/** `dueOnly` narrows the obligations to exhausted holds whose re-arm or + * fallback is due now. Cleanup, interrupted preparation and saved input that + * waits stay with the obligation pass. */ +export function retainedWorkspaceCandidates(db: Db, cursor?: string, unfinishedOnly = false, obligationsOnly = false, inFlightRunIds: string[] = [], + dueOnly = false) { const cleanup = sql`(((${heartbeatRuns.runnerProfileJson}#>>'{nativeWorkspaceSync,resourceDisposition}' = 'destroy' and ${heartbeatRuns.runnerProfileJson}#>>'{nativeWorkspaceSync,leaseId}' is not null) or (${heartbeatRuns.runnerProfileJson}->>'nativeResourceDisposition' = 'destroy' @@ -620,6 +624,8 @@ export function retainedWorkspaceCandidates(db: Db, cursor?: string, unfinishedO and ${issues.executionRunId} is null and ${issues.checkoutRunId} is null and (${issues.status} <> 'blocked' or ${issueRecoveryActions.evidence}#>>'{nativeBootstrapFailureBlock,statusVersion}' = ${issues.statusVersion}::text)))`; + const now = Date.now(); + const dueHold = sql`(${retainedSourceFallbackDueCondition(now)} or ${stopProofRearmDueCondition(now)} or ${capturePolicyRearmDueCondition(now)})`; return db.select({ action: issueRecoveryActions, run: heartbeatRuns }).from(issueRecoveryActions) .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.companyId, issueRecoveryActions.companyId), eq(heartbeatRuns.id, sql`case when ${issueRecoveryActions.evidence}->>'runId' ~* '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' @@ -632,11 +638,11 @@ export function retainedWorkspaceCandidates(db: Db, cursor?: string, unfinishedO and coalesce(${issueRecoveryActions.evidence}->>'bootstrapPreparationContinuation', 'pending') = 'pending'))` : undefined, obligationsOnly ? sql`(${issueRecoveryActions.evidence}#>>'{systemRecovery,state}' in ('scheduled', 'running', 'awaiting_evidence') or (${issueRecoveryActions.evidence}#>>'{systemRecovery,state}' = 'complete' and (${cleanup} or ${interruptedPreparation})) or ${waitingInput} - or ${retainedSourceFallbackDueCondition(Date.now())} or ${stopProofRearmDueCondition(Date.now())} - or ${capturePolicyRearmDueCondition(Date.now())})` : undefined, + or ${dueHold})` : undefined, + dueOnly ? dueHold : undefined, sql`(${issueRecoveryActions.evidence}#>>'{systemRecovery,state}' in ('scheduled', 'running', 'awaiting_evidence', 'complete', 'exhausted') and (${issueRecoveryActions.evidence}#>'{systemRecovery,nextAttemptAt}' = 'null'::jsonb - or ${issueRecoveryActions.evidence}#>'{systemRecovery,nextAttemptAt}' <= to_jsonb(${Date.now()}::bigint)))`, + or ${issueRecoveryActions.evidence}#>'{systemRecovery,nextAttemptAt}' <= to_jsonb(${now}::bigint)))`, sql`(${issueRecoveryActions.evidence}#>>'{workspaceRecovery,status}' is distinct from 'recovered' or ${cleanup} or ${waitingInput} or ${interruptedPreparation})`, cursor ? gt(issueRecoveryActions.id, cursor) : undefined, @@ -644,10 +650,26 @@ export function retainedWorkspaceCandidates(db: Db, cursor?: string, unfinishedO )).orderBy(asc(issueRecoveryActions.id)).limit(5); } -export type PassiveSweep = { cursor?: string; complete?: boolean; restart?: boolean }; +/** + * The worker's lane positions, kept in memory: a restart starts every lane + * from its first candidate, so no persisted state or migration is involved. + * `cursor`, `complete` and `restart` belong to the obligation pass; `dueCursor` + * to the due lane; `duePreferred` alternates the second slot between them. + */ +export type PassiveSweep = { cursor?: string; complete?: boolean; restart?: boolean; dueCursor?: string; duePreferred?: boolean }; + +type CandidateRow = Awaited>[number]; +type CandidateLane = { rows: CandidateRow[]; index: number; advance: (actionId: string) => void }; /** Bounded discovery shares the periodic recovery lifecycle. A repaired source - * still owes cleanup even without saved input or after task reconciliation. */ + * still owes cleanup even without saved input or after task reconciliation. + * + * Candidates come from lanes, each an ordered page that a cursor advances: + * unfinished saves; for the worker, exhausted holds whose re-arm or fallback + * is due; and the obligation pass (cleanup, interrupted preparation, saved + * input that waits, and the same due holds). Each lane with + * work takes one free slot before any lane takes a second, so no kind of load + * (a save another server owns, hundreds of waiting holds) starves another. */ export async function reconcileRetainedWorkspaces(db: Db, runtime: EnvironmentRuntimeService, onRecovered: (run: typeof heartbeatRuns.$inferSelect) => Promise, obligationsOnly = false, onFinished?: (continueSweep: boolean) => void, passiveSweep?: PassiveSweep) { @@ -658,39 +680,63 @@ export async function reconcileRetainedWorkspaces(db: Db, runtime: EnvironmentRu // Locally running exports already own their slot. Exclude them before // pagination so a page containing only held work cannot prevent wrapping. const inFlightRunIds = [...pending]; - let rows = await retainedWorkspaceCandidates(db, unfinishedCursors.get(db), true, false, inFlightRunIds); - if (!rows.length) rows = await retainedWorkspaceCandidates(db, undefined, true, false, inFlightRunIds); - const unfinished = rows.length > 0; - if (!rows.length) { - rows = await retainedWorkspaceCandidates(db, passiveSweep ? passiveSweep.cursor : cursors.get(db), false, obligationsOnly, inFlightRunIds); - if (passiveSweep && !rows.length && passiveSweep.restart) { - Object.assign(passiveSweep, { cursor: undefined, restart: false }); - rows = await retainedWorkspaceCandidates(db, undefined, false, obligationsOnly, inFlightRunIds); - } - if (passiveSweep) passiveSweep.complete = !rows.length; + let unfinishedRows = await retainedWorkspaceCandidates(db, unfinishedCursors.get(db), true, false, inFlightRunIds); + if (!unfinishedRows.length) unfinishedRows = await retainedWorkspaceCandidates(db, undefined, true, false, inFlightRunIds); + const unfinished: CandidateLane = { rows: unfinishedRows, index: 0, advance: id => unfinishedCursors.set(db, id) }; + let rows = await retainedWorkspaceCandidates(db, passiveSweep ? passiveSweep.cursor : cursors.get(db), false, obligationsOnly, inFlightRunIds); + if (passiveSweep && !rows.length && passiveSweep.restart) { + Object.assign(passiveSweep, { cursor: undefined, restart: false }); + rows = await retainedWorkspaceCandidates(db, undefined, false, obligationsOnly, inFlightRunIds); } - if (!rows.length && !passiveSweep) rows = await retainedWorkspaceCandidates(db, undefined, false, obligationsOnly, inFlightRunIds); + if (passiveSweep) passiveSweep.complete = !rows.length; + else if (!rows.length) rows = await retainedWorkspaceCandidates(db, undefined, false, obligationsOnly, inFlightRunIds); + const sweep: CandidateLane = { rows, index: 0, + advance: id => { if (passiveSweep) passiveSweep.cursor = id; else cursors.set(db, id); } }; + const lanes = [unfinished]; + if (passiveSweep) { + // The due lane wraps at its tail like the unfinished lane. A hold that + // becomes due by time alone (a lease expiry, a re-arm backoff) is reached + // even after the obligation pass completed with no committed change. + let dueRows = await retainedWorkspaceCandidates(db, passiveSweep.dueCursor, false, true, inFlightRunIds, true); + if (!dueRows.length && passiveSweep.dueCursor) + dueRows = await retainedWorkspaceCandidates(db, undefined, false, true, inFlightRunIds, true); + const due: CandidateLane = { rows: dueRows, index: 0, advance: id => { passiveSweep.dueCursor = id; } }; + lanes.push(...(passiveSweep.duePreferred === false ? [sweep, due] : [due, sweep])); + passiveSweep.duePreferred = passiveSweep.duePreferred === false; + } else lanes.push(sweep); const work: Promise[] = []; - let hasUnvisited = false; - const advance = (actionId: string) => { - if (unfinished) unfinishedCursors.set(db, actionId); - else if (passiveSweep) passiveSweep.cursor = actionId; - else cursors.set(db, actionId); - }; // Several actions of one run (an export hold and a later no-replay hold) // run one after another in that run's job. Skipping the later one would // starve it while the earlier one stays selectable (for example while saved // input waits for the later one to settle). const groups = new Map(); - for (const { action, run } of rows) { - const group = groups.get(run.id); - if (group) { group.actions.push(action); advance(action.id); continue; } - if (pending.size >= 2) { hasUnvisited = true; break; } - advance(action.id); - if (pending.has(run.id)) continue; - pending.add(run.id); - groups.set(run.id, { run, actions: [action] }); + const join = (lane: CandidateLane) => { + for (; lane.index < lane.rows.length; lane.index += 1) { + const { action, run } = lane.rows[lane.index]!; + const group = groups.get(run.id); + if (!group) return lane.rows[lane.index]!; + if (!group.actions.some(item => item.id === action.id)) group.actions.push(action); + lane.advance(action.id); + } + return undefined; + }; + // Takes the lane's next run into a free slot. Actions of a run already taken + // (by any lane) join its job without a slot. + const take = (lane: CandidateLane) => { + const next = join(lane); + if (!next || pending.size >= 2) return false; + lane.index += 1; + lane.advance(next.action.id); + pending.add(next.run.id); + groups.set(next.run.id, { run: next.run, actions: [next.action] }); + return true; + }; + for (let taken = true; taken && pending.size < 2;) { + taken = false; + for (const lane of lanes) if (pending.size < 2 && take(lane)) taken = true; } + for (const lane of lanes) join(lane); + const hasUnvisited = lanes.some(lane => lane.index < lane.rows.length); for (const { run, actions } of groups.values()) { const finish = beginIdleTrackedWork(); const physical = actions.some(action => ["scheduled", "running", "awaiting_evidence"].includes(readSystemRecovery(action.evidence)?.state ?? ""));