mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
fix(recovery): release a lapsed owner of a finished coordinator so stop-proof holds re-arm (#20661)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Workspace recovery saves the files of a run whose copy-back failed, and it holds the shared workspace until the save is done > - #20295 re-arms a save that was exhausted with `agent_stop_unconfirmed`, so the run's stop is proven again and the files are saved without a person > - On staging, four holds of this shape were never re-armed. Their runs ended six days ago, their sandbox leases are released with a provider stop receipt, and the servers that ran them were replaced many times. Their tasks stay blocked, and so do the parent tasks that wait for them > - The finished coordinator of each run still names a lease owner whose lease lapsed. Discovery treats that owner as "waiting for the abandoned-owner settlement", but that settlement handles only observed (unfinished) coordinators. Nothing ever releases the owner, so the worker never selects the hold, the re-arm refuses it, and the attempt's stop check refuses it too > - This pull request makes a lapsed owner of a finished coordinator count as stopped. The re-arm releases it in the same transaction, and then the new attempt proves the sandbox stop and saves the files as #20295 designed > - The benefit is that these holds settle on the worker's next pass after deploy, without a migration and without an administrator ## Linked Issues or Issue Description Refs #20295 **What happened?** An exhausted workspace save (`native_workspace_sync_out_retry_exhausted`, last failure `agent_stop_unconfirmed`) stayed blocked for six days. The task showed "Paperclip tries again automatically", but no re-arm was ever recorded (no `stopProofRecovery` marker, no `issue.workspace_recovery_rearmed` activity), and the shared workspace stayed held by the failed run. **What did you expect to happen?** The recovery worker re-arms the save once the run is provably over, saves the dirty files, and frees the workspace so the task and its parent can continue. **Steps to reproduce** 1. Exhaust a workspace save with `agent_stop_unconfirmed` (one attempt, window spent). 2. Leave a lease owner on the run's `terminal_failure` coordinator with an expired lease (or no expiry), as a finalizer pass of a replaced server does. 3. Run the workspace recovery worker. `nextWorkspaceRecoveryAt` returns null, discovery does not select the hold, and `rearmExhaustedStopProof` refuses it. ## What Changed - `workspace-recovery-stop-proof.ts`: add `lapsedFinishedCoordinatorOwner` and `releaseLapsedCoordinatorOwner`. A lease owner on a finished coordinator (committed, terminal failure, applied) whose lease lapsed, or that has no expiry, does not block `sourceRunProvablyOver`. The stop-proof re-arm clears that owner with a compare-and-swap in its locked transaction, and records `releasedLapsedOwner` in the marker and the activity. - `workspace-recovery-capture-policy.ts`: the capture-policy re-arm uses the same rule, because it shares the same check and discovery clause. - `workspace-recovery-fallback-discovery.ts`: the discovery clause still waits for the abandoned-owner settlement for a stale owner of an unfinished coordinator, but it no longer excludes a hold whose finished coordinator has a lapsed owner. A live lease still defers the re-arm to its expiry. - `system-recovery-state.ts`: add `FINISHED_COORDINATOR_PHASES`. - `doc/execution-semantics.md`: document the rule. - Tests: two red-then-green regressions in the staging shape (an expired finalizer lease, and an owner without an expiry saved by an older release), one strictness case (a live lease on a finished coordinator keeps the hold and is not released), and the existing stale-owner case now uses an observed coordinator, which still waits for the abandoned-owner settlement. ## Verification - Red before the fix: `pnpm --filter @paperclipai/server exec vitest run src/services/workspace-recovery-stop-proof.test.ts` failed the two new staging-shape cases (`nextWorkspaceRecoveryAt` returned null: the worker was never woken for the hold). - Green after the fix: `workspace-recovery-stop-proof.test.ts` and `workspace-recovery-capture-policy.test.ts` pass (15 tests). The staging-shape cases assert that the hold is re-armed, the save completes, the dirty files are saved, the workspace hold is free, and the lapsed owner is released. - `pnpm verify`: exit 2 (broader verification required, because `doc/execution-semantics.md` is a global CI input). Focused suites pass: `workspace-recovery-stop-proof`, `workspace-recovery-capture-policy`, `workspace-recovery-fallback`, `workspace-recovery-retry-copyback` (113 tests). Required CI supplies the broad coverage. - Staging evidence (read-only GET requests): the run is `failed`, finished six days ago, `controllerLeaseExpiresAt` is null, its only lease is `released` with `cleanupStatus: success`, a `stopped` termination receipt and a stop-and-retain receipt. The action's evidence has no re-arm marker, and four holds of the same shape were not touched after #20295 was deployed. ## Risks - Previous-release story: rows saved by earlier releases carry the lapsed owner and no marker. They are selected and re-armed on the worker's first pass after deploy. No migration. - Strictness stays where an owner could be live: a live lease on any coordinator defers the re-arm to its expiry, and a stale owner of an unfinished (observed) coordinator still waits for the abandoned-owner settlement, which proves its controller stopped. The new attempt still requires a provider stop receipt for the allocation before it reads files. - Clearing a lapsed owner matches the existing token-lease contract: an expired finalizer or cleanup lease is free for the next claim, and a holder's later write fails its token compare-and-swap. - Low risk otherwise: the change is limited to the two exhausted-hold re-arms and their discovery clause. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use (shell, code editing, browser GET reads) in Claude Code. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
a3b34a9d88
commit
686893636d
6 files changed
+149
-31
No files matched your search
@@ -2815,7 +2815,16 @@ export holds and restore holds alike. Once the source run is provably over (its
|
||||
run is terminal, no local execution remains, and neither its controller lease
|
||||
nor its finalizer lease is live), the recovery worker re-arms it with a new
|
||||
bounded system generation. A live lease defers the re-arm to that lease's
|
||||
expiry; the worker is not woken earlier. The new generation repeats the strict
|
||||
expiry; the worker is not woken earlier. A lease owner still recorded on the
|
||||
run's finished coordinator (committed, terminal failure or applied) after its
|
||||
lease lapsed, or recorded without an expiry by an older release, is not live:
|
||||
nothing runs under a finished coordinator, and no later claim or settlement
|
||||
would ever release it. The re-arm releases that owner in the same transaction
|
||||
(compare-and-swap on the owner it read) and records `releasedLapsedOwner`, so
|
||||
the new attempt's stop check does not refuse it; holds saved this way by
|
||||
earlier releases settle on the worker's next pass. A stale owner of an
|
||||
unfinished coordinator still waits for the abandoned-owner settlement, which
|
||||
proves its controller stopped first. The new generation repeats the strict
|
||||
stop proof (a provider stop receipt for every allocation of the run) and the
|
||||
save, then settles the hold exactly like any completed recovery, so other tasks
|
||||
on the workspace can start. Re-arms are bounded: at most four, the first at once
|
||||
|
||||
@@ -162,6 +162,8 @@ export const STOP_PROOF_RECOVERY_KEY = "stopProofRecovery";
|
||||
export const STOP_PROOF_RECOVERY_SCHEMA = "paperclip.stop-proof-recovery.v1";
|
||||
export const STOP_PROOF_MAX_REARMS = 4;
|
||||
export const STOP_PROOF_WAIT_CODE = "agent_stop_unconfirmed";
|
||||
/** Coordinator phases after which no finalizer or provider work runs under the coordinator. */
|
||||
export const FINISHED_COORDINATOR_PHASES = ["committed", "terminal_failure", "applied"] as const;
|
||||
|
||||
export function readStopProofRecovery(evidence: Record<string, unknown> | null | undefined): { rearms: number; nextRearmAt: number | null } | null {
|
||||
const marker = evidence?.[STOP_PROOF_RECOVERY_KEY] as Record<string, unknown> | undefined;
|
||||
|
||||
@@ -8,7 +8,7 @@ import { REMOTE_WORKSPACE_CAPTURE_POLICY_VERSION } from "./remote-workspace-snap
|
||||
import { CAPTURE_POLICY_RECOVERY_KEY, CAPTURE_POLICY_RECOVERY_SCHEMA, createSystemRecovery, isCapturePolicyRearmable,
|
||||
readSystemRecovery, type SystemRecoveryState } from "./system-recovery-state.js";
|
||||
import type { WorkspaceRecoverySourceIdentity } from "./workspace-recovery-source.js";
|
||||
import { sourceRunProvablyOver } from "./workspace-recovery-stop-proof.js";
|
||||
import { coordinatorOwnerColumns, releaseLapsedCoordinatorOwner, sourceRunProvablyOver } from "./workspace-recovery-stop-proof.js";
|
||||
|
||||
/** User-facing text while the retried save runs. It names no person who must act. */
|
||||
export const CAPTURE_POLICY_REARM_MESSAGE = "Paperclip is saving the previous run's files again with its updated save rules. "
|
||||
@@ -31,8 +31,7 @@ export async function rearmExhaustedCapturePolicy(db: Db, input: WorkspaceRecove
|
||||
// Lock order shared with the fallback and the stop-proof re-arm: task, coordinator, run, recovery action.
|
||||
const [issue] = await tx.select({ id: issues.id }).from(issues).where(and(eq(issues.companyId, input.companyId),
|
||||
eq(issues.id, input.issueId))).for("update");
|
||||
const [coordinator] = await tx.select({ leaseOwner: nativeRunFinalizations.leaseOwner })
|
||||
.from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, input.companyId),
|
||||
const [coordinator] = await tx.select(coordinatorOwnerColumns).from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, input.companyId),
|
||||
eq(nativeRunFinalizations.runId, input.runId))).for("update");
|
||||
const [run] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, input.companyId),
|
||||
eq(heartbeatRuns.id, input.runId))).for("update");
|
||||
@@ -48,12 +47,14 @@ export async function rearmExhaustedCapturePolicy(db: Db, input: WorkspaceRecove
|
||||
// still proves the sandbox stop itself before it reads any file.
|
||||
const at = new Date(now);
|
||||
if (!sourceRunProvablyOver(run, coordinator, at)) return null;
|
||||
// A lapsed owner of the finished coordinator would refuse the new attempt's stop check.
|
||||
const releasedLapsedOwner = await releaseLapsedCoordinatorOwner(tx, input, coordinator, at);
|
||||
const next: SystemRecoveryState = { ...createSystemRecovery(run.id), revision: system.revision + 1,
|
||||
preservationOperationIds: system.preservationOperationIds, providerCooldownAt: system.providerCooldownAt,
|
||||
nextAttemptAt: system.providerCooldownAt };
|
||||
const marker = { schema: CAPTURE_POLICY_RECOVERY_SCHEMA, policyVersion: REMOTE_WORKSPACE_CAPTURE_POLICY_VERSION,
|
||||
previousPolicyVersion: system.capturePolicyVersion ?? 1, generationId: next.generationId,
|
||||
previousGenerationId: system.generationId, rearmedAt: now };
|
||||
previousGenerationId: system.generationId, rearmedAt: now, ...(releasedLapsedOwner ? { releasedLapsedOwner: true } : {}) };
|
||||
const prior = action.evidence.workspaceRecovery as { requestIds?: unknown } | undefined;
|
||||
await notifyIssueRecovery(tx, { companyId: input.companyId });
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.workspaceRecovery);
|
||||
@@ -68,7 +69,8 @@ export async function rearmExhaustedCapturePolicy(db: Db, input: WorkspaceRecove
|
||||
action: "issue.workspace_recovery_rearmed", entityType: "issue", entityId: input.issueId, runId: run.id,
|
||||
details: { recoveryActionId: action.id, reason: "capture_policy_changed", failureCode: system.failure?.code ?? null,
|
||||
policyVersion: marker.policyVersion, previousPolicyVersion: marker.previousPolicyVersion,
|
||||
previousGenerationId: system.generationId, generationId: next.generationId } });
|
||||
previousGenerationId: system.generationId, generationId: next.generationId,
|
||||
...(releasedLapsedOwner ? { releasedLapsedOwner: coordinator!.leaseOwner } : {}) } });
|
||||
});
|
||||
if (!outcome) return false;
|
||||
try { publishActivity(outcome.publication); } catch { logger.warn({ runId: input.runId }, "Recovery activity delivery deferred"); }
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import { sql } from "drizzle-orm";
|
||||
import { issueRecoveryActions } from "@paperclipai/db";
|
||||
import { REMOTE_WORKSPACE_CAPTURE_POLICY_VERSION } from "./remote-workspace-snapshot-archive.js";
|
||||
import { CAPTURE_POLICY_REARM_CODES, isExhaustedStopProofWait, STOP_PROOF_MAX_REARMS, STOP_PROOF_RECOVERY_KEY, STOP_PROOF_RECOVERY_SCHEMA, STOP_PROOF_WAIT_CODE,
|
||||
import { CAPTURE_POLICY_REARM_CODES, FINISHED_COORDINATOR_PHASES, isExhaustedStopProofWait, STOP_PROOF_MAX_REARMS, STOP_PROOF_RECOVERY_KEY, STOP_PROOF_RECOVERY_SCHEMA, STOP_PROOF_WAIT_CODE,
|
||||
stopProofRearmsSpent, type SystemRecoveryState } from "./system-recovery-state.js";
|
||||
import { canonicalExecutionUuid } from "./execution-identity.js";
|
||||
|
||||
@@ -17,6 +17,7 @@ export const EXPORT_HOLD_CAUSES = ["native_workspace_sync_out_unrecoverable", "n
|
||||
export const NATIVE_NO_REPLAY_HOLD_CAUSE = "native_continuation_requires_reconciliation";
|
||||
|
||||
type HoldAction = { cause: string; status: string; outcome: string | null; evidence: Record<string, unknown> };
|
||||
const list = (values: readonly string[]) => sql.join(values.map(value => sql`${value}`), sql`, `);
|
||||
const field = (value: unknown, key: string) => value && typeof value === "object" ? (value as Record<string, unknown>)[key] : undefined;
|
||||
|
||||
/** Whether this recovery action is a retained-file hold the fallback may settle. */
|
||||
@@ -63,17 +64,23 @@ const exhaustedStopProofWait = sql`(${issueRecoveryActions.evidence}#>>'{systemR
|
||||
and ${issueRecoveryActions.evidence}#>>'{systemRecovery,failure,code}' = ${STOP_PROOF_WAIT_CODE}
|
||||
and ${issueRecoveryActions.evidence}#>>'{systemRecovery,failure,retryClass}' = 'recheck_evidence')`;
|
||||
|
||||
// A stale (lapsed or expiry-less) owner of an unfinished coordinator waits for
|
||||
// the abandoned-owner settlement: the attempt would refuse it, and the passive
|
||||
// sweep revisits the hold after it. The re-arm itself releases a lapsed owner
|
||||
// of a finished coordinator (lapsedFinishedCoordinatorOwner), because no
|
||||
// settlement ever would.
|
||||
const staleUnfinishedFinalizerOwner = sql`exists (select 1 from native_run_finalizations finalizer
|
||||
where finalizer.company_id = ${issueRecoveryActions.companyId} and finalizer.run_id = ${sourceRunId}
|
||||
and finalizer.lease_owner is not null and coalesce(finalizer.lease_expires_at <= now(), true)
|
||||
and finalizer.phase not in (${list(FINISHED_COORDINATOR_PHASES)}))`;
|
||||
|
||||
/** An exhausted unproven-stop hold whose next bounded re-arm is pending. Holds
|
||||
* saved by older releases carry no marker and are due immediately. A finalizer
|
||||
* owner whose lease is not live waits for the abandoned-owner settlement (the
|
||||
* attempt would refuse it, and the passive sweep revisits the hold after it);
|
||||
* a live one makes the re-arm due at its expiry (stopProofRearmDueAt). */
|
||||
* saved by older releases carry no marker and are due immediately. A live
|
||||
* finalizer lease makes the re-arm due at its expiry (stopProofRearmDueAt). */
|
||||
export function stopProofRearmPendingCondition() {
|
||||
return sql`(${issueRecoveryActions.status} in ('active', 'escalated')
|
||||
and ${exhaustedStopProofWait}
|
||||
and not exists (select 1 from native_run_finalizations finalizer
|
||||
where finalizer.company_id = ${issueRecoveryActions.companyId} and finalizer.run_id = ${sourceRunId}
|
||||
and finalizer.lease_owner is not null and coalesce(finalizer.lease_expires_at <= now(), true))
|
||||
and not ${staleUnfinishedFinalizerOwner}
|
||||
and ${issueRecoveryActions.evidence}->'retainedSourceFallback' is null
|
||||
and ${issueRecoveryActions.evidence}->'executionReconciliation' is null
|
||||
and ${stopProofRearms} < ${STOP_PROOF_MAX_REARMS})`;
|
||||
@@ -101,7 +108,6 @@ export function stopProofRearmDueCondition(now: number) {
|
||||
return sql`(${stopProofRearmPendingCondition()} and ${stopProofRearmDueAt()} <= ${now}::bigint)`;
|
||||
}
|
||||
|
||||
const list = (values: readonly string[]) => sql.join(values.map(value => sql`${value}`), sql`, `);
|
||||
const capturePolicyVersion = sql`case when jsonb_typeof(${issueRecoveryActions.evidence}#>'{systemRecovery,capturePolicyVersion}') = 'number'
|
||||
then (${issueRecoveryActions.evidence}#>>'{systemRecovery,capturePolicyVersion}')::int else 1 end`;
|
||||
|
||||
@@ -114,9 +120,7 @@ export function capturePolicyRearmPendingCondition() {
|
||||
and ${issueRecoveryActions.evidence}#>>'{systemRecovery,phase}' = 'preserve_files'
|
||||
and ${issueRecoveryActions.evidence}#>>'{systemRecovery,failure,code}' in (${list(CAPTURE_POLICY_REARM_CODES)})
|
||||
and ${capturePolicyVersion} < ${REMOTE_WORKSPACE_CAPTURE_POLICY_VERSION}
|
||||
and not exists (select 1 from native_run_finalizations finalizer
|
||||
where finalizer.company_id = ${issueRecoveryActions.companyId} and finalizer.run_id = ${sourceRunId}
|
||||
and finalizer.lease_owner is not null and coalesce(finalizer.lease_expires_at <= now(), true))
|
||||
and not ${staleUnfinishedFinalizerOwner}
|
||||
and ${issueRecoveryActions.evidence}->'retainedSourceFallback' is null
|
||||
and ${issueRecoveryActions.evidence}->'executionReconciliation' is null)`;
|
||||
}
|
||||
|
||||
@@ -214,9 +214,69 @@ describe("exhausted stop-proof holds", () => {
|
||||
expect(rearmed.ownerType).toBe("system");
|
||||
}, 60_000);
|
||||
|
||||
it("leaves a stale finalizer owner to the abandoned-owner settlement instead of spending re-arms", async () => {
|
||||
/**
|
||||
* The staging shape (Nov 10): the finished coordinator of the
|
||||
* source run still names the lease owner of a finalizer pass from a server
|
||||
* that was replaced days ago. Its lease lapsed long ago, nothing settles an
|
||||
* owner on a finished coordinator, and every stop check refused it, so the
|
||||
* hold was never re-armed and its task stayed blocked.
|
||||
*/
|
||||
it.each([
|
||||
["an expired finalizer lease", { leaseOwner: `native-finalizer:${randomUUID()}`, expiresAgoMs: 6 * 86_400_000 }],
|
||||
["an owner recorded without an expiry by an older release", { leaseOwner: `boot:${randomUUID()}`, expiresAgoMs: null }],
|
||||
] as const)("re-arms a hold whose finished coordinator still names %s", async (_shape, owner) => {
|
||||
const f = await exhaustedStopProofHold();
|
||||
await db.update(nativeRunFinalizations).set({ leaseOwner: `native-finalizer:${randomUUID()}`, leaseExpiresAt: new Date(Date.now() - 60_000) })
|
||||
await db.update(nativeRunFinalizations).set({ leaseOwner: owner.leaseOwner,
|
||||
leaseExpiresAt: owner.expiresAgoMs === null ? null : new Date(Date.now() - owner.expiresAgoMs),
|
||||
controllerBootId: randomUUID(), controllerPid: 2_147_000_000, controllerGeneration: 3 })
|
||||
.where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0])
|
||||
.toMatchObject({ phase: "terminal_failure", failureCode: "native_workspace_sync_out_retry_exhausted" });
|
||||
|
||||
// The worker discovers the hold and wakes for it now.
|
||||
expect(await nextWorkspaceRecoveryAt(db)).toBeLessThanOrEqual(Date.now());
|
||||
expect((await retainedWorkspaceCandidates(db, undefined, false, true)).map(row => row.action.id)).toContain(f.actionId);
|
||||
await Promise.all(await reconcileRetainedWorkspaces(db, f.environmentRuntime, async () => undefined, true) ?? []);
|
||||
|
||||
const settled = await action(f.actionId);
|
||||
expect(settled.evidence.stopProofRecovery).toMatchObject({ rearms: 1, releasedLapsedOwner: true });
|
||||
expect(readSystemRecovery(settled.evidence)).toMatchObject({ state: "complete" });
|
||||
expect(readWorkspaceRecovery(settled.evidence)?.status).toBe("recovered");
|
||||
expect(String(settled.nextAction)).not.toMatch(adminText);
|
||||
// The lapsed owner is released (the saved result then finalizes as usual).
|
||||
expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]?.leaseOwner)
|
||||
.not.toBe(owner.leaseOwner);
|
||||
const [rearmed] = await db.select().from(activityLog).where(eq(activityLog.entityId, f.issueId))
|
||||
.then(rows => rows.filter(row => row.action === "issue.workspace_recovery_rearmed"));
|
||||
expect(rearmed?.details).toMatchObject({ rearm: 1, releasedLapsedOwner: owner.leaseOwner });
|
||||
// The dirty files are saved into the shared workspace; the parent task can run.
|
||||
expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toMatchObject({ dirty: false });
|
||||
expect(await parentHold(f)).toEqual({ kind: "free" });
|
||||
expect(await readFile(path.join(f.remote, "saved.txt"), "utf8")).toBe("finished work\n");
|
||||
}, 60_000);
|
||||
|
||||
it("keeps the hold while a finished coordinator's owner still holds a live lease", async () => {
|
||||
const f = await exhaustedStopProofHold();
|
||||
const expiresAt = Date.now() + 10 * 60_000;
|
||||
await db.update(nativeRunFinalizations).set({ leaseOwner: `native-finalizer:${randomUUID()}`, leaseExpiresAt: new Date(expiresAt) })
|
||||
.where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
|
||||
expect((await retainedWorkspaceCandidates(db, undefined, false, true)).map(row => row.action.id)).not.toContain(f.actionId);
|
||||
expect(await nextWorkspaceRecoveryAt(db)).toBeGreaterThanOrEqual(expiresAt - 1_000);
|
||||
await recoverRetainedWorkspace(f.input);
|
||||
const held = await action(f.actionId);
|
||||
expect(held.evidence.stopProofRecovery).toBeUndefined();
|
||||
expect(readSystemRecovery(held.evidence)?.state).toBe("exhausted");
|
||||
expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]?.leaseOwner)
|
||||
.toMatch(/^native-finalizer:/);
|
||||
expect(f.runtime.resumeRunLease).not.toHaveBeenCalled();
|
||||
}, 60_000);
|
||||
|
||||
it("leaves a stale owner of an unfinished coordinator to the abandoned-owner settlement instead of spending re-arms", async () => {
|
||||
const f = await exhaustedStopProofHold();
|
||||
// An observed (not finished) coordinator: the abandoned-owner settlement
|
||||
// must prove its controller stopped before anything releases it.
|
||||
await db.update(nativeRunFinalizations).set({ phase: "observed", leaseOwner: `boot:${randomUUID()}`, leaseExpiresAt: new Date(Date.now() - 60_000) })
|
||||
.where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
|
||||
expect((await retainedWorkspaceCandidates(db, undefined, false, true)).map(row => row.action.id)).not.toContain(f.actionId);
|
||||
|
||||
@@ -1,26 +1,64 @@
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { and, eq, inArray } from "drizzle-orm";
|
||||
import { heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, type Db } from "@paperclipai/db";
|
||||
import { logger } from "../middleware/logger.js";
|
||||
import { adapterExecutionControls } from "./adapter-execution-control.js";
|
||||
import { persistActivity, publishActivity } from "./activity-log.js";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
|
||||
import { createSystemRecovery, isExhaustedStopProofWait, readStopProofRecovery, readSystemRecovery, STOP_PROOF_MAX_REARMS,
|
||||
import { createSystemRecovery, FINISHED_COORDINATOR_PHASES, isExhaustedStopProofWait, readStopProofRecovery, readSystemRecovery, STOP_PROOF_MAX_REARMS,
|
||||
STOP_PROOF_RECOVERY_KEY, STOP_PROOF_RECOVERY_SCHEMA, type SystemRecoveryState } from "./system-recovery-state.js";
|
||||
import { terminalWorkspaceRunStates, type WorkspaceRecoverySourceIdentity } from "./workspace-recovery-source.js";
|
||||
|
||||
type Coordinator = { leaseOwner: string | null; leaseExpiresAt?: Date | null; phase?: string | null };
|
||||
|
||||
/**
|
||||
* A lease owner still recorded on a finished coordinator whose lease has
|
||||
* lapsed (or that an older release recorded without an expiry). Nothing runs
|
||||
* under a finished coordinator, and a lapsed token lease is free for the next
|
||||
* claim, but no later claim comes for a finished run and no settlement
|
||||
* releases it (the abandoned-owner settlement covers observed coordinators
|
||||
* only). Left in place it refuses every stop check of the run for good. A live
|
||||
* lease, or any owner of an unfinished coordinator, is never lapsed.
|
||||
*/
|
||||
export function lapsedFinishedCoordinatorOwner(coordinator: Coordinator | undefined, at: Date) {
|
||||
return Boolean(coordinator?.leaseOwner) && (FINISHED_COORDINATOR_PHASES as readonly string[]).includes(coordinator!.phase ?? "")
|
||||
&& (!coordinator!.leaseExpiresAt || coordinator!.leaseExpiresAt <= at);
|
||||
}
|
||||
|
||||
/**
|
||||
* The source run ended and nothing still executes or finalizes it, so a new
|
||||
* recovery generation may start. Shared by every exhausted-hold re-arm. The
|
||||
* attempt refuses any recorded finalizer owner (inspectWorkspaceRecoverySource);
|
||||
* a stale one is released by the abandoned-owner settlement first.
|
||||
* attempt refuses any recorded finalizer owner (inspectWorkspaceRecoverySource):
|
||||
* the re-arm releases a lapsed owner of a finished coordinator first
|
||||
* (releaseLapsedCoordinatorOwner), and the abandoned-owner settlement releases
|
||||
* a stale owner of an unfinished one.
|
||||
*/
|
||||
export function sourceRunProvablyOver(run: typeof heartbeatRuns.$inferSelect,
|
||||
coordinator: { leaseOwner: string | null } | undefined, at: Date) {
|
||||
export function sourceRunProvablyOver(run: typeof heartbeatRuns.$inferSelect, coordinator: Coordinator | undefined, at: Date) {
|
||||
return terminalWorkspaceRunStates.includes(run.status) && Boolean(run.finishedAt) && !adapterExecutionControls.has(run.id)
|
||||
&& !(run.controllerLeaseExpiresAt && run.controllerLeaseExpiresAt > at) && !coordinator?.leaseOwner;
|
||||
&& !(run.controllerLeaseExpiresAt && run.controllerLeaseExpiresAt > at)
|
||||
&& (!coordinator?.leaseOwner || lapsedFinishedCoordinatorOwner(coordinator, at));
|
||||
}
|
||||
|
||||
/**
|
||||
* Clears a lapsed owner of a finished coordinator, inside the re-arm
|
||||
* transaction that holds the coordinator row lock. Compare-and-swap on the
|
||||
* owner it read, so a fresh claim is never released. Returns whether it cleared one.
|
||||
*/
|
||||
export async function releaseLapsedCoordinatorOwner(tx: Pick<Db, "update">, input: { companyId: string; runId: string },
|
||||
coordinator: Coordinator | undefined, at: Date): Promise<boolean> {
|
||||
if (!coordinator?.leaseOwner || !lapsedFinishedCoordinatorOwner(coordinator, at)) return false;
|
||||
const cleared = await tx.update(nativeRunFinalizations).set({ leaseOwner: null, leaseExpiresAt: null, updatedAt: at })
|
||||
.where(and(eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.runId, input.runId),
|
||||
eq(nativeRunFinalizations.leaseOwner, coordinator.leaseOwner),
|
||||
inArray(nativeRunFinalizations.phase, [...FINISHED_COORDINATOR_PHASES])))
|
||||
.returning({ runId: nativeRunFinalizations.runId });
|
||||
return cleared.length > 0;
|
||||
}
|
||||
|
||||
/** The coordinator columns the re-arms read under the row lock. */
|
||||
export const coordinatorOwnerColumns = { leaseOwner: nativeRunFinalizations.leaseOwner,
|
||||
leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt, phase: nativeRunFinalizations.phase };
|
||||
|
||||
/** Delay before re-arm n+1, measured from re-arm n. The first re-arm is due at once. */
|
||||
const STOP_PROOF_REARM_DELAYS_MS = [30 * 60_000, 2 * 60 * 60_000, 8 * 60 * 60_000];
|
||||
|
||||
@@ -48,8 +86,7 @@ export async function rearmExhaustedStopProof(db: Db, input: WorkspaceRecoverySo
|
||||
// Lock order shared with the fallback: task, coordinator, run, recovery action.
|
||||
const [issue] = await tx.select({ id: issues.id }).from(issues).where(and(eq(issues.companyId, input.companyId),
|
||||
eq(issues.id, input.issueId))).for("update");
|
||||
const [coordinator] = await tx.select({ leaseOwner: nativeRunFinalizations.leaseOwner, leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt })
|
||||
.from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, input.companyId),
|
||||
const [coordinator] = await tx.select(coordinatorOwnerColumns).from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, input.companyId),
|
||||
eq(nativeRunFinalizations.runId, input.runId))).for("update");
|
||||
const [run] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, input.companyId),
|
||||
eq(heartbeatRuns.id, input.runId))).for("update");
|
||||
@@ -68,6 +105,8 @@ export async function rearmExhaustedStopProof(db: Db, input: WorkspaceRecoverySo
|
||||
const marker = readStopProofRecovery(action.evidence);
|
||||
const rearms = marker?.rearms ?? 0;
|
||||
if (rearms >= STOP_PROOF_MAX_REARMS || (marker?.nextRearmAt ?? 0) > now) return null;
|
||||
// A lapsed owner of the finished coordinator would refuse the new attempt's stop check.
|
||||
const releasedLapsedOwner = await releaseLapsedCoordinatorOwner(tx, input, coordinator, at);
|
||||
// Any workspace recovery the worker owns: export holds and restore holds alike.
|
||||
const next: SystemRecoveryState = { ...createSystemRecovery(run.id), revision: system.revision + 1,
|
||||
preservationOperationIds: system.preservationOperationIds, providerCooldownAt: system.providerCooldownAt,
|
||||
@@ -75,7 +114,8 @@ export async function rearmExhaustedStopProof(db: Db, input: WorkspaceRecoverySo
|
||||
const count = rearms + 1;
|
||||
const delay = STOP_PROOF_REARM_DELAYS_MS[Math.min(count - 1, STOP_PROOF_REARM_DELAYS_MS.length - 1)];
|
||||
const stopProofRecovery = { schema: STOP_PROOF_RECOVERY_SCHEMA, rearms: count, generationId: next.generationId,
|
||||
previousGenerationId: system.generationId, rearmedAt: now, nextRearmAt: count < STOP_PROOF_MAX_REARMS ? now + delay : null };
|
||||
previousGenerationId: system.generationId, rearmedAt: now, nextRearmAt: count < STOP_PROOF_MAX_REARMS ? now + delay : null,
|
||||
...(releasedLapsedOwner ? { releasedLapsedOwner: true } : {}) };
|
||||
const prior = action.evidence.workspaceRecovery as { requestIds?: unknown } | undefined;
|
||||
await notifyIssueRecovery(tx, { companyId: input.companyId });
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.workspaceRecovery);
|
||||
@@ -89,7 +129,8 @@ export async function rearmExhaustedStopProof(db: Db, input: WorkspaceRecoverySo
|
||||
return persistActivity(tx as unknown as Db, { companyId: input.companyId, actorType: "system", actorId: "workspace-recovery",
|
||||
action: "issue.workspace_recovery_rearmed", entityType: "issue", entityId: input.issueId, runId: run.id,
|
||||
details: { recoveryActionId: action.id, failureCode: system.failure?.code ?? null, rearm: count,
|
||||
maxRearms: STOP_PROOF_MAX_REARMS, previousGenerationId: system.generationId, generationId: next.generationId } });
|
||||
maxRearms: STOP_PROOF_MAX_REARMS, previousGenerationId: system.generationId, generationId: next.generationId,
|
||||
...(releasedLapsedOwner ? { releasedLapsedOwner: coordinator!.leaseOwner } : {}) } });
|
||||
});
|
||||
if (!outcome) return false;
|
||||
try { publishActivity(outcome.publication); } catch { logger.warn({ runId: input.runId }, "Recovery activity delivery deferred"); }
|
||||
|
||||
Reference in new issue
Block a user