mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
feat(recovery): settle teardown claims of dead claimants on the shared ownership path (#19966)
<!-- Write all pull request text in Simplified Technical English (ASD-STE100): short sentences, one instruction per sentence, simple approved vocabulary, and the active voice. --> ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - A sandbox teardown writes a claim on its environment lease that fences other cleanup for 15 minutes > - After a restart, the cleanup sweep ended the fence of a claim whose claimant had exited, with its own process check (`releaseLostPendingCleanupClaims`) > - Phase 2 of the durable ownership plan moves each in-progress record kind onto one shared owner proof and deletes its bespoke check > - This pull request registers a settle for `environment_teardown_claim` and calls the shared path where the sweep called the bespoke release > - The benefit is one proof for every kind: a local claimant proves death by its stamp; another host stays a fence handoff ## Linked Issues or Issue Description Refs #19965 Refs #19964 Refs #19933 **What happened?** The teardown claim had its own liveness check (`controllerProcessExited`) in the pending-cleanup sweep. Each kind with its own check can drift from the shared rules (PID space, PID reuse, current boot). **Expected behavior** The shared prover judges the claimant stamped for this exact attempt. On gone, the fence ends now and a counted sweep attempt is refunded once. A claim held as a fence (another claimant or another host) waits for its deadline. **Steps to reproduce** 1. Start a sandbox teardown so the lease holds an in-flight claim. 2. Kill the server. 3. Restart. The claim must be released by the shared ownership path, not by a separate check. ## What Changed - `environment-teardown-claim-settlement.ts` (new): compare-and-swap settle on the exact attempt and claimant. It ends the fence, refunds a counted attempt, and is single-use. It never acts on a fence owner. - `recovery/ownership-registry.ts`: `environment_teardown_claim` registers the settle. - `heartbeat/recovery.ts`: the sweep calls `settleOwnableKind(db, "environment_teardown_claim")`. Deleted `releaseLostPendingCleanupClaims` and its use of `controllerProcessExited`. - `environment-cleanup-state.ts`: the attempt cap, attempt key and their SQL readers move here, so the settle and the sweep share one reader. - `heartbeat-pending-cleanup-sweep.test.ts`: the overlap barrier now holds the shared prover instead of the deleted check. ## Verification - `cd server && npx vitest run src/__tests__/heartbeat-pending-cleanup-sweep.test.ts src/__tests__/environment-lease-owner-stamps.test.ts src/__tests__/ownership-reconciliation-report.test.ts` (embedded Postgres). - The existing lost-claim matrix stays green on the shared path: retry now; no failed-attempt charge for a dead server; one refund when sweeps overlap; keep a live claim; keep a newer claim that kept an older record; keep another host's claim; keep a claim without identity; page past fenced claims. - New, red before and green here (written after the code): the reconciler pass ends a dead claimant's fence and refunds its counted attempt. - New counter-cases (green): a read-only pass never settles; a live claimant, another PID space, and an unstamped fence claim keep their fence. - Full typecheck is left to CI (local memory limit). ## Risks - Low risk. The settle keeps the previous release semantics (fence ends, lease stays `pending_cleanup` and in flight until the sweep claims it again). - The stop-only claim still does not renew its fence; a live stop that runs longer than 15 minutes can still be taken over. This change does not alter that. - Stacked on #19965 and #19964. Merge those first. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, via Claude Code with tool use (shell, file edit, embedded Postgres tests). ## 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
65f41593d1
commit
bf0d45a4d2
7 files changed
+220
-111
No files matched your search
@@ -1,12 +1,19 @@
|
||||
import { spawn } from "node:child_process";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { agents, companies, createDb, environmentLeases, environments } from "@paperclipai/db";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import { currentControllerProcessStamp, currentProcessSpaceHash } from "../services/controller-process-identity.js";
|
||||
import { ENVIRONMENT_LEASE_CONTROLLER_KEY } from "../services/environment-lease-controller.js";
|
||||
import { environmentService } from "../services/environments.js";
|
||||
import { createOwnershipReportState, runOwnershipReconciliationPass } from "../services/recovery/ownership-invariant-report.js";
|
||||
import { nextEnvironmentCleanupAt } from "../services/environment-cleanup-state.js";
|
||||
import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-work-notifications.js";
|
||||
import {
|
||||
createOwnershipReportState,
|
||||
recheckSettlingKinds,
|
||||
runOwnershipReconciliationPass,
|
||||
} from "../services/recovery/ownership-invariant-report.js";
|
||||
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
const suite = support.supported ? describe : describe.skip;
|
||||
@@ -50,6 +57,11 @@ suite("environment lease owner stamps (login/probe lease, teardown claim)", () =
|
||||
return { kind: result.kinds.find(entry => entry.kind === kind)!, violations: result.violations.filter(entry => entry.kind === kind) };
|
||||
}
|
||||
|
||||
async function lease(id: string) {
|
||||
const [found] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, id));
|
||||
return found!;
|
||||
}
|
||||
|
||||
async function claim(metadata: Record<string, unknown>) {
|
||||
const id = randomUUID();
|
||||
await db.insert(environmentLeases).values({ id, companyId, environmentId, status: "pending_cleanup",
|
||||
@@ -107,6 +119,61 @@ suite("environment lease owner stamps (login/probe lease, teardown claim)", () =
|
||||
expect(kind).toMatchObject({ verdicts: { unprovable: 1 }, reasons: { fence_expired_without_provider_proof: 1 } });
|
||||
});
|
||||
|
||||
it("ends a dead claimant's fence on the next reconciler pass, refunding its counted attempt", async () => {
|
||||
const attemptId = randomUUID();
|
||||
const leaseId = await claim({ pendingCleanupAttemptId: attemptId, pendingCleanupLeaseExpiresAtMs: Date.now() + 14 * 60_000,
|
||||
pendingCleanupRetryAttempts: 3, pendingCleanupController: { bootId: "earlier-boot", pid: await exitedPid(),
|
||||
hostHash: currentProcessSpaceHash(), processStartedAt: null, attemptId, countedAttempt: true } });
|
||||
// A read-only pass never settles.
|
||||
await report("environment_teardown_claim");
|
||||
expect((await lease(leaseId)).metadata?.pendingCleanupLeaseExpiresAtMs).toBeGreaterThan(Date.now());
|
||||
const result = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState() });
|
||||
expect(result.kinds.find(entry => entry.kind === "environment_teardown_claim")).toMatchObject({ settled: 1 });
|
||||
const settled = (await lease(leaseId)).metadata!;
|
||||
expect(settled.pendingCleanupLeaseExpiresAtMs).toBeLessThanOrEqual(Date.now());
|
||||
expect(settled).toMatchObject({ pendingCleanupInFlight: true, pendingCleanupAttemptId: attemptId,
|
||||
pendingCleanupRetryAttempts: 2, pendingCleanupController: { countedAttempt: false } });
|
||||
});
|
||||
|
||||
it("wakes the cleanup worker for the ended fence, and a malformed flag beside it never blocks the settle", async () => {
|
||||
const attemptId = randomUUID();
|
||||
const malformed = await claim({ pendingCleanupInFlight: "maybe", pendingCleanupAttemptId: randomUUID(),
|
||||
pendingCleanupRetryAfterMs: Date.now() + 20 * 60_000 });
|
||||
const leaseId = await claim({ pendingCleanupAttemptId: attemptId, pendingCleanupLeaseExpiresAtMs: Date.now() + 14 * 60_000,
|
||||
pendingCleanupController: { bootId: "earlier-boot", pid: await exitedPid(), hostHash: currentProcessSpaceHash(),
|
||||
processStartedAt: null, attemptId } });
|
||||
expect(await nextEnvironmentCleanupAt(db)).toBeGreaterThan(Date.now() + 10 * 60_000);
|
||||
let wakes = 0;
|
||||
const unsubscribe = subscribeDeliveryWork(db, DELIVERY_QUEUES.environmentCleanup, () => { wakes += 1; });
|
||||
try {
|
||||
const result = await recheckSettlingKinds(db);
|
||||
expect(result.settled.map(record => record.recordId)).toEqual([leaseId]);
|
||||
await vi.waitFor(() => expect(wakes).toBeGreaterThan(0));
|
||||
} finally {
|
||||
unsubscribe();
|
||||
}
|
||||
// The worker's next run follows the ended fence, not the old deadline.
|
||||
expect(await nextEnvironmentCleanupAt(db)).toBeLessThan(Date.now() + 10 * 60_000);
|
||||
expect((await lease(malformed)).metadata?.pendingCleanupInFlight).toBe("maybe");
|
||||
});
|
||||
|
||||
it("never ends the fence of a live claimant, another PID space, or a claim held as a fence", async () => {
|
||||
const live = randomUUID(), foreign = randomUUID(), fenced = randomUUID();
|
||||
const expires = Date.now() + 14 * 60_000;
|
||||
const ids = [
|
||||
await claim({ pendingCleanupAttemptId: live, pendingCleanupLeaseExpiresAtMs: expires,
|
||||
pendingCleanupController: { ...currentControllerProcessStamp(), attemptId: live } }),
|
||||
await claim({ pendingCleanupAttemptId: foreign, pendingCleanupLeaseExpiresAtMs: expires,
|
||||
pendingCleanupController: { bootId: "other", pid: await exitedPid(), hostHash: "f".repeat(32), processStartedAt: null,
|
||||
attemptId: foreign } }),
|
||||
await claim({ pendingCleanupAttemptId: fenced, pendingCleanupLeaseExpiresAtMs: expires }),
|
||||
];
|
||||
const result = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState() });
|
||||
expect(result.kinds.find(entry => entry.kind === "environment_teardown_claim")).toMatchObject({ settled: 0,
|
||||
reasons: { current_process: 1, different_process_space: 1, fence_held: 1 } });
|
||||
for (const id of ids) expect((await lease(id)).metadata?.pendingCleanupLeaseExpiresAtMs).toBe(expires);
|
||||
});
|
||||
|
||||
it("names no owner for an in-flight flag without an attempt", async () => {
|
||||
await claim({});
|
||||
const { kind, violations } = await report("environment_teardown_claim");
|
||||
|
||||
@@ -21,12 +21,12 @@ import {
|
||||
const mockTelemetryClient = vi.hoisted(() => ({ track: vi.fn() }));
|
||||
vi.mock("../telemetry.ts", () => ({ getTelemetryClient: () => mockTelemetryClient }));
|
||||
|
||||
// A test can hold every caller of the exited-process proof until N callers
|
||||
// A test can hold every caller of the shared owner proof until N callers
|
||||
// arrive, so overlapping sweeps all read a claim before any of them updates it.
|
||||
const exitProofBarrier = vi.hoisted(() => ({ parties: 0, waiting: [] as Array<() => void> }));
|
||||
vi.mock("../services/controller-process-identity.js", async importOriginal => {
|
||||
const actual = await importOriginal<typeof import("../services/controller-process-identity.js")>();
|
||||
return { ...actual, controllerProcessExited: async (...args: Parameters<typeof actual.controllerProcessExited>) => {
|
||||
vi.mock("../services/owner-proof.js", async importOriginal => {
|
||||
const actual = await importOriginal<typeof import("../services/owner-proof.js")>();
|
||||
const proveOwnerGone = async (...args: unknown[]) => {
|
||||
if (exitProofBarrier.parties > 1) {
|
||||
await new Promise<void>(resolve => {
|
||||
exitProofBarrier.waiting.push(resolve);
|
||||
@@ -36,8 +36,9 @@ vi.mock("../services/controller-process-identity.js", async importOriginal => {
|
||||
}
|
||||
});
|
||||
}
|
||||
return actual.controllerProcessExited(...args);
|
||||
} };
|
||||
return (actual.proveOwnerGone as (...values: unknown[]) => ReturnType<typeof actual.proveOwnerGone>)(...args);
|
||||
};
|
||||
return { ...actual, proveOwnerGone };
|
||||
});
|
||||
|
||||
vi.mock("../middleware/logger.js", () => ({
|
||||
@@ -423,6 +424,8 @@ describeEmbeddedPostgres("heartbeat sweepPendingCleanupLeases", () => {
|
||||
heartbeatService(db, { environmentRuntime: runtime }).sweepPendingCleanupLeases(),
|
||||
heartbeatService(db, { environmentRuntime: runtime }).sweepPendingCleanupLeases(),
|
||||
]);
|
||||
// Both sweeps proved the claimant gone before either settled it.
|
||||
expect(exitProofBarrier.parties).toBe(0);
|
||||
// One refund (3 -> 2) and one new failed attempt (2 -> 3); never a second refund.
|
||||
expect(teardown).toHaveBeenCalledOnce();
|
||||
expect(await readAttempts(leaseId)).toBe(3);
|
||||
|
||||
@@ -142,7 +142,7 @@ describe("report-only ownership reconciliation", () => {
|
||||
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", "legacy_controller_lease"]);
|
||||
"environment_teardown_claim", "wake_request_claimed", "legacy_controller_lease"]);
|
||||
expect(OWNABLE_KINDS.filter(kind => !kind.listInProgress).every(kind => kind.coverageGap)).toBe(true);
|
||||
});
|
||||
|
||||
|
||||
@@ -4,6 +4,45 @@ import { hasDeferredEnvironmentCleanup } from "./environment-cleanup-notificatio
|
||||
|
||||
export const ENVIRONMENT_CLEANUP_RETRY_MS = 5 * 60_000;
|
||||
|
||||
// Escalate and slow cleanup after this many attempts; never abandon a live lease.
|
||||
export const PENDING_CLEANUP_SWEEP_ATTEMPT_CAP = 5;
|
||||
|
||||
// The reaper stores its retry attempt count under this key in the lease metadata.
|
||||
export const PENDING_CLEANUP_ATTEMPTS_METADATA_KEY = "pendingCleanupRetryAttempts";
|
||||
|
||||
// Read the stored retry attempt count as a safe value, directly in SQL. A
|
||||
// provider can write a malformed value under the attempts key. The type guard
|
||||
// makes any non-number value read as zero. The reader computes as numeric and
|
||||
// never casts to int, so a finite number outside the 32-bit range (for example
|
||||
// 1e300) never throws. The reader clamps a negative value to zero and a positive
|
||||
// value to the attempt cap. One malformed lease therefore never aborts the page
|
||||
// sweep. This matches the TypeScript reader `readPendingCleanupRetryAttempts`,
|
||||
// which clamps to the same range. The claim predicate compares the two readers,
|
||||
// so both must yield the same value for every input.
|
||||
export function pendingCleanupAttemptsSql() {
|
||||
return sql`
|
||||
case
|
||||
when jsonb_typeof(${environmentLeases.metadata} -> ${PENDING_CLEANUP_ATTEMPTS_METADATA_KEY}) = 'number'
|
||||
then least(
|
||||
greatest(
|
||||
floor((${environmentLeases.metadata} ->> ${PENDING_CLEANUP_ATTEMPTS_METADATA_KEY})::numeric),
|
||||
0
|
||||
),
|
||||
${PENDING_CLEANUP_SWEEP_ATTEMPT_CAP}
|
||||
)
|
||||
else 0
|
||||
end`;
|
||||
}
|
||||
|
||||
// Choose the `jsonb_set` target root. A provider can write a scalar or array
|
||||
// metadata root. `jsonb_set` fails on a non-object root, so the reader uses the
|
||||
// stored metadata only when its root is an object. A NULL, scalar, or array root
|
||||
// reads as an empty object. `jsonb_typeof(NULL)` is NULL, so the else branch also
|
||||
// covers a NULL root.
|
||||
export function pendingCleanupMetadataObjectSql() {
|
||||
return sql`case when jsonb_typeof(${environmentLeases.metadata}) = 'object' then ${environmentLeases.metadata} else '{}'::jsonb end`;
|
||||
}
|
||||
|
||||
export function recoverableSandboxCreationCondition() {
|
||||
return or(eq(sandboxCreationOperations.state, "pending"), and(
|
||||
eq(sandboxCreationOperations.state, "needs_reconciliation"),
|
||||
|
||||
@@ -0,0 +1,74 @@
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { environmentLeases, type Db } from "@paperclipai/db";
|
||||
import {
|
||||
PENDING_CLEANUP_ATTEMPTS_METADATA_KEY,
|
||||
pendingCleanupAttemptsSql,
|
||||
pendingCleanupMetadataObjectSql,
|
||||
} from "./environment-cleanup-state.js";
|
||||
import { notifyEnvironmentCleanup } from "./environment-cleanup-notifications.js";
|
||||
import type { OwnerProof } from "./owner-proof.js";
|
||||
import type { InProgressRecord, SettleOutcome } from "./recovery/ownership-registry.js";
|
||||
|
||||
/**
|
||||
* Settle of the `environment_teardown_claim` ownership kind (plan Phase 2).
|
||||
*
|
||||
* A teardown claim fences other cleanup for 15 minutes while its claimant
|
||||
* destroys or stops a sandbox. A claimant process that died cannot still be
|
||||
* running the teardown, so once the shared prover proves the claimant stamped
|
||||
* for this exact attempt gone, the fence ends now and the pending-cleanup
|
||||
* sweep may claim the lease again. A dead server is not a failed teardown: a
|
||||
* sweep claim that counted its attempt gets it back.
|
||||
*
|
||||
* The reconciler calls this only for a process owner. A claim held as a fence
|
||||
* (another claimant, or another host) is never proven gone by a process proof;
|
||||
* it waits for its deadline, which is the remote handoff.
|
||||
*
|
||||
* The update compares and swaps on the exact attempt and claimant the proof
|
||||
* judged, and it is single-use: it matches only while the fence is still in
|
||||
* the future and it consumes `countedAttempt`, so an overlapping settle cannot
|
||||
* refund the attempt twice. It settles ownership only; whether the sandbox
|
||||
* still owes a stop, a destroy or a retained workspace is unchanged.
|
||||
*/
|
||||
export async function settleEnvironmentTeardownClaim(db: Db, record: InProgressRecord,
|
||||
context: { now: Date; proof: OwnerProof }): Promise<SettleOutcome> {
|
||||
if (record.owner.kind !== "process") return { settled: false, reason: "claim_held_as_fence" };
|
||||
const stamp = record.owner.stamp;
|
||||
const nowMs = context.now.getTime();
|
||||
return db.transaction(async transaction => {
|
||||
const tx = transaction as unknown as Db;
|
||||
const [lease] = await tx.select({ metadata: environmentLeases.metadata }).from(environmentLeases)
|
||||
.where(and(eq(environmentLeases.id, record.recordId), eq(environmentLeases.status, "pending_cleanup")))
|
||||
.for("update");
|
||||
const metadata = lease?.metadata ?? null;
|
||||
const attemptId = typeof metadata?.pendingCleanupAttemptId === "string" ? metadata.pendingCleanupAttemptId : null;
|
||||
const controller = metadata?.pendingCleanupController as Record<string, unknown> | undefined;
|
||||
if (!attemptId || metadata?.pendingCleanupInFlight !== true) return { settled: false, reason: "claim_no_longer_in_flight" };
|
||||
if (controller?.attemptId !== attemptId || controller.bootId !== stamp.bootId || controller.pid !== stamp.pid
|
||||
|| controller.hostHash !== stamp.processSpace) {
|
||||
return { settled: false, reason: "owner_changed_since_proof" };
|
||||
}
|
||||
const expiresAtMs = metadata?.pendingCleanupLeaseExpiresAtMs;
|
||||
if (typeof expiresAtMs !== "number" || expiresAtMs <= nowMs) return { settled: false, reason: "fence_already_ended" };
|
||||
const returnAttempt = controller?.countedAttempt === true;
|
||||
// Wake the cleanup worker before the write it announces, so it reads the
|
||||
// ended fence after commit instead of sleeping until the old deadline.
|
||||
if (record.companyId) await notifyEnvironmentCleanup(tx, { companyId: record.companyId, leaseId: record.recordId });
|
||||
const updated = await tx.update(environmentLeases)
|
||||
.set({ metadata: sql`jsonb_set(jsonb_set(jsonb_set(${pendingCleanupMetadataObjectSql()},
|
||||
'{pendingCleanupLeaseExpiresAtMs}', to_jsonb(${nowMs}::bigint)),
|
||||
array[${PENDING_CLEANUP_ATTEMPTS_METADATA_KEY}],
|
||||
to_jsonb(greatest(${pendingCleanupAttemptsSql()} - ${returnAttempt ? 1 : 0}, 0)::int), true),
|
||||
'{pendingCleanupController,countedAttempt}', 'false'::jsonb, true)` })
|
||||
.where(and(
|
||||
eq(environmentLeases.id, record.recordId),
|
||||
eq(environmentLeases.status, "pending_cleanup"),
|
||||
sql`${environmentLeases.metadata}->>'pendingCleanupInFlight' = 'true'`,
|
||||
sql`(${environmentLeases.metadata}->>'pendingCleanupLeaseExpiresAtMs')::numeric > ${nowMs}`,
|
||||
sql`${environmentLeases.metadata}->>'pendingCleanupAttemptId' = ${attemptId}`,
|
||||
sql`${environmentLeases.metadata}->'pendingCleanupController'->>'attemptId' = ${attemptId}`,
|
||||
sql`coalesce(${environmentLeases.metadata}->'pendingCleanupController'->>'countedAttempt', 'false') = ${returnAttempt ? "true" : "false"}`,
|
||||
))
|
||||
.returning({ id: environmentLeases.id });
|
||||
return updated.length ? { settled: true, reason: context.proof.reason } : { settled: false, reason: "claim_no_longer_in_flight" };
|
||||
});
|
||||
}
|
||||
@@ -12,8 +12,15 @@ import { attemptNativeWorkspaceFinalization } from "../native-runtime/native-wor
|
||||
import { retainedNativeSessionCleanupWork } from "../native-runtime/retained-native-session-cleanup-work.js";
|
||||
import { reconcileRetainedNativeSessionCleanups } from "../native-runtime/retained-native-session-cleanup.js";
|
||||
import { createScheduledRetryCancellationRecovery } from "./scheduled-retry-cancellation.js";
|
||||
import { pendingCleanupDeadlineSql, recoveredOrphanFirstAttemptSql } from "../environment-cleanup-state.js";
|
||||
import { controllerProcessExited, currentControllerBootId, currentControllerProcess, readControllerProcess } from "../controller-process-identity.js";
|
||||
import {
|
||||
PENDING_CLEANUP_ATTEMPTS_METADATA_KEY,
|
||||
PENDING_CLEANUP_SWEEP_ATTEMPT_CAP,
|
||||
pendingCleanupAttemptsSql,
|
||||
pendingCleanupDeadlineSql,
|
||||
pendingCleanupMetadataObjectSql,
|
||||
recoveredOrphanFirstAttemptSql,
|
||||
} from "../environment-cleanup-state.js";
|
||||
import { currentControllerProcess } from "../controller-process-identity.js";
|
||||
import { writeEnvironmentCleanup } from "../environment-cleanup-notifications.js";
|
||||
import { startNativeSessionExecutionLeaseRenewal, type NativeExecutionLeaseRenewal } from "../native-runtime/native-execution-lease.js";
|
||||
import type { WorkspaceWorkers } from "../plugin-workspaces.js";
|
||||
@@ -198,12 +205,8 @@ const ENVIRONMENT_CLEANUP_RETRY_LIMIT = 1_000;
|
||||
|
||||
const pendingCleanupAttemptsInFlight = new Set<string>();
|
||||
|
||||
// Escalate and slow cleanup after this many attempts; never abandon a live lease.
|
||||
const PENDING_CLEANUP_SWEEP_ATTEMPT_CAP = 5;
|
||||
|
||||
// The reaper stores its retry state under these keys in the lease metadata.
|
||||
const PENDING_CLEANUP_ATTEMPTS_METADATA_KEY = "pendingCleanupRetryAttempts";
|
||||
|
||||
// The reaper stores its retry state in the lease metadata (the attempt cap and
|
||||
// count key live in environment-cleanup-state.ts).
|
||||
const PENDING_CLEANUP_CAP_WARNED_METADATA_KEY = "pendingCleanupRetryCapWarned";
|
||||
|
||||
// The reaper sweeps at most this many orphaned active leases per tick.
|
||||
@@ -226,45 +229,12 @@ const PENDING_CLEANUP_SWEEP_ERROR_KIND = "sweep_failed";
|
||||
|
||||
const ORPHANED_ACTIVE_LEASE_SWEEP_ERROR_KIND = "orphaned_active_lease_sweep_failed";
|
||||
|
||||
// Read the stored retry attempt count as a safe value, directly in SQL. A
|
||||
// provider can write a malformed value under the attempts key. The type guard
|
||||
// makes any non-number value read as zero. The reader computes as numeric and
|
||||
// never casts to int, so a finite number outside the 32-bit range (for example
|
||||
// 1e300) never throws. The reader clamps a negative value to zero and a positive
|
||||
// value to the attempt cap. One malformed lease therefore never aborts the page
|
||||
// sweep. This matches the TypeScript reader `readPendingCleanupRetryAttempts`,
|
||||
// which clamps to the same range. The claim predicate compares the two readers,
|
||||
// so both must yield the same value for every input.
|
||||
function pendingCleanupAttemptsSql() {
|
||||
return sql`
|
||||
case
|
||||
when jsonb_typeof(${environmentLeases.metadata} -> ${PENDING_CLEANUP_ATTEMPTS_METADATA_KEY}) = 'number'
|
||||
then least(
|
||||
greatest(
|
||||
floor((${environmentLeases.metadata} ->> ${PENDING_CLEANUP_ATTEMPTS_METADATA_KEY})::numeric),
|
||||
0
|
||||
),
|
||||
${PENDING_CLEANUP_SWEEP_ATTEMPT_CAP}
|
||||
)
|
||||
else 0
|
||||
end`;
|
||||
}
|
||||
|
||||
function pendingCleanupRetryDueSql(explicitRetry = false) {
|
||||
return sql`case
|
||||
when ${explicitRetry} and coalesce(${environmentLeases.metadata}->>'pendingCleanupInFlight', 'false') != 'true' then true
|
||||
else ${pendingCleanupDeadlineSql()} <= ${Date.now()} end`;
|
||||
}
|
||||
|
||||
// Choose the `jsonb_set` target root. A provider can write a scalar or array
|
||||
// metadata root. `jsonb_set` fails on a non-object root, so the reader uses the
|
||||
// stored metadata only when its root is an object. A NULL, scalar, or array root
|
||||
// reads as an empty object. `jsonb_typeof(NULL)` is NULL, so the else branch also
|
||||
// covers a NULL root.
|
||||
function pendingCleanupMetadataObjectSql() {
|
||||
return sql`case when jsonb_typeof(${environmentLeases.metadata}) = 'object' then ${environmentLeases.metadata} else '{}'::jsonb end`;
|
||||
}
|
||||
|
||||
// Read the stored cap-warned flag as a safe boolean, directly in SQL. A
|
||||
// malformed value reads as false, so the boolean cast never throws.
|
||||
function pendingCleanupCapWarnedSql() {
|
||||
@@ -1449,62 +1419,6 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
|
||||
return Math.min(Math.floor(value), PENDING_CLEANUP_SWEEP_ATTEMPT_CAP);
|
||||
}
|
||||
|
||||
// A teardown claim fences other sweeps for 15 minutes. After a restart, a
|
||||
// claim whose server process has exited cannot still be running, so its
|
||||
// fence ends now. The saved process is bound to its own attempt: a later
|
||||
// claim from another path that kept the old record is never released. A live
|
||||
// or unidentified claimer keeps its fence. Pages pass every skipped claim.
|
||||
async function releaseLostPendingCleanupClaims(): Promise<void> {
|
||||
const nowMs = Date.now();
|
||||
let after: string | null = null;
|
||||
for (;;) {
|
||||
const claimed: Array<{ id: string; metadata: Record<string, unknown> | null }> = await db
|
||||
.select({ id: environmentLeases.id, metadata: environmentLeases.metadata })
|
||||
.from(environmentLeases)
|
||||
.where(and(
|
||||
eq(environmentLeases.status, "pending_cleanup"),
|
||||
sql`${environmentLeases.metadata}->>'pendingCleanupInFlight' = 'true'`,
|
||||
sql`jsonb_typeof(${environmentLeases.metadata}->'pendingCleanupLeaseExpiresAtMs') = 'number'`,
|
||||
sql`(${environmentLeases.metadata}->>'pendingCleanupLeaseExpiresAtMs')::numeric > ${nowMs}`,
|
||||
sql`${environmentLeases.metadata}->'pendingCleanupController'->>'attemptId' = ${environmentLeases.metadata}->>'pendingCleanupAttemptId'`,
|
||||
sql`${environmentLeases.metadata}->'pendingCleanupController'->>'bootId' is distinct from ${currentControllerBootId()}`,
|
||||
after === null ? undefined : gt(environmentLeases.id, after),
|
||||
))
|
||||
.orderBy(asc(environmentLeases.id))
|
||||
.limit(PENDING_CLEANUP_SWEEP_PAGE_SIZE);
|
||||
for (const row of claimed) {
|
||||
const metadata = row.metadata ?? {};
|
||||
const saved = metadata.pendingCleanupController as Record<string, unknown> | undefined;
|
||||
const controller = readControllerProcess(saved);
|
||||
const attemptId = saved?.attemptId;
|
||||
if (!controller || typeof attemptId !== "string" || !await controllerProcessExited(controller)) continue;
|
||||
// A dead server is not a failed teardown. A sweep claim already counted
|
||||
// its attempt, so give it back; claims from other paths never counted.
|
||||
// The update is single-use: it matches only while the fence is still in
|
||||
// the future and it consumes `countedAttempt`, so an overlapping sweep
|
||||
// that read the same claim cannot refund the attempt again.
|
||||
const returnAttempt = saved?.countedAttempt === true;
|
||||
await db.update(environmentLeases)
|
||||
.set({ metadata: sql`jsonb_set(jsonb_set(jsonb_set(${pendingCleanupMetadataObjectSql()},
|
||||
'{pendingCleanupLeaseExpiresAtMs}', to_jsonb(${nowMs}::bigint)),
|
||||
array[${PENDING_CLEANUP_ATTEMPTS_METADATA_KEY}],
|
||||
to_jsonb(greatest(${pendingCleanupAttemptsSql()} - ${returnAttempt ? 1 : 0}, 0)::int), true),
|
||||
'{pendingCleanupController,countedAttempt}', 'false'::jsonb, true)` })
|
||||
.where(and(
|
||||
eq(environmentLeases.id, row.id),
|
||||
eq(environmentLeases.status, "pending_cleanup"),
|
||||
sql`${environmentLeases.metadata}->>'pendingCleanupInFlight' = 'true'`,
|
||||
sql`(${environmentLeases.metadata}->>'pendingCleanupLeaseExpiresAtMs')::numeric > ${nowMs}`,
|
||||
sql`${environmentLeases.metadata}->>'pendingCleanupAttemptId' = ${attemptId}`,
|
||||
sql`${environmentLeases.metadata}->'pendingCleanupController'->>'attemptId' = ${attemptId}`,
|
||||
sql`coalesce(${environmentLeases.metadata}->'pendingCleanupController'->>'countedAttempt', 'false') = ${returnAttempt ? "true" : "false"}`,
|
||||
));
|
||||
}
|
||||
if (claimed.length < PENDING_CLEANUP_SWEEP_PAGE_SIZE) return;
|
||||
after = claimed[claimed.length - 1]!.id;
|
||||
}
|
||||
}
|
||||
|
||||
// Atomically claim one retry attempt on a pending_cleanup lease. The update
|
||||
// only matches when the lease is still pending_cleanup and its stored attempt
|
||||
// count still equals `expectedAttempts`. Two concurrent sweeps read the same
|
||||
@@ -1798,7 +1712,12 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
|
||||
deferredFlushFailed = true;
|
||||
}
|
||||
|
||||
await releaseLostPendingCleanupClaims();
|
||||
// A teardown claim whose claimant process is proven gone ends its fence
|
||||
// now, through the shared ownership path. A claim held as a fence (another
|
||||
// host, or a claimant without a process stamp) waits for its deadline.
|
||||
await settleOwnableKind(db, "environment_teardown_claim").catch(() => {
|
||||
logger.warn({ errorKind: "teardown_claim_settle_failed" }, "teardown claim settle failed; a later sweep retries");
|
||||
});
|
||||
|
||||
const rows = await db
|
||||
.select()
|
||||
|
||||
@@ -42,6 +42,7 @@ import {
|
||||
import { ephemeralServerEnabled } from "../ephemeral-server-mode.js";
|
||||
import { ENVIRONMENT_LEASE_CONTROLLER_KEY } from "../environment-lease-controller.js";
|
||||
import { settleEnvironmentRunLease } from "../environment-run-lease-settlement.js";
|
||||
import { settleEnvironmentTeardownClaim } from "../environment-teardown-claim-settlement.js";
|
||||
|
||||
/**
|
||||
* Registry of ownable record kinds (plan section C).
|
||||
@@ -593,12 +594,18 @@ export const OWNABLE_KINDS: readonly OwnableKind[] = [
|
||||
},
|
||||
},
|
||||
runLeaseKind(),
|
||||
leaseKind("environment_teardown_claim", "Teardown claim (in-flight cleanup)", "ownership",
|
||||
and(eq(environmentLeases.status, "pending_cleanup"), sql`(${environmentLeases.metadata}->>'pendingCleanupInFlight')::boolean is true`),
|
||||
teardownClaimOwner),
|
||||
{
|
||||
// Shared path since Phase 2: a claimant proven gone ends its fence now.
|
||||
...leaseKind("environment_teardown_claim", "Teardown claim (in-flight cleanup)", "ownership",
|
||||
// Text compare: a malformed flag never makes the whole listing fail.
|
||||
and(eq(environmentLeases.status, "pending_cleanup"),
|
||||
sql`coalesce(${environmentLeases.metadata}->>'pendingCleanupInFlight', 'false') = 'true'`),
|
||||
teardownClaimOwner),
|
||||
settle: settleEnvironmentTeardownClaim,
|
||||
},
|
||||
leaseKind("environment_pending_cleanup", "Pending-cleanup obligation (not in flight)", "ownership",
|
||||
and(eq(environmentLeases.status, "pending_cleanup"),
|
||||
sql`coalesce((${environmentLeases.metadata}->>'pendingCleanupInFlight')::boolean, false) is false`),
|
||||
sql`coalesce(${environmentLeases.metadata}->>'pendingCleanupInFlight', 'false') <> 'true'`),
|
||||
() => ({ kind: "none" }), [1, 2, 7]),
|
||||
leaseKind("environment_stop_and_retain_intent", "Stop-and-retain intent", "workspace_preservation",
|
||||
and(eq(environmentLeases.status, "pending_cleanup"), sql`${environmentLeases.metadata} ? 'sandboxStopAndRetain'`),
|
||||
|
||||
Reference in new issue
Block a user