mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
feat(recovery): settle the finalizer and cleanup leases on the shared ownership path (#19950)
## 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 <noreply@anthropic.com>
This commit is contained in:
1 parent
ca34bed257
commit
bf3a067076
11 files changed
+374
-10
No files matched your search
@@ -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:%';
|
||||
+68
-2
@@ -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": [
|
||||
+7
@@ -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
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -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<Record<string, unknown>>(),
|
||||
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')`),
|
||||
|
||||
@@ -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<ReturnType<typeof seedAcceptedResult>>, 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<ReturnType<typeof seedAcceptedResult>>, 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<ReturnType<typeof seedAcceptedResult>>) {
|
||||
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<Holder> => ({ bootId: randomUUID(),
|
||||
pid: process.pid, hostHash: currentProcessSpaceHash(),
|
||||
processStartedAt: (await currentNativeControllerIdentity()).processStartedAt.toISOString() }) },
|
||||
{ label: "current boot", proof: { verdict: "alive", reason: "current_process" }, holder: async (): Promise<Holder> => ({ 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);
|
||||
});
|
||||
@@ -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);
|
||||
});
|
||||
|
||||
|
||||
@@ -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, {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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(),
|
||||
})
|
||||
|
||||
@@ -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<NativeTokenLeaseKind, string> = {
|
||||
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<string, unknown> {
|
||||
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<SettleOutcome> {
|
||||
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" };
|
||||
});
|
||||
}
|
||||
@@ -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")),
|
||||
|
||||
Reference in new issue
Block a user