mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
feat(recovery): settle the legacy controller lease on the shared ownership path (#19953)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Legacy (non-native) runs hold a controller lease on `heartbeat_runs` while the server prepares and dispatches the adapter. The shared ownership reconciler now settles proven-gone owners (Phase 2 of the durable ownership plan) > - The lease stamped only a boot id. The reconciler could never prove its owner gone (`missing_identity` in the Phase 2 data), and recovery released it by age: lease expiry (60 s), then a 5-minute age guard on dispatched runs > - A hard-killed server's legacy run therefore stayed `running` for up to 6 minutes after the next boot > - This pull request records the controller process with the claim, registers a compare-and-swap settle, and makes live-run recovery use the shared prover instead of the age guard when the controller is proven gone > - The benefit is that a dead controller's run is settled at once, while a live, unprovable, or previous-release controller keeps today's behavior ## Linked Issues or Issue Description Refs #19933 (the settle hook, merged). Stacked on #19950 (migration 0446, registry) and targets its branch until it merges; then it retargets to master. **What happened?** After a hard kill, a dispatched legacy run stayed `running` until its controller lease expired and its last update was 5 minutes old. **What did you expect?** The next boot proves the old controller gone and fails the run as a lost process at once. **Steps to reproduce** 1. Start a legacy (non-native) run and let it dispatch the adapter. 2. Kill the server (SIGKILL). 3. Start the server. The run stays `running` for up to 6 minutes. ## What Changed - `heartbeat_runs.controller_process` (jsonb, migration 0447): `{bootId, pid, hostHash, processStartedAt}` of the claiming server process. It names the controller only while its `bootId` equals `controller_boot_id`. Revocation swaps the boot id, so a revoked lease never keeps it. - `legacyControllerClaim` writes it in the same UPDATE (from `currentControllerProcessStamp()`, no I/O). - `legacy-controller-lease.ts`: `legacyControllerOwnerStamp`, `legacyControllerProvenGone` (calls `proveOwnerGone`), and `settleLegacyControllerLease` (compare-and-swap expiry of the proven controller's outstanding lease after it wakes live-run recovery). - `ownership-registry.ts`: `legacy_controller_lease` registers `settle` and `listForRuns`. The owner is the bound process; other rows are boot-id fences. The listing covers outstanding leases only. - `heartbeat/recovery.ts` (3 small hunks): a controller of another boot that the shared prover proves gone has its live lease settled through `settleOwnableKind` and skips the age guard. The local child-process check after it is unchanged, so a live child still keeps the run running. - Migration 0447 builds the partial index with `CREATE INDEX CONCURRENTLY` (heartbeat_runs is a known-large table) and is added to `MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS`. Bespoke code deleted: none. The kind had no liveness proof; the age guard stays for controllers that are not proven gone. ## Verification - `cd server && npx vitest run src/services/legacy-controller-lease.test.ts`: 33 passed. - `cd server && npx vitest run src/__tests__/ownership-reconciliation-report.test.ts`: 17 passed. - Red with `recovery.ts` and `ownership-registry.ts` from the base commit (tests, schema and lease module from this branch): 4 failed, 29 passed. Red: the crash-point reaper test, the expired-lease age-guard test, the reconciler settle test, the exact-controller CAS test. The tests were written after the code. - New tests: - the claim stamps this process, and a changed boot id unbinds it; - crash point: a controller dispatched the adapter and was killed with its lease live; the startup reaper (default 5-minute age guard) fails the run as `process_lost` at once; - a dead controller's expired lease skips the age guard; - counter-cases keep the run and lease through the reaper and a reconciler pass: a live earlier-boot PID, a dead PID in another PID space, and a previous-release row with no `controller_process`; an expired lease of a live controller stays behind the age guard; - a read-only pass (`record: false`) never settles; a recording pass expires the lease and leaves the run to live-run recovery; - settle refuses a controller that changed after the 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 `controller_process`. They stay boot-id fences with today's release (expiry, revocation, age guard). A rollback ignores the column. - Unprovable controllers (another host or PID space, unknown space, a live PID with an unreadable start time) never release early. Their handoff stays the lease expiry plus revocation, and the age guard. A different host is never proof. - Behavior change: a dispatched run whose controller is proven gone is failed as `process_lost` without the 5-minute wait. The retry decision is unchanged. - The settle expires the lease only. Live-run recovery still settles the run under its own checks. Accounting and workspace obligations are not touched. - The index build is concurrent. A cancelled build is retried by the migrator's invalid-index handling. ## 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:
DottaandClaude Opus 5.5 authored and GitHub committed
1 parent
0cb9770d1c
commit
576ba29fff
10 files changed
+372
-16
No files matched your search
@@ -613,6 +613,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([
|
||||
"0432_strong_mole_man.sql",
|
||||
"0435_premium_puff_adder.sql",
|
||||
"0436_legal_reavers.sql",
|
||||
"0447_legacy_controller_process.sql",
|
||||
|
||||
|
||||
"0384_funny_adam_warlock.sql",
|
||||
|
||||
@@ -0,0 +1,2 @@
|
||||
ALTER TABLE "heartbeat_runs" ADD COLUMN IF NOT EXISTS "controller_process" jsonb;--> statement-breakpoint
|
||||
CREATE INDEX CONCURRENTLY IF NOT EXISTS "heartbeat_runs_legacy_controller_lease_idx" ON "heartbeat_runs" USING btree ("controller_lease_expires_at") WHERE "heartbeat_runs"."runtime_mode" = 'legacy' AND "heartbeat_runs"."status" = 'running' AND "heartbeat_runs"."controller_boot_id" is not null;
|
||||
+84
-2
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"id": "a2ec96c6-3cf1-4c51-8ee8-b35f090bfd42",
|
||||
"prevId": "0c990bdc-6df0-44ca-8f21-28124f7b39e7",
|
||||
"id": "76f06e29-53a8-4ecc-809a-d52b6bbb6520",
|
||||
"prevId": "782688f1-7843-4a93-8ea3-2cec362d269e",
|
||||
"version": "7",
|
||||
"dialect": "postgresql",
|
||||
"tables": {
|
||||
@@ -34856,6 +34856,12 @@
|
||||
"primaryKey": false,
|
||||
"notNull": false
|
||||
},
|
||||
"controller_process": {
|
||||
"name": "controller_process",
|
||||
"type": "jsonb",
|
||||
"primaryKey": false,
|
||||
"notNull": false
|
||||
},
|
||||
"execution_stage": {
|
||||
"name": "execution_stage",
|
||||
"type": "text",
|
||||
@@ -34880,6 +34886,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",
|
||||
@@ -35119,6 +35131,22 @@
|
||||
"method": "btree",
|
||||
"with": {}
|
||||
},
|
||||
"heartbeat_runs_legacy_controller_lease_idx": {
|
||||
"name": "heartbeat_runs_legacy_controller_lease_idx",
|
||||
"columns": [
|
||||
{
|
||||
"expression": "controller_lease_expires_at",
|
||||
"isExpression": false,
|
||||
"asc": true,
|
||||
"nulls": "last"
|
||||
}
|
||||
],
|
||||
"isUnique": false,
|
||||
"where": "\"heartbeat_runs\".\"runtime_mode\" = 'legacy' AND \"heartbeat_runs\".\"status\" = 'running' AND \"heartbeat_runs\".\"controller_boot_id\" is not null",
|
||||
"concurrently": false,
|
||||
"method": "btree",
|
||||
"with": {}
|
||||
},
|
||||
"heartbeat_runs_cost_accounting_pending_idx": {
|
||||
"name": "heartbeat_runs_cost_accounting_pending_idx",
|
||||
"columns": [
|
||||
@@ -44760,6 +44788,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",
|
||||
@@ -44924,6 +44958,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
@@ -3116,6 +3116,13 @@
|
||||
"when": 1794105010331,
|
||||
"tag": "0446_native_lease_holder",
|
||||
"breakpoints": true
|
||||
},
|
||||
{
|
||||
"idx": 447,
|
||||
"version": "7",
|
||||
"when": 1794105389611,
|
||||
"tag": "0447_legacy_controller_process",
|
||||
"breakpoints": true
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -92,6 +92,11 @@ export const heartbeatRuns = pgTable(
|
||||
// Legacy controller lease. A PID alone is not an identity across containers.
|
||||
controllerBootId: uuid("controller_boot_id"),
|
||||
controllerLeaseExpiresAt: timestamp("controller_lease_expires_at", { withTimezone: true }),
|
||||
// Server process that claimed the legacy controller lease ({bootId, pid,
|
||||
// hostHash, processStartedAt}). It names the controller only while its
|
||||
// bootId equals controller_boot_id; revocation swaps the boot id, so a
|
||||
// revoked lease never keeps it. Null on rows saved before it existed.
|
||||
controllerProcess: jsonb("controller_process").$type<Record<string, unknown>>(),
|
||||
executionStage: text("execution_stage"),
|
||||
processPid: integer("process_pid"),
|
||||
processGroupId: integer("process_group_id"),
|
||||
@@ -151,6 +156,11 @@ export const heartbeatRuns = pgTable(
|
||||
treeHoldSuppressionIdx: index("heartbeat_runs_tree_hold_suppression_idx")
|
||||
.on(table.companyId, sql`(${table.resultJson}->'treeHoldSuppression'->>'holdId')`)
|
||||
.where(sql`${table.status} = 'cancelled' and ${table.resultJson} ? 'treeHoldSuppression'`),
|
||||
// The `legacy_controller_lease` ownership kind (reconciler listing and
|
||||
// settle watch): outstanding legacy controller leases of running runs.
|
||||
legacyControllerLeaseIdx: index("heartbeat_runs_legacy_controller_lease_idx")
|
||||
.on(table.controllerLeaseExpiresAt)
|
||||
.where(sql`${table.runtimeMode} = 'legacy' AND ${table.status} = 'running' AND ${table.controllerBootId} is not null`),
|
||||
costAccountingPendingIdx: index("heartbeat_runs_cost_accounting_pending_idx").on(table.updatedAt, table.id).where(sql`${table.costAccountingPending} = true`),
|
||||
costDataQualityIdx: index("heartbeat_runs_cost_data_quality_idx")
|
||||
.on(table.companyId, sql`coalesce(${table.finishedAt}, ${table.createdAt}) desc`, table.id)
|
||||
|
||||
@@ -140,8 +140,9 @@ 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", "native_finalizer_lease", "native_cleanup_lease", "environment_run_lease_active", "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", "environment_run_lease_active",
|
||||
"wake_request_claimed", "legacy_controller_lease"]);
|
||||
expect(OWNABLE_KINDS.filter(kind => !kind.listInProgress).every(kind => kind.coverageGap)).toBe(true);
|
||||
});
|
||||
|
||||
|
||||
@@ -32,6 +32,7 @@ import { pendingCaseHandoffForRun } from "./case-handoff-lifecycle.js";
|
||||
import { retireInheritedRemoteRunnerIdentity } from "../native-runtime/native-inherited-runner-retry.js";
|
||||
import {
|
||||
legacyControllerBootId,
|
||||
legacyControllerProvenGone,
|
||||
liveLegacyControllerExpiresAt,
|
||||
revokeExpiredLegacyControllerWithState,
|
||||
} from "../legacy-controller-lease.js";
|
||||
@@ -2590,8 +2591,15 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
|
||||
scope.onDeferredDeadline?.(nativeControllerLeaseExpiresAt.getTime());
|
||||
continue;
|
||||
}
|
||||
// The shared prover (Phase 2): a controller of another boot whose
|
||||
// recorded process is proven gone holds no lease and needs no age guard.
|
||||
const legacyControllerGone = run.runtimeMode === "legacy" && run.controllerBootId !== legacyControllerBootId
|
||||
&& await legacyControllerProvenGone(run);
|
||||
const legacyLease = await liveLegacyControllerExpiresAt(db, run);
|
||||
if (legacyLease) { scope.onDeferredDeadline?.(legacyLease.getTime()); continue; }
|
||||
if (legacyLease && !(legacyControllerGone
|
||||
&& (await settleOwnableKind(db, "legacy_controller_lease", { runIds: [run.id] })).length > 0)) {
|
||||
scope.onDeferredDeadline?.(legacyLease.getTime()); continue;
|
||||
}
|
||||
|
||||
// A legacy controller proves it is alive only by renewing its lease, and
|
||||
// the check above found no live lease (a missing expiry also lands here;
|
||||
@@ -2609,7 +2617,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
|
||||
!run.processPid &&
|
||||
!run.processGroupId;
|
||||
// Apply staleness threshold to avoid false positives
|
||||
if (staleThresholdMs > 0 && !priorBootSetupExpired) {
|
||||
if (staleThresholdMs > 0 && !priorBootSetupExpired && !legacyControllerGone) {
|
||||
const refTime = run.updatedAt ? new Date(run.updatedAt).getTime() : 0;
|
||||
if (now.getTime() - refTime < staleThresholdMs) {
|
||||
scope.onDeferredDeadline?.(refTime + staleThresholdMs);
|
||||
@@ -2680,10 +2688,19 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
|
||||
(!monitorNextCheckAt || monitorNextCheckAt.getTime() <= now.getTime());
|
||||
const revokedState = await revokeExpiredLegacyControllerWithState(db, run);
|
||||
if (!revokedState) continue;
|
||||
// The scan snapshot can predate a child the old controller saved just
|
||||
// before it died. The child checks above used that snapshot, so a proven
|
||||
// gone controller is never enough here: when revocation reports a child
|
||||
// the snapshot did not, re-check on the next pass from the fresh row.
|
||||
if (legacyControllerGone && ((revokedState.processPid ?? null) !== (run.processPid ?? null)
|
||||
|| (revokedState.processGroupId ?? null) !== (run.processGroupId ?? null))) {
|
||||
scope.onDeferredDeadline?.(Date.now() + 1_000);
|
||||
continue;
|
||||
}
|
||||
// The earlier snapshot may predate a final renewal that admitted the
|
||||
// adapter. Revocation leaves a fresh foreign lease, so a later pass
|
||||
// applies the age guard to that dispatched run.
|
||||
if (priorBootSetupExpired && staleThresholdMs > 0 && (revokedState.executionStage !== "preparing"
|
||||
if (priorBootSetupExpired && !legacyControllerGone && staleThresholdMs > 0 && (revokedState.executionStage !== "preparing"
|
||||
|| revokedState.processPid || revokedState.processGroupId)) {
|
||||
const refTime = run.updatedAt ? new Date(run.updatedAt).getTime() : 0;
|
||||
scope.onDeferredDeadline?.(Math.max(refTime + staleThresholdMs, Date.now() + 60_000));
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { and, eq, ne, sql } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { agents, closeLivenessDb, companies, createDb, heartbeatRuns, livenessDb, subscribeDatabaseWork } from "@paperclipai/db";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js";
|
||||
@@ -8,8 +8,24 @@ import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
|
||||
import { heartbeatService } from "./heartbeat.js";
|
||||
import { hasLiveLegacyController, LEGACY_CONTROLLER_LEASE_MS, LEGACY_CONTROLLER_RENEW_MS,
|
||||
LEGACY_CONTROLLER_RENEWAL_RETRY_DELAY_MS, legacyControllerBootId, legacyControllerClaim, LegacyControllerLeaseLostError,
|
||||
renewLegacyControllerLease, revokeExpiredLegacyController, revokeExpiredLegacyControllerWithState,
|
||||
watchLegacyControllerLease } from "./legacy-controller-lease.js";
|
||||
legacyControllerOwnerStamp, renewLegacyControllerLease, revokeExpiredLegacyController,
|
||||
revokeExpiredLegacyControllerWithState, watchLegacyControllerLease } from "./legacy-controller-lease.js";
|
||||
import { currentProcessSpaceHash, currentProcessStartedAtMs } from "./controller-process-identity.js";
|
||||
import { proveOwnerGone } from "./owner-proof.js";
|
||||
import * as legacyControllerLease from "./legacy-controller-lease.js";
|
||||
|
||||
/** Runs just before recovery's revocation, to model a controller that saved
|
||||
* its child's PID after recovery's scan and then died. Null in other cases. */
|
||||
const beforeRevocation: { hook: null | ((runId: string) => Promise<void>) } = { hook: null };
|
||||
vi.mock("./legacy-controller-lease.js", async importOriginal => {
|
||||
const actual = await importOriginal<typeof legacyControllerLease>();
|
||||
return { ...actual, revokeExpiredLegacyControllerWithState: async (...args: Parameters<typeof actual.revokeExpiredLegacyControllerWithState>) => {
|
||||
if (beforeRevocation.hook) await beforeRevocation.hook(args[1].id);
|
||||
return actual.revokeExpiredLegacyControllerWithState(...args);
|
||||
} };
|
||||
});
|
||||
import { createOwnershipReportState, runOwnershipReconciliationPass } from "./recovery/ownership-invariant-report.js";
|
||||
import { OWNABLE_KINDS } from "./recovery/ownership-registry.js";
|
||||
|
||||
type WatchArgs = Parameters<typeof watchLegacyControllerLease>;
|
||||
// A renewal test double: each attempt runs in a transaction on the liveness pool.
|
||||
@@ -249,6 +265,119 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
|
||||
it("leaves native controller ownership to the native coordinator", () => {
|
||||
expect(legacyControllerClaim("native")).toEqual({});
|
||||
});
|
||||
|
||||
// Phase 2: the claim records its controller process, and a controller of
|
||||
// another boot proven gone by the shared prover holds no lease and needs no
|
||||
// age guard. Proof is by identity, never by lease age.
|
||||
it("stamps the claiming process, bound to the lease's boot id", async () => {
|
||||
const run = await seed();
|
||||
expect(run.controllerProcess).toMatchObject({ bootId: legacyControllerBootId, pid: process.pid,
|
||||
hostHash: currentProcessSpaceHash() });
|
||||
expect(legacyControllerOwnerStamp(run)).toMatchObject({ bootId: legacyControllerBootId, pid: process.pid });
|
||||
// Revocation swaps the boot id, so the revoked lease names nobody.
|
||||
expect(legacyControllerOwnerStamp({ ...run, controllerBootId: randomUUID() })).toBeNull();
|
||||
});
|
||||
|
||||
type Controller = { bootId: string; pid: number; hostHash: string; processStartedAt: string | null };
|
||||
const deadController = (hostHash = currentProcessSpaceHash()): Controller => ({ bootId: randomUUID(),
|
||||
pid: 2_000_000_001, hostHash, processStartedAt: new Date(Date.now() - 60_000).toISOString() });
|
||||
/** A run whose adapter was dispatched by `controller`, with its lease still live. */
|
||||
async function dispatchedBy(controller: Controller | null) {
|
||||
const run = await seed();
|
||||
await db.update(heartbeatRuns).set({ controllerBootId: controller?.bootId ?? randomUUID(),
|
||||
controllerProcess: controller, executionStage: "dispatching" }).where(eq(heartbeatRuns.id, run.id));
|
||||
return run.id;
|
||||
}
|
||||
const saved = async (id: string) => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)))[0]!;
|
||||
/** The startup reaper with the default age guard, over this case's run only:
|
||||
* earlier cases leave running rows in the shared database. */
|
||||
async function reapOnly(id: string) {
|
||||
await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date() })
|
||||
.where(and(eq(heartbeatRuns.status, "running"), ne(heartbeatRuns.id, id)));
|
||||
await heartbeatService(db).reapOrphanedRuns({ staleThresholdMs: 5 * 60_000 });
|
||||
}
|
||||
|
||||
it("settles a hard-killed controller's run at once, before its lease expires and without the age guard", async () => {
|
||||
// Crash point: the controller claimed the run, dispatched the adapter, and
|
||||
// was killed. The next boot proves it gone and does not wait 60 s plus 5 min.
|
||||
const id = await dispatchedBy(deadController());
|
||||
await reapOnly(id);
|
||||
expect(await saved(id)).toMatchObject({ status: "failed", errorCode: "process_lost" });
|
||||
});
|
||||
|
||||
it("never fails a run whose dead controller saved a live child after recovery's scan", async () => {
|
||||
// Race: the scan read the run before the old controller saved its child's
|
||||
// PID and died. The controller is proven gone, but its child still runs.
|
||||
const id = await dispatchedBy(deadController());
|
||||
beforeRevocation.hook = async runId => {
|
||||
await db.update(heartbeatRuns).set({ processPid: process.pid, processHostHash: currentProcessSpaceHash(),
|
||||
processStartedAt: new Date((await currentProcessStartedAtMs()) ?? Date.now()) }).where(eq(heartbeatRuns.id, runId));
|
||||
};
|
||||
try {
|
||||
await reapOnly(id);
|
||||
} finally { beforeRevocation.hook = null; }
|
||||
expect((await saved(id)).status).not.toBe("failed");
|
||||
});
|
||||
|
||||
it("settles a dead controller's expired lease without waiting out the age guard", async () => {
|
||||
const id = await dispatchedBy(deadController());
|
||||
await expire(id);
|
||||
await reapOnly(id);
|
||||
expect(await saved(id)).toMatchObject({ status: "failed", errorCode: "process_lost" });
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ label: "a live earlier-boot controller", controller: async (): Promise<Controller> => ({ bootId: randomUUID(),
|
||||
pid: process.pid, hostHash: currentProcessSpaceHash(),
|
||||
processStartedAt: new Date((await currentProcessStartedAtMs())!).toISOString() }) },
|
||||
{ label: "a dead PID in another PID space", controller: async () => deadController("another-pid-space") },
|
||||
{ label: "a lease saved by the previous release", controller: async () => null },
|
||||
])("keeps the run and lease of $label", async ({ controller }) => {
|
||||
const id = await dispatchedBy(await controller());
|
||||
const before = await saved(id);
|
||||
await reapOnly(id);
|
||||
await runOwnershipReconciliationPass(db, { state: createOwnershipReportState() });
|
||||
const after = await saved(id);
|
||||
expect({ status: after.status, errorCode: after.errorCode, expiresAt: after.controllerLeaseExpiresAt })
|
||||
.toEqual({ status: "running", errorCode: null, expiresAt: before.controllerLeaseExpiresAt });
|
||||
});
|
||||
|
||||
it("keeps an expired lease of a live earlier-boot controller behind the age guard", async () => {
|
||||
const id = await dispatchedBy({ bootId: randomUUID(), pid: process.pid, hostHash: currentProcessSpaceHash(),
|
||||
processStartedAt: new Date((await currentProcessStartedAtMs())!).toISOString() });
|
||||
await expire(id);
|
||||
await reapOnly(id);
|
||||
expect(await saved(id)).toMatchObject({ status: "running", errorCode: null });
|
||||
});
|
||||
|
||||
it("settles a dead controller's lease in the reconciler pass, never in a read-only pass", async () => {
|
||||
const id = await dispatchedBy(deadController());
|
||||
const kind = OWNABLE_KINDS.find(candidate => candidate.id === "legacy_controller_lease")!;
|
||||
const [record] = await kind.listForRuns!(db, [id]);
|
||||
expect(record!.owner.kind === "process" && await proveOwnerGone(record!.owner.stamp))
|
||||
.toEqual({ verdict: "gone", reason: "pid_dead" });
|
||||
const before = await saved(id);
|
||||
await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false });
|
||||
expect((await saved(id)).controllerLeaseExpiresAt).toEqual(before.controllerLeaseExpiresAt);
|
||||
await runOwnershipReconciliationPass(db, { state: createOwnershipReportState() });
|
||||
const after = await saved(id);
|
||||
expect(after.controllerLeaseExpiresAt!.getTime()).toBeLessThanOrEqual(Date.now());
|
||||
// Ownership only: live-run recovery, woken by the settle, settles the run.
|
||||
expect(after.status).toBe("running");
|
||||
expect(await kind.listForRuns!(db, [id])).toEqual([]);
|
||||
});
|
||||
|
||||
it("settles only the exact controller it proved", async () => {
|
||||
const id = await dispatchedBy(deadController());
|
||||
const kind = OWNABLE_KINDS.find(candidate => candidate.id === "legacy_controller_lease")!;
|
||||
const [record] = await kind.listForRuns!(db, [id]);
|
||||
const before = await saved(id);
|
||||
await db.update(heartbeatRuns).set({ controllerProcess: { ...before.controllerProcess, pid: 2_000_000_002 } })
|
||||
.where(eq(heartbeatRuns.id, id));
|
||||
expect(await kind.settle!(db, record!, { now: new Date(), proof: { verdict: "gone", reason: "pid_dead" } }))
|
||||
.toEqual({ settled: false, reason: "owner_changed_since_proof" });
|
||||
expect((await saved(id)).controllerLeaseExpiresAt).toEqual(before.controllerLeaseExpiresAt);
|
||||
});
|
||||
});
|
||||
|
||||
describe("legacy controller lease loss attribution", () => {
|
||||
|
||||
@@ -3,7 +3,13 @@ import { and, eq, gt, lte, sql } from "drizzle-orm";
|
||||
import { heartbeatRuns, livenessDb, type Db } from "@paperclipai/db";
|
||||
import type { LegacyControllerLeaseLossCause, LegacyControllerLeaseLossTiming } from "./run-cancellation.js";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { currentControllerBootId } from "./controller-process-identity.js";
|
||||
import {
|
||||
currentControllerBootId,
|
||||
currentControllerProcessStamp,
|
||||
readControllerProcess,
|
||||
} from "./controller-process-identity.js";
|
||||
import { proveOwnerGone, type LocalOwnerStamp, type OwnerProof } from "./owner-proof.js";
|
||||
import type { InProgressRecord, SettleOutcome } from "./recovery/ownership-registry.js";
|
||||
|
||||
// A boot UUID has meaning across containers; a numeric PID does not. Legacy
|
||||
// leases stamp the single process boot id shared with native claims.
|
||||
@@ -34,6 +40,9 @@ export function legacyControllerClaim(runtimeMode: string) {
|
||||
return {
|
||||
legacyProcessSettlement: null,
|
||||
controllerBootId: legacyControllerBootId,
|
||||
// This process, so a hard-killed controller is proven gone on the shared
|
||||
// ownership path (no I/O: the claim is on the dispatch path).
|
||||
controllerProcess: currentControllerProcessStamp(),
|
||||
controllerLeaseExpiresAt: sql`clock_timestamp() + interval '60 seconds'`,
|
||||
executionStage: "preparing",
|
||||
};
|
||||
@@ -142,6 +151,74 @@ export async function revokeExpiredLegacyControllerWithState(
|
||||
return revoked ?? null;
|
||||
}
|
||||
|
||||
/**
|
||||
* The recorded controller process of a legacy lease (plan Phase 2), or null.
|
||||
* `controller_process` names the controller only while its boot id equals
|
||||
* `controller_boot_id`: revocation swaps in a random boot id, so a revoked
|
||||
* lease never keeps it. Rows saved by the previous release have none; their
|
||||
* lease stays a boot-id fence (expiry, revocation and the age guard).
|
||||
*/
|
||||
export function legacyControllerOwnerStamp(
|
||||
run: Pick<Run, "controllerBootId" | "controllerProcess">,
|
||||
): LocalOwnerStamp | null {
|
||||
const controller = readControllerProcess(run.controllerProcess);
|
||||
if (!controller || !run.controllerBootId || controller.bootId !== run.controllerBootId) return null;
|
||||
return { kind: "local", scope: "legacy_controller_lease", generation: null, host: null,
|
||||
processSpace: controller.hostHash, bootId: controller.bootId, pid: controller.pid,
|
||||
processStartedAt: controller.processStartedAt, processGroupId: null };
|
||||
}
|
||||
|
||||
/** True only when the shared prover proves the recorded legacy controller gone. */
|
||||
export async function legacyControllerProvenGone(
|
||||
run: Pick<Run, "runtimeMode" | "controllerBootId" | "controllerProcess">,
|
||||
): Promise<boolean> {
|
||||
if (run.runtimeMode === "native") return false;
|
||||
const stamp = legacyControllerOwnerStamp(run);
|
||||
return stamp !== null && (await proveOwnerGone(stamp)).verdict === "gone";
|
||||
}
|
||||
|
||||
/**
|
||||
* Settle of the `legacy_controller_lease` ownership kind. The shared
|
||||
* reconciler calls it only after `proveOwnerGone` proved the recorded
|
||||
* controller gone. It expires exactly that controller's outstanding lease
|
||||
* (compare-and-swap on run, boot id and recorded pid) after waking live-run
|
||||
* recovery, which then revokes it and settles the run as a lost process under
|
||||
* its own checks (a live child process still keeps it running). It settles
|
||||
* ownership only.
|
||||
*/
|
||||
export async function settleLegacyControllerLease(db: Db, record: InProgressRecord,
|
||||
context: { now: Date; proof: OwnerProof }): Promise<SettleOutcome> {
|
||||
if (record.owner.kind !== "process" || !record.companyId) return { settled: false, reason: "owner_not_a_process" };
|
||||
const stamp = record.owner.stamp;
|
||||
if (!stamp.bootId) return { settled: false, reason: "owner_not_a_process" };
|
||||
const companyId = record.companyId;
|
||||
const bootId = stamp.bootId;
|
||||
return db.transaction(async transaction => {
|
||||
const tx = transaction as unknown as Db;
|
||||
const [row] = await tx.select({ agentId: heartbeatRuns.agentId, controllerProcess: heartbeatRuns.controllerProcess,
|
||||
controllerBootId: heartbeatRuns.controllerBootId })
|
||||
.from(heartbeatRuns).where(and(eq(heartbeatRuns.id, record.recordId), eq(heartbeatRuns.companyId, companyId),
|
||||
eq(heartbeatRuns.runtimeMode, "legacy"), eq(heartbeatRuns.status, "running"),
|
||||
eq(heartbeatRuns.controllerBootId, bootId), gt(heartbeatRuns.controllerLeaseExpiresAt, sql`clock_timestamp()`)))
|
||||
.for("update");
|
||||
if (!row) return { settled: false, reason: "lease_no_longer_held" };
|
||||
const current = legacyControllerOwnerStamp(row);
|
||||
// The proof judged this exact process; the lease has moved on since.
|
||||
if (!current || current.pid !== stamp.pid || current.processSpace !== stamp.processSpace
|
||||
|| (current.processStartedAt ?? null) !== (stamp.processStartedAt ?? null)) {
|
||||
return { settled: false, reason: "owner_changed_since_proof" };
|
||||
}
|
||||
// Work intent before the write it announces (sleep-aware background work).
|
||||
await notifyLiveRunRecovery(tx, { companyId, runId: record.recordId, agentId: row.agentId });
|
||||
const updated = await tx.update(heartbeatRuns).set({ controllerLeaseExpiresAt: sql`clock_timestamp()` })
|
||||
.where(and(eq(heartbeatRuns.id, record.recordId), eq(heartbeatRuns.companyId, companyId),
|
||||
eq(heartbeatRuns.status, "running"), eq(heartbeatRuns.controllerBootId, bootId),
|
||||
gt(heartbeatRuns.controllerLeaseExpiresAt, sql`clock_timestamp()`)))
|
||||
.returning({ id: heartbeatRuns.id });
|
||||
return updated.length ? { settled: true, reason: context.proof.reason } : { settled: false, reason: "lease_no_longer_held" };
|
||||
});
|
||||
}
|
||||
|
||||
export async function revokeExpiredLegacyController(db: Db, run: Run): Promise<boolean> {
|
||||
return (await revokeExpiredLegacyControllerWithState(db, run)) !== null;
|
||||
}
|
||||
|
||||
@@ -32,6 +32,7 @@ import {
|
||||
settleNativeAcceptedResultLease,
|
||||
} from "../native-runtime/native-accepted-result-lease-settlement.js";
|
||||
import { settleClaimedWakeRequest } from "./wake-request-claimed-settlement.js";
|
||||
import { legacyControllerOwnerStamp, settleLegacyControllerLease } from "../legacy-controller-lease.js";
|
||||
import {
|
||||
nativeLeaseHolderStamp,
|
||||
readNativeLeaseHolder,
|
||||
@@ -208,7 +209,7 @@ function nrfTokenLeaseOwner(row: NrfRow, scope: NativeTokenLeaseKind): RecordOwn
|
||||
|
||||
type RunRow = Pick<typeof heartbeatRuns.$inferSelect, "id" | "companyId" | "issueId" | "status" | "controllerBootId"
|
||||
| "controllerLeaseExpiresAt" | "processPid" | "processGroupId" | "processStartedAt" | "processHostHash" | "updatedAt"
|
||||
| "runnerProfileJson">;
|
||||
| "runnerProfileJson" | "controllerProcess">;
|
||||
|
||||
const runColumns = {
|
||||
id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, issueId: heartbeatRuns.issueId, status: heartbeatRuns.status,
|
||||
@@ -217,6 +218,7 @@ const runColumns = {
|
||||
processStartedAt: heartbeatRuns.processStartedAt, processHostHash: heartbeatRuns.processHostHash,
|
||||
updatedAt: heartbeatRuns.updatedAt,
|
||||
runnerProfileJson: heartbeatRuns.runnerProfileJson,
|
||||
controllerProcess: heartbeatRuns.controllerProcess,
|
||||
};
|
||||
|
||||
function runKind(id: string, title: string, obligation: Obligation, where: SQL | undefined,
|
||||
@@ -245,6 +247,17 @@ function runProcessStamp(row: RunRow, scope: string): LocalOwnerStamp {
|
||||
};
|
||||
}
|
||||
|
||||
const legacyControllerLeaseWhere = and(eq(heartbeatRuns.runtimeMode, "legacy"), eq(heartbeatRuns.status, "running"),
|
||||
isNotNull(heartbeatRuns.controllerBootId), sql`${heartbeatRuns.controllerLeaseExpiresAt} > now()`);
|
||||
|
||||
function legacyControllerLeaseOwner(row: RunRow): RecordOwner {
|
||||
const stamp = legacyControllerOwnerStamp(row);
|
||||
if (stamp) return { kind: "process", stamp };
|
||||
return { kind: "fence", stamp: { kind: "remote", scope: "legacy_controller_lease", generation: null,
|
||||
fenceToken: row.controllerBootId },
|
||||
fence: { currentFenceToken: row.controllerBootId, currentGeneration: null, leaseExpiresAt: row.controllerLeaseExpiresAt } };
|
||||
}
|
||||
|
||||
// ----------------------------------------------------------------------------
|
||||
// environment_leases
|
||||
|
||||
@@ -775,10 +788,27 @@ export const OWNABLE_KINDS: readonly OwnableKind[] = [
|
||||
})) };
|
||||
},
|
||||
},
|
||||
runKind("legacy_controller_lease", "Legacy controller lease", "ownership",
|
||||
and(eq(heartbeatRuns.runtimeMode, "legacy"), eq(heartbeatRuns.status, "running"), isNotNull(heartbeatRuns.controllerBootId)),
|
||||
row => ({ kind: "process", stamp: { kind: "local", scope: "legacy_controller_lease", generation: null, host: null,
|
||||
processSpace: null, bootId: row.controllerBootId, pid: null, processStartedAt: null } })),
|
||||
{
|
||||
// Shared path since Phase 2: the claim records its controller process
|
||||
// (`controller_process`, bound to the boot id); settle expires the lease of
|
||||
// a controller proven gone, and live-run recovery settles the run now
|
||||
// instead of after lease expiry plus the age guard. Outstanding leases
|
||||
// only. Rows saved before that column, and revoked leases, stay boot-id
|
||||
// fences: alive while leased, then handled by the existing revocation.
|
||||
...runKind("legacy_controller_lease", "Legacy controller lease", "ownership", legacyControllerLeaseWhere,
|
||||
legacyControllerLeaseOwner),
|
||||
listForRuns: async (db, runIds) => {
|
||||
const records: InProgressRecord[] = [];
|
||||
for (let index = 0; index < runIds.length; index += 500) {
|
||||
const rows = await db.select(runColumns).from(heartbeatRuns)
|
||||
.where(and(legacyControllerLeaseWhere, inArray(heartbeatRuns.id, [...runIds.slice(index, index + 500)])));
|
||||
records.push(...(rows as RunRow[]).map(row => ({ recordId: row.id, companyId: row.companyId, runId: row.id,
|
||||
issueId: row.issueId, owner: legacyControllerLeaseOwner(row), updatedAt: row.updatedAt })));
|
||||
}
|
||||
return records;
|
||||
},
|
||||
settle: settleLegacyControllerLease,
|
||||
},
|
||||
runKind("detached_process_marker", "Detached-process marker", "ownership",
|
||||
and(eq(heartbeatRuns.status, "running"), eq(heartbeatRuns.errorCode, "process_detached")),
|
||||
row => row.processPid ? { kind: "process", stamp: runProcessStamp(row, "detached_process_marker") } : { kind: "none" },
|
||||
|
||||
Reference in new issue
Block a user