From bf3a0670766a08cb2f404075e07c470519cf57a5 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sat, 7 Nov 2026 22:16:34 -0600 Subject: [PATCH] feat(recovery): settle the finalizer and cleanup leases on the shared ownership path (#19950) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Native runs keep their ownership in `native_run_finalizations`, and the shared ownership reconciler now settles proven-gone owners (Phase 2 of the durable ownership plan) > - The finalizer lease (`native-finalizer:*`, 5 min) and the retained-session cleanup lease (`native-cleanup:*`, 60 s) recorded only a random token. They named no holder process > - A server killed while it finalized or cleaned up kept the run waiting for the whole lease. The lease was released by age alone. A graceful shutdown also keeps an in-progress finalizer lease for the next boot to wait out > - This pull request records the holder process with each claim and moves both kinds onto the shared prover and a compare-and-swap settle > - The benefit is that a dead holder's run continues in the first pass after boot, and a live or unprovable holder is never released ## Linked Issues or Issue Description Refs #19933 (the settle hook and the first moved kind, merged). **What happened?** A finalizer or cleanup lease whose server process died stayed held until the lease expired (5 min or 60 s). The run stayed `running` and its follow-up work waited. **What did you expect?** The next boot proves the dead holder gone and frees the lease at once. A live holder, or one that cannot be proven gone, keeps its lease. **Steps to reproduce** 1. Accept a native result and let the finalizer claim it. 2. Kill the server (SIGKILL) before the finalizer commits. 3. Start the server. The run stays `running` for up to 5 minutes. ## What Changed - `native_run_finalizations.lease_holder` (jsonb, migration 0446): `{token, bootId, pid, hostHash, processStartedAt}` of the process that took a token-only lease. It names the holder only while `token` equals `lease_owner`, so renewals keep it and a later lease never inherits it. The executor's `controller_*` columns are not changed. - The finalizer claim (`native-run-finalizer.ts`) and the cleanup claim (`native-session-executor.ts`) write it in the same UPDATE as the lease, from `currentControllerProcessStamp()` (no I/O). - New `native-token-lease-settlement.ts`: compare-and-swap settle that expires exactly the proven-gone holder's outstanding lease, after it notifies the finalization or cleanup work that the expiry frees. - Both claims also wake the ownership worker queue in their transaction, so a quiet worker watches the new holder (Greptile round 1). - `ownership-registry.ts`: `native_finalizer_lease` and `native_cleanup_lease` register `settle`. Their owner is the holder process; rows without a holder keep the token fence. Listings cover outstanding leases only. - `heartbeat/recovery.ts`: restart recovery settles both kinds before it finalizes and schedules cleanup. A failure is logged and the reconciler pass retries it. - Two sparse partial indexes on `lease_expires_at` for the two listings, so the settle watch re-check stays indexed. Bespoke code deleted: none. These kinds had no liveness check; their only release was lease age (the takeover in the claim), which stays as the fence for unprovable holders. ## Verification - `cd server && npx vitest run src/__tests__/heartbeat-native-restart-suspension.test.ts -t "finalizer|cleanup|holder|exact lease"`: 12 passed (rerun after the rebase on master). The crash-point test also asserts the ownership wake for its run; it fails without the signal. - `cd server && npx vitest run src/__tests__/ownership-reconciliation-report.test.ts`: 17 passed. - Red on the base branch (server code at `bea3bfb20b`, tests and schema from this branch): 7 new tests fail, 5 pass (the 5 are existing tests). Red: the crash-point restart test, the two reconciler settle tests, the three live/unprovable counter-cases (owner was a token fence), and the exact-lease CAS test. The tests were written after the code. - New tests: - crash point: a finalizer claim was written and its holder killed before commit; the next restart finalizes the run, and this boot's claim stamps itself as holder; - a dead finalizer or cleanup holder is settled by a recording pass, never by a read-only pass (`record: false`); the accepted result and phase are untouched; - live counter-cases: a live earlier-boot PID (`pid_alive`), the current boot (`current_process`), a dead PID in another PID space (`different_process_space`, unprovable) keep the lease through a pass and a restart; - a holder written for another token, and a holderless row of the previous release, stay token fences, also with `PAPERCLIP_EPHEMERAL_SERVER=true`; - settle refuses a lease taken over after the proof (`owner_changed_since_proof`). - Not run locally: typecheck (CI runs it) and the wide suites. ## Risks - Previous-release story (rules/upgrade-safe-run-state.mdc): rows saved by the previous release have no `lease_holder`. They stay token fences and the existing claim takes them over at expiry. Neither lease outlives 5 min, so this covers at most 5 min after the upgrade. A rollback ignores the new column. - Unprovable holders (other host or PID space, unknown space): no operator action. The lease is short and self-expiring; at expiry the next claim takes it over and the old holder's next token compare-and-swap fails. That fence is the handoff. A different host is never proof. - The settle only expires a lease. Finalization and cleanup still run under their own checks (cleanup keeps its maintenance-history proofs). Settling ownership does not settle the accepted result, workspace preservation, or accounting. - Migration adds a nullable column and two partial indexes on a sparse predicate. Low risk. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, via Claude Code with tool use (shell, file edits, local test runs). ## 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 --- .../migrations/0446_native_lease_holder.sql | 3 + ...{0440_snapshot.json => 0446_snapshot.json} | 70 +++++++++- packages/db/src/migrations/meta/_journal.json | 7 + .../db/src/schema/native_run_finalizations.ts | 13 ++ ...eartbeat-native-restart-suspension.test.ts | 120 +++++++++++++++++- .../ownership-reconciliation-report.test.ts | 3 +- server/src/services/heartbeat/recovery.ts | 6 + .../native-runtime/native-run-finalizer.ts | 6 + .../native-runtime/native-session-executor.ts | 4 + .../native-token-lease-settlement.ts | 112 ++++++++++++++++ .../services/recovery/ownership-registry.ts | 40 +++++- 11 files changed, 374 insertions(+), 10 deletions(-) create mode 100644 packages/db/src/migrations/0446_native_lease_holder.sql rename packages/db/src/migrations/meta/{0440_snapshot.json => 0446_snapshot.json} (99%) create mode 100644 server/src/services/native-runtime/native-token-lease-settlement.ts diff --git a/packages/db/src/migrations/0446_native_lease_holder.sql b/packages/db/src/migrations/0446_native_lease_holder.sql new file mode 100644 index 00000000000..f7592a083ed --- /dev/null +++ b/packages/db/src/migrations/0446_native_lease_holder.sql @@ -0,0 +1,3 @@ +ALTER TABLE "native_run_finalizations" ADD COLUMN "lease_holder" jsonb;--> statement-breakpoint +CREATE INDEX "native_run_finalizations_finalizer_lease_idx" ON "native_run_finalizations" USING btree ("lease_expires_at") WHERE "native_run_finalizations"."lease_owner" like 'native-finalizer:%';--> statement-breakpoint +CREATE INDEX "native_run_finalizations_cleanup_lease_idx" ON "native_run_finalizations" USING btree ("lease_expires_at") WHERE "native_run_finalizations"."lease_owner" like 'native-cleanup:%'; \ No newline at end of file diff --git a/packages/db/src/migrations/meta/0440_snapshot.json b/packages/db/src/migrations/meta/0446_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0440_snapshot.json rename to packages/db/src/migrations/meta/0446_snapshot.json index b62a9e53a00..f7c4f4c2756 100644 --- a/packages/db/src/migrations/meta/0440_snapshot.json +++ b/packages/db/src/migrations/meta/0446_snapshot.json @@ -1,6 +1,6 @@ { - "id": "0c990bdc-6df0-44ca-8f21-28124f7b39e7", - "prevId": "d290d707-7d29-43a0-a697-06eecbdb743d", + "id": "782688f1-7843-4a93-8ea3-2cec362d269e", + "prevId": "f0d8430b-7abf-4232-86a9-11a268a0d20e", "version": "7", "dialect": "postgresql", "tables": { @@ -34880,6 +34880,12 @@ "primaryKey": false, "notNull": false }, + "process_host_hash": { + "name": "process_host_hash", + "type": "text", + "primaryKey": false, + "notNull": false + }, "legacy_process_settlement": { "name": "legacy_process_settlement", "type": "jsonb", @@ -44760,6 +44766,12 @@ "primaryKey": false, "notNull": false }, + "lease_holder": { + "name": "lease_holder", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, "controller_boot_id": { "name": "controller_boot_id", "type": "text", @@ -44772,6 +44784,12 @@ "primaryKey": false, "notNull": false }, + "controller_host_hash": { + "name": "controller_host_hash", + "type": "text", + "primaryKey": false, + "notNull": false + }, "controller_process_started_at": { "name": "controller_process_started_at", "type": "timestamp with time zone", @@ -44918,6 +44936,54 @@ "method": "btree", "with": {} }, + "native_run_finalizations_finalizer_lease_idx": { + "name": "native_run_finalizations_finalizer_lease_idx", + "columns": [ + { + "expression": "lease_expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"native_run_finalizations\".\"lease_owner\" like 'native-finalizer:%'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "native_run_finalizations_cleanup_lease_idx": { + "name": "native_run_finalizations_cleanup_lease_idx", + "columns": [ + { + "expression": "lease_expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"native_run_finalizations\".\"lease_owner\" like 'native-cleanup:%'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "native_run_finalizations_accepted_result_lease_idx": { + "name": "native_run_finalizations_accepted_result_lease_idx", + "columns": [ + { + "expression": "lease_expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"native_run_finalizations\".\"result_id\" is not null AND \"native_run_finalizations\".\"lease_owner\" is not null AND \"native_run_finalizations\".\"phase\" not in ('committed', 'terminal_failure', 'applied')", + "concurrently": false, + "method": "btree", + "with": {} + }, "native_run_finalizations_control_deadline_idx": { "name": "native_run_finalizations_control_deadline_idx", "columns": [ diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 6c3bfb7b73b..b2011fce23e 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -3109,6 +3109,13 @@ "when": 1794104103100, "tag": "0445_native_accepted_result_lease_index", "breakpoints": true + }, + { + "idx": 446, + "version": "7", + "when": 1794105010331, + "tag": "0446_native_lease_holder", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/native_run_finalizations.ts b/packages/db/src/schema/native_run_finalizations.ts index 613a9580b7c..fc4fa11076c 100644 --- a/packages/db/src/schema/native_run_finalizations.ts +++ b/packages/db/src/schema/native_run_finalizations.ts @@ -27,6 +27,11 @@ export const nativeRunFinalizations = pgTable( attempt: integer("attempt").notNull().default(0), leaseOwner: text("lease_owner"), leaseExpiresAt: timestamp("lease_expires_at", { withTimezone: true }), + // Server process that took a token-only lease (finalizer or cleanup): + // {token, bootId, pid, hostHash, processStartedAt}. It names the holder only + // while `token` equals lease_owner, so a later lease never inherits it. The + // controller_* columns stay the executor's. Null on rows saved before it existed. + leaseHolder: jsonb("lease_holder").$type>(), controllerBootId: text("controller_boot_id"), controllerPid: integer("controller_pid"), // PID space of controllerPid (controller-process-identity hostHash). Null @@ -61,6 +66,14 @@ export const nativeRunFinalizations = pgTable( .where(sql`${table.phase} = 'terminal_failure' AND ${table.failureCode} = 'native_workspace_sync_out_unsafe_archive'`), // The `native_accepted_result_lease` ownership kind: outstanding executor // leases on an accepted result that is not final yet (reconciler listing). + // The `native_finalizer_lease` and `native_cleanup_lease` ownership kinds + // (reconciler listing and settle watch). Both are sparse. + finalizerLeaseIdx: index("native_run_finalizations_finalizer_lease_idx") + .on(table.leaseExpiresAt) + .where(sql`${table.leaseOwner} like 'native-finalizer:%'`), + cleanupLeaseIdx: index("native_run_finalizations_cleanup_lease_idx") + .on(table.leaseExpiresAt) + .where(sql`${table.leaseOwner} like 'native-cleanup:%'`), acceptedResultLeaseIdx: index("native_run_finalizations_accepted_result_lease_idx") .on(table.leaseExpiresAt) .where(sql`${table.resultId} is not null AND ${table.leaseOwner} is not null AND ${table.phase} not in ('committed', 'terminal_failure', 'applied')`), diff --git a/server/src/__tests__/heartbeat-native-restart-suspension.test.ts b/server/src/__tests__/heartbeat-native-restart-suspension.test.ts index bc77912b7ca..9ac9b952ed1 100644 --- a/server/src/__tests__/heartbeat-native-restart-suspension.test.ts +++ b/server/src/__tests__/heartbeat-native-restart-suspension.test.ts @@ -2,6 +2,8 @@ import { randomUUID } from "node:crypto"; import { and, eq, sql } from "drizzle-orm"; import { activityLog, authUsers, budgetReservations, issueRecoveryActions, companyMemberships, completionContracts, heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, workspaceOperations } from "@paperclipai/db"; import { expect, it, vi } from "vitest"; +import { subscribeDatabaseWork } from "@paperclipai/db/work-signals"; +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; import { resolveLocalRuntimeApiUrl } from "../runtime-api.js"; import { heartbeatService } from "../services/heartbeat.js"; import { instanceSettingsService } from "../services/instance-settings.js"; @@ -10,11 +12,12 @@ import type { NativeSessionBackend } from "@paperclipai/paperclip-runner"; import { PaperclipControlPlanePort } from "../services/native-runtime/paperclip-control-plane-port.js"; import { CONTROL_PLANE_CONFORMANCE_OPEN, CONTROL_PLANE_CONFORMANCE_RESULT, CONTROL_PLANE_CONFORMANCE_TERMINAL } from "../vendor/paperclip-runner/testing.js"; import { describeEmbeddedPostgres, nativeRestartFixture } from "./helpers/native-restart.js"; -import { currentProcessSpaceHash } from "../services/controller-process-identity.js"; +import { currentControllerBootId, currentProcessSpaceHash } from "../services/controller-process-identity.js"; import { createOwnershipReportState, runOwnershipReconciliationPass, settleOwnableKind, type OwnershipReport } from "../services/recovery/ownership-invariant-report.js"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; import { registerOwnershipReconciliationWork } from "../services/native-runtime/native-finalization-reconciler.js"; import { OWNABLE_KINDS } from "../services/recovery/ownership-registry.js"; +import { proveOwnerGone } from "../services/owner-proof.js"; describeEmbeddedPostgres("heartbeat native recovery interrupted by graceful shutdown", () => { @@ -572,4 +575,119 @@ describeEmbeddedPostgres("heartbeat native recovery interrupted by graceful shut await coordinator.stop(); } }, 180_000); + + // Phase 2: the token-only finalizer and cleanup leases record their holder + // process (`lease_holder`, bound to the token) and settle on the shared + // ownership path when it is proven gone, never by lease age. + type Holder = { bootId: string; pid: number; hostHash: string; processStartedAt: string | null }; + const deadHolder = (hostHash = currentProcessSpaceHash()): Holder => ({ bootId: randomUUID(), pid: 2_000_000_001, + hostHash, processStartedAt: new Date(Date.now() - 60_000).toISOString() }); + async function heldBy(f: Awaited>, holder: Holder, token?: string) { + const [row] = await f.db.select({ leaseOwner: nativeRunFinalizations.leaseOwner }).from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, f.runId)); + await f.db.update(nativeRunFinalizations).set({ leaseHolder: { token: token ?? row.leaseOwner, ...holder } }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + } + const tokenKindId = (prefix: string) => prefix === "native-finalizer" ? "native_finalizer_lease" : "native_cleanup_lease"; + /** This run's record of the kind and its proof (the fixture database is shared across cases). */ + async function proofOf(f: Awaited>, prefix: string) { + const kind = OWNABLE_KINDS.find(candidate => candidate.id === tokenKindId(prefix))!; + const records = await kind.listForRuns!(f.db, [f.runId]); + if (!records.length) return null; + const owner = records[0]!.owner; + if (owner.kind === "process") return { owner: owner.kind, ...await proveOwnerGone(owner.stamp) }; + if (owner.kind === "fence") return { owner: owner.kind, ...await proveOwnerGone(owner.stamp, owner.fence) }; + return { owner: owner.kind }; + } + async function leaseOf(f: Awaited>) { + const [owner] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + return owner; + } + + it("finalizes promptly when a hard kill left the finalizer lease to its dead holder, and stamps the next holder", async () => { + // Crash point: the finalizer claimed the accepted result (lease written) + // and the server was killed before it committed. The 5-minute lease must + // not hold the run `running`; the next boot proves the holder gone. + const f = await seedAcceptedResult("FINALIZER-THEN-SIGKILL", () => `native-finalizer:${randomUUID()}`); + await heldBy(f, deadHolder()); + // A claim wakes the ownership worker, which may have gone quiet, so it + // watches the new holder. + const ownershipWakes: string[] = []; + const stop = subscribeDatabaseWork(f.db, DELIVERY_QUEUES.ownershipInvariantReport, (event, hints) => { + if (event === "settled") ownershipWakes.push(...(hints ?? [])); + }); + const { run, owner } = await restart(f).finally(stop); + expect(ownershipWakes.some(hint => hint.includes(f.runId))).toBe(true); + expect({ status: run.status, phase: owner.phase }).toEqual({ status: "succeeded", phase: "committed" }); + // This boot's own finalizer claim recorded itself as the holder of its token, + // without rewriting the executor's controller identity. + expect(owner.leaseHolder).toMatchObject({ token: expect.stringMatching(/^native-finalizer:/), + bootId: currentControllerBootId(), pid: process.pid, hostHash: currentProcessSpaceHash() }); + expect(owner.controllerPid).toBe((await currentNativeControllerIdentity()).pid); + }, 90_000); + + it.each(["native-finalizer", "native-cleanup"])("settles a dead %s holder in the reconciler pass, never in a read-only pass", async (prefix) => { + const f = await seedAcceptedResult(`TOKEN-PASS-${prefix}`, () => `${prefix}:${randomUUID()}`); + await heldBy(f, deadHolder()); + expect(await proofOf(f, prefix)).toEqual({ owner: "process", verdict: "gone", reason: "pid_dead" }); + await runOwnershipReconciliationPass(f.db, { state: createOwnershipReportState(), record: false }); + expect((await leaseOf(f)).leaseExpiresAt).toEqual(f.leaseExpiresAt); + const pass = await runOwnershipReconciliationPass(f.db, { state: createOwnershipReportState() }); + expect(pass.kinds.find(kind => kind.kind === tokenKindId(prefix))!.settled).toBeGreaterThanOrEqual(1); + const released = await leaseOf(f); + expect(released.leaseExpiresAt!.getTime()).toBeLessThanOrEqual(Date.now()); + // Ownership only: the accepted result and its workspace work are untouched. + expect({ phase: released.phase, resultId: released.resultId }).toEqual({ phase: "workspace_finalizing", + resultId: expect.any(String) }); + // An expired lease is free for the next claim and no longer in progress. + expect(await proofOf(f, prefix)).toBeNull(); + }, 60_000); + + it.each([ + { label: "live earlier-boot holder", proof: { verdict: "alive", reason: "pid_alive" }, holder: async (): Promise => ({ bootId: randomUUID(), + pid: process.pid, hostHash: currentProcessSpaceHash(), + processStartedAt: (await currentNativeControllerIdentity()).processStartedAt.toISOString() }) }, + { label: "current boot", proof: { verdict: "alive", reason: "current_process" }, holder: async (): Promise => ({ bootId: currentControllerBootId(), + pid: process.pid, hostHash: currentProcessSpaceHash(), processStartedAt: null }) }, + { label: "dead PID in another PID space", proof: { verdict: "unprovable", reason: "different_process_space" }, holder: async () => deadHolder("another-pid-space") }, + ])("keeps a finalizer lease held by a $label", async ({ holder, proof }) => { + const f = await seedAcceptedResult(`FINALIZER-KEPT-${randomUUID()}`, () => `native-finalizer:${randomUUID()}`); + await heldBy(f, await holder()); + expect(await proofOf(f, "native-finalizer")).toEqual({ owner: "process", ...proof }); + await runOwnershipReconciliationPass(f.db, { state: createOwnershipReportState() }); + await restart(f); + expect((await leaseOf(f)).leaseExpiresAt).toEqual(f.leaseExpiresAt); + }, 60_000); + + it("never lets a later lease inherit an earlier holder, and keeps holderless leases of the previous release", async () => { + // A holder written for another token names nobody; a row saved by the + // previous release has none. Both stay token fences until expiry, also on + // an ephemeral server. + const f = await seedAcceptedResult("FINALIZER-STALE-HOLDER", () => `native-finalizer:${randomUUID()}`); + await heldBy(f, deadHolder(), `native-finalizer:${randomUUID()}`); + const legacy = await seedAcceptedResult("FINALIZER-PREVIOUS-RELEASE", () => `native-finalizer:${randomUUID()}`); + vi.stubEnv("PAPERCLIP_EPHEMERAL_SERVER", "true"); + try { + for (const row of [f, legacy]) { + expect(await proofOf(row, "native-finalizer")).toEqual({ owner: "fence", verdict: "alive", reason: "fence_held" }); + } + await runOwnershipReconciliationPass(f.db, { state: createOwnershipReportState() }); + } finally { vi.unstubAllEnvs(); } + await restart(f); + expect((await leaseOf(f)).leaseExpiresAt).toEqual(f.leaseExpiresAt); + expect((await leaseOf(legacy)).leaseExpiresAt).toEqual(legacy.leaseExpiresAt); + }, 90_000); + + it("settles only the exact lease it proved: a takeover after the proof is kept", async () => { + const f = await seedAcceptedResult("FINALIZER-TAKEN-OVER", () => `native-finalizer:${randomUUID()}`); + await heldBy(f, deadHolder()); + const kind = OWNABLE_KINDS.find(candidate => candidate.id === "native_finalizer_lease")!; + const [record] = await kind.listForRuns!(f.db, [f.runId]); + const successor = `native-finalizer:${randomUUID()}`; + await f.db.update(nativeRunFinalizations).set({ leaseOwner: successor, + leaseHolder: { token: successor, ...deadHolder() } }).where(eq(nativeRunFinalizations.runId, f.runId)); + const outcome = await kind.settle!(f.db, record!, { now: new Date(), proof: { verdict: "gone", reason: "pid_dead" } }); + expect(outcome).toEqual({ settled: false, reason: "owner_changed_since_proof" }); + expect((await leaseOf(f)).leaseExpiresAt).toEqual(f.leaseExpiresAt); + }, 60_000); }); diff --git a/server/src/__tests__/ownership-reconciliation-report.test.ts b/server/src/__tests__/ownership-reconciliation-report.test.ts index ecc246bb638..1d26ce393a7 100644 --- a/server/src/__tests__/ownership-reconciliation-report.test.ts +++ b/server/src/__tests__/ownership-reconciliation-report.test.ts @@ -140,7 +140,8 @@ describe("report-only ownership reconciliation", () => { const names = OWNABLE_KINDS.map(kind => kind.id); expect(new Set(names).size).toBe(names.length); expect(names.length).toBeGreaterThanOrEqual(40); - expect(OWNABLE_KINDS.filter(kind => kind.settle !== null).map(kind => kind.id)).toEqual(["native_accepted_result_lease", "wake_request_claimed"]); + expect(OWNABLE_KINDS.filter(kind => kind.settle !== null).map(kind => kind.id)) + .toEqual(["native_accepted_result_lease", "native_finalizer_lease", "native_cleanup_lease", "wake_request_claimed"]); expect(OWNABLE_KINDS.filter(kind => !kind.listInProgress).every(kind => kind.coverageGap)).toBe(true); }); diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index cd33ccb1ed3..5858b719e0a 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -954,6 +954,12 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD payload: { restartKind: "hard", runnerDisposition: "awaiting_result_finalization", reason: released.reason }, }).catch(error => logger.warn({ err: error, runId: released.runId }, "failed to record released native result lease")); } + // Finalizer and cleanup leases of a stopped holder, so the finalization + // and cleanup below run now. A failure is retried by the reconciler pass. + for (const kind of ["native_finalizer_lease", "native_cleanup_lease"]) { + await settleOwnableKind(db, kind, { now, runIds }) + .catch(error => logger.warn({ err: error, kind }, "failed to settle native token leases at restart")); + } if (!runIds && !options?.skipDelegatedWork) { await reconcileRetainedWorkspaces(db, environmentRuntime, resumeRemoteStopComments); await reconcileNativeFinalizations(db, undefined, { diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index e584d1fb407..1fae0f2768f 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -3,6 +3,7 @@ import { isContextCompactionRun } from "../context-compaction.js"; import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { notifyNativeRecoveryNotices } from "../native-recovery-notice-notifications.js"; import { resolveCommittedNativeRecoveryNotices } from "./native-recovery-notices.js"; +import { nativeLeaseHolder, notifyNativeTokenLeaseClaimed } from "./native-token-lease-settlement.js"; import { notifyNativeRunProjection } from "../native-run-projection-notifications.js"; import { projectCommittedRun, recoveredExecutionFailureMetadata, settleNativeTerminalWake } from "./native-run-projection.js"; import { notifyNativeResultCommit } from "../native-result-commit-notifications.js"; @@ -237,11 +238,16 @@ async function claimCoordinator(input: { throw new NativeFinalizationDeferredError("lease_busy"); if (coordinator.decisionId) await notifyNativeFinalizationWork(tx, coordinator); else await notifyNativeResultCommit(tx, coordinator); + await notifyNativeTokenLeaseClaimed(tx, coordinator); const [updated] = await tx .update(nativeRunFinalizations) .set({ leaseOwner, leaseExpiresAt: new Date(now.getTime() + 5 * 60_000), + // This process, bound to the token, so a hard kill mid-finalization is + // proven and settled on the shared ownership path. The controller_* + // columns stay the executor's. + leaseHolder: nativeLeaseHolder(leaseOwner), attempt: input.preserveProviderAttempt || coordinator.controllerBootId !== null ? coordinator.attempt diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 0ffe3f696a8..a8f48d8c39b 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -1,4 +1,5 @@ import { remoteRecoveryLeaseMismatch } from "./remote-recovery-lease-mismatch.js"; +import { nativeLeaseHolder, notifyNativeTokenLeaseClaimed } from "./native-token-lease-settlement.js"; import { captureRemoteNativeProviderArtifacts } from "./native-provider-artifact-capture.js"; import { NativeSessionDiagnostics, nativeSessionDiagnostics, type NativeDiagnosticLease } from "./native-session-diagnostics.js"; import { recordAgentProviderRateLimits } from "../agent-provider-rate-limits.js"; @@ -3896,11 +3897,14 @@ async function inspectRetainedNativeSessionCleanup( }, ]), ]; + await notifyNativeTokenLeaseClaimed(tx, { companyId: run.companyId, runId: run.id }); await tx .update(nativeRunFinalizations) .set({ leaseOwner, leaseExpiresAt: new Date(Date.now() + 60_000), + // This process, bound to the token (shared ownership path). + leaseHolder: nativeLeaseHolder(leaseOwner), recoveryHistory: history, updatedAt: new Date(), }) diff --git a/server/src/services/native-runtime/native-token-lease-settlement.ts b/server/src/services/native-runtime/native-token-lease-settlement.ts new file mode 100644 index 00000000000..61f512c9ba9 --- /dev/null +++ b/server/src/services/native-runtime/native-token-lease-settlement.ts @@ -0,0 +1,112 @@ +import { and, eq, gt, like, sql } from "drizzle-orm"; +import { heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { currentControllerProcessStamp, readControllerProcess, type ControllerProcess } from "../controller-process-identity.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; +import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js"; +import { notifyNativeResultCommit } from "../native-result-commit-notifications.js"; +import { notifyRetainedNativeSessionCleanup } from "../retained-native-session-cleanup-notifications.js"; +import type { LocalOwnerStamp, OwnerProof } from "../owner-proof.js"; +import type { InProgressRecord, SettleOutcome } from "../recovery/ownership-registry.js"; + +/** + * Holder stamps and settle for the token-only native leases (plan Phase 2): + * the finalizer lease (`native-finalizer:*`, 5 min) and the retained-session + * cleanup lease (`native-cleanup:*`, 60 s). + * + * Both run inside one server process, but their claims used to record only a + * random token: the controller_* columns belong to the executor and must not + * be overwritten (the finalizer's attempt rule and restart recovery read + * them). A holder killed mid-lease therefore kept the run waiting for the + * whole TTL. The claim now writes `lease_holder` = this process's identity, + * bound to the token, in the same UPDATE (no I/O: `currentControllerProcessStamp`). + * The holder names the lease owner only while its token equals `lease_owner`, + * so renewals keep it and any later owner never inherits it. + * + * Previous release: rows it saved have no holder. They stay token fences, and + * the existing claim takes them over at expiry (at most 5 min / 60 s after the + * upgrade, since neither lease outlives that). Settle never touches them. + */ + +export type NativeTokenLeaseKind = "native_finalizer_lease" | "native_cleanup_lease"; + +const PREFIX: Record = { + native_finalizer_lease: "native-finalizer:", + native_cleanup_lease: "native-cleanup:", +}; + +/** Wakes the ownership worker for a newly claimed token lease, so a quiet + * worker (on any server) starts watching the holder. Call it in the claim's + * transaction, before the write it announces. */ +export async function notifyNativeTokenLeaseClaimed(tx: object, scope: { companyId: string; runId: string }) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.ownershipInvariantReport, + JSON.stringify({ companyId: scope.companyId, runId: scope.runId })); +} + +/** The `lease_holder` value for a token-only lease claimed by this process. */ +export function nativeLeaseHolder(token: string): Record { + return { token, ...currentControllerProcessStamp() }; +} + +/** The process holding `token`, or null when the holder was written for + * another token (or not at all, on rows saved by the previous release). */ +export function readNativeLeaseHolder(value: unknown, token: string | null | undefined): ControllerProcess | null { + if (!token || !value || typeof value !== "object" || (value as { token?: unknown }).token !== token) return null; + return readControllerProcess(value); +} + +export function nativeLeaseHolderStamp(holder: ControllerProcess, scope: NativeTokenLeaseKind): LocalOwnerStamp { + return { + kind: "local", scope, generation: null, host: null, processSpace: holder.hostHash, bootId: holder.bootId, + pid: holder.pid, processStartedAt: holder.processStartedAt, processGroupId: null, + }; +} + +/** + * Called by the shared reconciler only after `proveOwnerGone` proved the + * recorded holder gone. Expires exactly that lease (compare-and-swap on the + * token while it is still outstanding and still held by the proven process), + * after announcing the work the expiry releases. It settles ownership only: + * the finalizer or cleanup that the notification wakes does the rest, under + * its own checks. + */ +export async function settleNativeTokenLease(kind: NativeTokenLeaseKind, db: Db, record: InProgressRecord, + context: { now: Date; proof: OwnerProof }): Promise { + if (record.owner.kind !== "process") return { settled: false, reason: "owner_not_a_process" }; + const stamp = record.owner.stamp; + const { now } = context; + return db.transaction(async transaction => { + const tx = transaction as unknown as Db; + const [row] = await tx.select({ + runId: nativeRunFinalizations.runId, companyId: nativeRunFinalizations.companyId, + issueId: nativeRunFinalizations.issueId, decisionId: nativeRunFinalizations.decisionId, + agentId: heartbeatRuns.agentId, leaseOwner: nativeRunFinalizations.leaseOwner, + leaseHolder: nativeRunFinalizations.leaseHolder, + }).from(nativeRunFinalizations) + .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunFinalizations.runId), + eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId))) + .where(and(eq(nativeRunFinalizations.runId, record.recordId), like(nativeRunFinalizations.leaseOwner, `${PREFIX[kind]}%`), + gt(nativeRunFinalizations.leaseExpiresAt, now))) + .for("update", { of: nativeRunFinalizations }); + if (!row?.leaseOwner) return { settled: false, reason: "lease_no_longer_held" }; + const holder = readNativeLeaseHolder(row.leaseHolder, row.leaseOwner); + // The proof judged this exact process; the lease has moved on since. + if (!holder || holder.bootId !== stamp.bootId || holder.pid !== stamp.pid || holder.hostHash !== stamp.processSpace + || holder.processStartedAt !== (stamp.processStartedAt ?? null)) { + return { settled: false, reason: "owner_changed_since_proof" }; + } + // Work intents before the write they announce (sleep-aware background work). + if (kind === "native_finalizer_lease") { + if (row.decisionId) await notifyNativeFinalizationWork(tx, { companyId: row.companyId, runId: row.runId }); + else await notifyNativeResultCommit(tx, { companyId: row.companyId, runId: row.runId, issueId: row.issueId }); + } else { + await notifyRetainedNativeSessionCleanup(tx, { companyId: row.companyId, runId: row.runId, + issueId: row.issueId, agentId: row.agentId }); + } + const updated = await tx.update(nativeRunFinalizations).set({ leaseExpiresAt: now, updatedAt: now }) + .where(and(eq(nativeRunFinalizations.runId, row.runId), eq(nativeRunFinalizations.leaseOwner, row.leaseOwner), + gt(nativeRunFinalizations.leaseExpiresAt, now), + sql`${nativeRunFinalizations.leaseHolder}->>'token' = ${row.leaseOwner}`)) + .returning({ runId: nativeRunFinalizations.runId }); + return updated.length ? { settled: true, reason: context.proof.reason } : { settled: false, reason: "lease_no_longer_held" }; + }); +} diff --git a/server/src/services/recovery/ownership-registry.ts b/server/src/services/recovery/ownership-registry.ts index 988e568c767..8374222b023 100644 --- a/server/src/services/recovery/ownership-registry.ts +++ b/server/src/services/recovery/ownership-registry.ts @@ -32,6 +32,12 @@ import { settleNativeAcceptedResultLease, } from "../native-runtime/native-accepted-result-lease-settlement.js"; import { settleClaimedWakeRequest } from "./wake-request-claimed-settlement.js"; +import { + nativeLeaseHolderStamp, + readNativeLeaseHolder, + settleNativeTokenLease, + type NativeTokenLeaseKind, +} from "../native-runtime/native-token-lease-settlement.js"; /** * Registry of ownable record kinds (plan section C). @@ -187,6 +193,13 @@ function nrfFenceOwner(row: NrfRow, scope: string): RecordOwner { }; } +/** A finalizer or cleanup lease is owned by the process in `lease_holder`. + * Rows saved before that column keep the token fence (expiry plus takeover). */ +function nrfTokenLeaseOwner(row: NrfRow, scope: NativeTokenLeaseKind): RecordOwner { + const holder = readNativeLeaseHolder(row.leaseHolder, row.leaseOwner); + return holder ? { kind: "process", stamp: nativeLeaseHolderStamp(holder, scope) } : nrfFenceOwner(row, scope); +} + // ---------------------------------------------------------------------------- // heartbeat_runs @@ -382,12 +395,27 @@ export const OWNABLE_KINDS: readonly OwnableKind[] = [ settle: settleNativeAcceptedResultLease, onUnprovable: requestNativeAcceptedResultLeaseOperatorAction, }, - nrfKind("native_finalizer_lease", "Finalizer lease", "accepted_result", - and(like(nativeRunFinalizations.leaseOwner, "native-finalizer:%"), notInArray(nativeRunFinalizations.phase, TERMINAL_PHASES)), - row => nrfFenceOwner(row, "native_finalizer_lease")), - nrfKind("native_cleanup_lease", "Cleanup lease", "workspace_preservation", - like(nativeRunFinalizations.leaseOwner, "native-cleanup:%"), - row => nrfFenceOwner(row, "native_cleanup_lease")), + // Shared path since Phase 2: the token-only finalizer and cleanup leases + // record their holder process (`lease_holder`, bound to the token); settle + // expires the exact lease of a holder proven gone, so finalization or + // cleanup runs now instead of after the TTL. Outstanding leases only: an + // expired one is free for the next claim. A holder on another host or PID + // space is unprovable and keeps its lease until expiry, when the next claim + // takes it over and the old holder's token compare-and-swap fails: that + // fence is the handoff, never a host difference. + { + ...nrfKind("native_finalizer_lease", "Finalizer lease", "accepted_result", + and(like(nativeRunFinalizations.leaseOwner, "native-finalizer:%"), notInArray(nativeRunFinalizations.phase, TERMINAL_PHASES), + sql`${nativeRunFinalizations.leaseExpiresAt} > now()`), + row => nrfTokenLeaseOwner(row, "native_finalizer_lease")), + settle: (db, record, context) => settleNativeTokenLease("native_finalizer_lease", db, record, context), + }, + { + ...nrfKind("native_cleanup_lease", "Cleanup lease", "workspace_preservation", + and(like(nativeRunFinalizations.leaseOwner, "native-cleanup:%"), sql`${nativeRunFinalizations.leaseExpiresAt} > now()`), + row => nrfTokenLeaseOwner(row, "native_cleanup_lease")), + settle: (db, record, context) => settleNativeTokenLease("native_cleanup_lease", db, record, context), + }, runKind("native_ownership_unverified_hold", "Ownership-unverified hold", "ownership", and(eq(heartbeatRuns.runtimeMode, "native"), eq(heartbeatRuns.status, "running"), or( and(eq(heartbeatRuns.errorCode, NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE), eq(heartbeatRuns.nativePhase, "terminal_failure")),