mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-12 22:56:58 +01:00
feat(recovery): prove a replaced container's records gone through a fenced controller epoch (#20054)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - After a restart, the control plane settles in-progress records (leases, claims, intents) only once it proves their owner process is gone > - The proof compares host and PID space, so a stamp from a replaced container or a rebooted machine stays unprovable forever, and its records never settle > - Self-hosted users on Docker, Kubernetes and Fly hit this on every redeploy; a Railway deployment with an exclusive volume is covered by `deployment_replaced` (#20026) > - This pull request adds a fenced controller epoch: each boot holds a database lease, and an expired boot is fenced so its records can be proven gone > - The benefit is that work stranded by a replaced container is recovered, without keeping an idle database awake ## Linked Issues or Issue Description Follow-up to #20026 and the Phase 2 durable-ownership data. Related: #19965 (owner stamps), #19883 (process space). No duplicate PR was found. **What happened?** A record stamped by a server process in another container reports `different_process_space`. A dead PID elsewhere is never proof, so the record stays `unprovable` and is never settled. Every redeploy of a self-hosted container leaves such records behind. **Expected behavior** A record owned by a server boot that was replaced (stopped cleanly, or lost its lease and was fenced) is proven gone and settled exactly once. A live boot's records stay with it. An idle server issues no recurring queries. **Steps to reproduce** 1. Run Paperclip in a container and start work, so an environment lease is stamped by that boot. 2. Replace the container. The new container has a new PID namespace. 3. The ownership reconciler reports `different_process_space` (unprovable) for the old lease, forever. **Paperclip version or commit** `master` before this PR. ## What Changed - New `controller_instances` table (migration 0449): boot id, epoch (bigserial), process space, pid, a lease on the database clock, `fenced_at` / `fenced_by_epoch`, `stopped_at`. - `controller-instance-lease.ts`, registered with the shared delivery-work coordinator (`controller-instance-lease` queue, runs during drain): - **Holds work → renews; idle → silent.** One upsert creates the row on first use and extends the lease only if the boot holds work: a stamp within the last 90 s, or a stamp whose ownership write may still be pending (held until the server's tracked work settles, capped at 10 minutes; a renewal that sees some other held record does not clear it, so a slow ownership write cannot be outlived), or an in-progress record held by its boot id. Those records are active environment leases, in-flight cleanup claims, sandbox creation intents, workspace finalize attempts, agent lifecycle steps, native finalizer and cleanup leases (`lease_holder` bound to the token), open native coordinator claims, and running legacy runs. Renewal runs every 30 s while the boot holds work. When it finds none, the lease lapses and the worker sleeps until the next stamp (`onControllerStamp`). - **No idle trailing statements.** When the server's last tracked work settles, the worker re-checks holdings at once, at most every 5 s (the rate limit also absorbs the worker's own run). So the last check happens with the work, not 30 s later. - **Quiesce, don't crash.** Each renewal has a 10 s bound. A failed or timed-out renewal applies an internal task drain, so no new work is dispatched. A recovered renewal resumes. The boot exits only after 60 s of failure (counted from the first failed attempt's start) while it still holds work, or at once when it finds itself fenced. - **Fence on demand, by a newer epoch.** For a stamp from another host or PID space, the prover reads that boot's row: - `stopped` or `fenced`: gone. - live lease: alive. - otherwise, the boot's last sign of life is the later of its lease expiry and the record's last write plus the TTL. Within 30 s of that, the owner stays `alive` (`controller_lease_grace`), so the settle watch keeps rechecking. After that, the prover fences the boot, but only with a lease row of its own (created on demand) and only an older epoch. - A graceful shutdown records `stopped_at`. - A reconciliation pass that settles nothing (`record: false` or `settle: false`) never fences and never creates the boot's row. - `owner-proof.ts`: these proofs apply only to server-held records. A stamp with a recorded process group stays unprovable, because agent processes can outlive their server. The reconciler passes each record's `updatedAt` to the proof. - `task-admission.ts`: - the quiesce drain never replaces an operator's drain, and has no idle owner, so HTTP stays open; - a `subscribeIdleWorkSettled` hook fires when the last tracked work finishes. - Off on a Railway deployment with an exclusive volume (it has its own replacement proof). `PAPERCLIP_CONTROLLER_INSTANCE_LEASE=on|off` overrides this. A stateless Railway service keeps the lease. A server without its own lease row never fences. - `maintenance-server-routes.test.ts`: in the post-run quiet window, the lease may check once more that the run's records were released. The test now excludes those statements, then asserts that the lease issues nothing in a further window. - Docs: `doc/execution-semantics.md`; the Phase 2 data doc records the two known gaps. ## Verification - `controller-instance-lease.test.ts` (15 cases): - a replaced container with all Railway variables empty is fenced and settled exactly once over two passes; - a live lease is alive; - an owner in grace, and work written after an idle lapse, stay alive and watched; - an older epoch never fences a newer one; - no fencing without its own lease row (off or unregistered), and an idle prover creates its row on demand; - a stopped boot is gone; - process-group stamps stay unprovable; - idle issues zero statements, and one upsert renews only while records are held; - **a 45 s outage:** quiesced, no exit, resumes; - **a renewal that never answers:** bounded, and the boot exits on time; - **a 70 s outage:** exits at 60–65 s, and the successor recovers the work exactly once; - fenced: stops; - the settle re-check; - graceful stop; off on a volume-exclusive Railway deployment, on for a stateless one, and the on/off override; - a running legacy run keeps the lease live. - Red, with the proof's instance check removed: the replaced-container, live, stopped and successor cases fail. - Wider: task-drain, idle-sleep, idle-local-work, owner-proof and ownership suites, plus the maintenance quiet-window tests. Server low-memory typecheck is clean. ## Risks - **Clock skew.** The proof compares a record's `updatedAt` (often the writer's clock) with the lease (database clock). The 30 s grace absorbs ordinary skew. - **A new exit path exists.** A server whose database is unreachable for more than 60 s while it holds work now exits, after first quiescing. This is the intended safety backstop. - **Bounded trailing activity.** A lease that holds work issues one statement every 30 s, plus at most one check when its work settles. An idle server issues none. - **Boots from before this release** have no row, and their records stay unprovable. The board "mark owner gone" action for those follows in a separate PR. ## Model Used - Anthropic Claude Opus 5.5 (`claude-opus-5-5`), via Claude Code with tool use. ## 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
0a64c9b4ab
commit
e8a3f1f6f2
16 files changed
+6106
-5018
No files matched your search
@@ -0,0 +1,418 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { agents, companies, controllerInstances, createDb, environmentLeases, heartbeatRuns, issues, nativeRunFinalizations } from "@paperclipai/db";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import {
|
||||
controllerInstanceLeaseEnabled, proveControllerInstance, registerControllerInstanceDb, registerControllerInstanceLeaseWork,
|
||||
resetControllerInstanceLeaseForTests, stopControllerInstanceLease, tickControllerInstanceLease,
|
||||
} from "../services/controller-instance-lease.js";
|
||||
import { currentControllerBootId, currentControllerProcessStamp, currentProcessSpaceHash } from "../services/controller-process-identity.js";
|
||||
import type { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
|
||||
import { ENVIRONMENT_LEASE_CONTROLLER_KEY } from "../services/environment-lease-controller.js";
|
||||
import { proveOwnerGone, type LocalOwnerStamp } from "../services/owner-proof.js";
|
||||
import { createOwnershipReportState, runOwnershipReconciliationPass } from "../services/recovery/ownership-invariant-report.js";
|
||||
import { beginIdleTrackedWork, readTaskDrain } from "../services/task-admission.js";
|
||||
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
const suite = support.supported ? describe : describe.skip;
|
||||
|
||||
suite("controller instance lease: a replaced container's records are provable", () => {
|
||||
let db!: ReturnType<typeof createDb>;
|
||||
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
|
||||
let companyId = "";
|
||||
|
||||
beforeAll(async () => {
|
||||
temporary = await startEmbeddedPostgresTestDatabase("paperclip-controller-instance-lease-");
|
||||
db = createDb(temporary.connectionString);
|
||||
companyId = randomUUID();
|
||||
await db.insert(companies).values({ id: companyId, name: "Paperclip", issuePrefix: "CIL", requireBoardApprovalForNewAgents: false });
|
||||
}, 30_000);
|
||||
|
||||
afterEach(async () => {
|
||||
vi.unstubAllEnvs();
|
||||
resetControllerInstanceLeaseForTests();
|
||||
await db.delete(environmentLeases);
|
||||
await db.delete(controllerInstances);
|
||||
});
|
||||
|
||||
afterAll(async () => { await temporary?.cleanup(); });
|
||||
|
||||
const coordinator = { register: () => ({ ready: Promise.resolve(), wake: vi.fn(), stop: async () => {} }) } as unknown as
|
||||
ReturnType<typeof createDeliveryWorkCoordinator>;
|
||||
|
||||
/** A server boot in another container: its own PID space, no platform deployment identity. */
|
||||
async function otherBoot(lease: { expiredMsAgo?: number; liveForMs?: number; epoch?: number } = {}) {
|
||||
const bootId = `boot-${randomUUID()}`;
|
||||
const expires = lease.liveForMs !== undefined ? sql`now() + ${`${lease.liveForMs} milliseconds`}::interval`
|
||||
: sql`now() - ${`${lease.expiredMsAgo ?? 120_000} milliseconds`}::interval`;
|
||||
await db.insert(controllerInstances).values({ bootId, processSpace: "old-container", pid: 7, leaseExpiresAt: expires,
|
||||
...(lease.epoch !== undefined ? { epoch: lease.epoch } : {}) });
|
||||
return bootId;
|
||||
}
|
||||
|
||||
const minutesAgo = (minutes: number) => new Date(Date.now() - minutes * 60_000).toISOString();
|
||||
const stampOf = (bootId: string, extra: Partial<LocalOwnerStamp> = {}): LocalOwnerStamp => ({
|
||||
kind: "local", scope: "test", generation: null, host: null, processSpace: "old-container", bootId, pid: 7,
|
||||
processStartedAt: null, processGroupId: null, touchedAt: minutesAgo(10), ...extra });
|
||||
|
||||
/** Registers this boot's lease; a stamp plus one tick creates its row and epoch. */
|
||||
async function startHere(options: Parameters<typeof registerControllerInstanceLeaseWork>[2] = {}, stamp = true) {
|
||||
registerControllerInstanceDb(db);
|
||||
const lost = vi.fn();
|
||||
registerControllerInstanceLeaseWork(db, coordinator, { onLost: lost, ...options });
|
||||
if (stamp) { currentControllerProcessStamp(); await tickControllerInstanceLease(); }
|
||||
const [row] = await db.select().from(controllerInstances).where(eq(controllerInstances.bootId, currentControllerBootId()));
|
||||
return { lost, epoch: row?.epoch ?? null };
|
||||
}
|
||||
|
||||
it("proves a replaced container's records gone by fencing its boot, with no Railway environment", async () => {
|
||||
for (const key of ["RAILWAY_PROJECT_ID", "RAILWAY_ENVIRONMENT_ID", "RAILWAY_SERVICE_ID", "RAILWAY_DEPLOYMENT_ID"]) vi.stubEnv(key, "");
|
||||
expect(controllerInstanceLeaseEnabled()).toBe(true);
|
||||
const replaced = await otherBoot({ expiredMsAgo: 120_000 });
|
||||
const another = await otherBoot({ expiredMsAgo: 120_000 });
|
||||
const { epoch } = await startHere();
|
||||
expect(await proveOwnerGone(stampOf(replaced))).toEqual({ verdict: "gone", reason: "controller_fenced" });
|
||||
const [row] = await db.select().from(controllerInstances).where(eq(controllerInstances.bootId, replaced));
|
||||
expect(row).toMatchObject({ fencedByEpoch: epoch });
|
||||
|
||||
const leaseId = randomUUID();
|
||||
await db.insert(environmentLeases).values({ id: leaseId, companyId, heartbeatRunId: null, status: "active",
|
||||
leasePolicy: "ephemeral", provider: "fake", providerLeaseId: `sandbox://fake/${leaseId}`, updatedAt: new Date(minutesAgo(10)),
|
||||
metadata: { [ENVIRONMENT_LEASE_CONTROLLER_KEY]: { bootId: another, pid: 7, hostHash: "old-container", processStartedAt: null } } });
|
||||
const first = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState() });
|
||||
expect(first.kinds.find(kind => kind.kind === "environment_login_probe_lease")).toMatchObject({
|
||||
settled: 1, reasons: { controller_fenced: 1 } });
|
||||
const second = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState() });
|
||||
expect(second.kinds.find(kind => kind.kind === "environment_login_probe_lease")).toMatchObject({ settled: 0 });
|
||||
});
|
||||
|
||||
it("treats a live lease as alive", async () => {
|
||||
const live = await otherBoot({ liveForMs: 60_000 });
|
||||
registerControllerInstanceDb(db);
|
||||
expect(await proveOwnerGone(stampOf(live))).toEqual({ verdict: "alive", reason: "controller_lease_held" });
|
||||
});
|
||||
|
||||
it("keeps an owner alive (and watched) through the grace, and for work written after an idle lapse", async () => {
|
||||
const justExpired = await otherBoot({ expiredMsAgo: 5_000 });
|
||||
const resumed = await otherBoot({ expiredMsAgo: 600_000 });
|
||||
await startHere();
|
||||
expect(await proveOwnerGone(stampOf(justExpired))).toEqual({ verdict: "alive", reason: "controller_lease_grace" });
|
||||
expect(await proveOwnerGone(stampOf(resumed, { touchedAt: minutesAgo(1) }))).toEqual({ verdict: "alive", reason: "controller_lease_grace" });
|
||||
expect(await proveOwnerGone(stampOf(resumed, { touchedAt: null }))).toEqual({ verdict: "unprovable", reason: "different_process_space" });
|
||||
for (const bootId of [justExpired, resumed]) expect(await proveControllerInstance(db, bootId, null)).toBe("expired");
|
||||
});
|
||||
|
||||
it("never lets an older epoch fence a newer one", async () => {
|
||||
await startHere();
|
||||
const newer = await otherBoot({ expiredMsAgo: 600_000, epoch: 9_000_000 });
|
||||
expect(await proveOwnerGone(stampOf(newer))).toEqual({ verdict: "unprovable", reason: "different_process_space" });
|
||||
expect(await proveControllerInstance(db, newer, null)).toBe("expired");
|
||||
});
|
||||
|
||||
it("never fences without a lease row of its own: off, or not registered", async () => {
|
||||
const replaced = await otherBoot({ expiredMsAgo: 600_000 });
|
||||
registerControllerInstanceDb(db);
|
||||
expect(await proveOwnerGone(stampOf(replaced))).toEqual({ verdict: "unprovable", reason: "different_process_space" });
|
||||
expect(await proveControllerInstance(db, replaced, null)).toBe("expired");
|
||||
});
|
||||
|
||||
it("an idle prover creates its own row on demand, then fences", async () => {
|
||||
const replaced = await otherBoot({ expiredMsAgo: 600_000 });
|
||||
await startHere({}, false);
|
||||
expect(await db.select().from(controllerInstances).where(eq(controllerInstances.bootId, currentControllerBootId()))).toEqual([]);
|
||||
expect(await proveOwnerGone(stampOf(replaced))).toEqual({ verdict: "gone", reason: "controller_fenced" });
|
||||
const [own] = await db.select().from(controllerInstances).where(eq(controllerInstances.bootId, currentControllerBootId()));
|
||||
const [fenced] = await db.select().from(controllerInstances).where(eq(controllerInstances.bootId, replaced));
|
||||
expect(fenced?.fencedByEpoch).toBe(own?.epoch);
|
||||
});
|
||||
|
||||
it("never fences on a read-only pass", async () => {
|
||||
const replaced = await otherBoot({ expiredMsAgo: 600_000 });
|
||||
await startHere();
|
||||
const leaseId = randomUUID();
|
||||
await db.insert(environmentLeases).values({ id: leaseId, companyId, heartbeatRunId: null, status: "active",
|
||||
leasePolicy: "ephemeral", provider: "fake", providerLeaseId: `sandbox://fake/${leaseId}`, updatedAt: new Date(minutesAgo(15)),
|
||||
metadata: { [ENVIRONMENT_LEASE_CONTROLLER_KEY]: { bootId: replaced, pid: 7, hostHash: "old-container", processStartedAt: null } } });
|
||||
const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false });
|
||||
expect(report.kinds.find(kind => kind.kind === "environment_login_probe_lease")).toMatchObject({ settled: 0 });
|
||||
expect(await proveControllerInstance(db, replaced, null)).toBe("expired");
|
||||
});
|
||||
|
||||
it("keeps renewing while this boot holds a native finalizer lease", async () => {
|
||||
vi.useFakeTimers({ toFake: ["Date"], now: new Date() });
|
||||
try {
|
||||
await startHere();
|
||||
vi.setSystemTime(Date.now() + 120_000);
|
||||
const holder = { token: "token-1", bootId: currentControllerBootId(), pid: process.pid, hostHash: currentProcessSpaceHash(), processStartedAt: null };
|
||||
const agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID();
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Finalizer", adapterType: "codex_local", status: "idle" });
|
||||
await db.insert(issues).values({ id: issueId, companyId, title: "Native task" });
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "succeeded", invocationSource: "on_demand", runtimeMode: "native", nativeIssueId: issueId, issueId });
|
||||
await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, phase: "observed",
|
||||
leaseOwner: "token-1", leaseHolder: holder, leaseExpiresAt: new Date(Date.now() + 300_000) });
|
||||
await db.update(controllerInstances).set({ leaseExpiresAt: sql`now()` });
|
||||
currentControllerProcessStamp(); vi.setSystemTime(Date.now() + 120_000);
|
||||
await tickControllerInstanceLease();
|
||||
expect(await proveControllerInstance(db, currentControllerBootId(), null)).toBe("live");
|
||||
await db.delete(nativeRunFinalizations);
|
||||
await db.delete(heartbeatRuns);
|
||||
await db.delete(issues);
|
||||
await db.delete(agents);
|
||||
} finally { vi.useRealTimers(); }
|
||||
});
|
||||
|
||||
it("keeps renewing while this boot runs a legacy run", async () => {
|
||||
vi.useFakeTimers({ toFake: ["Date"], now: new Date() });
|
||||
try {
|
||||
await startHere();
|
||||
vi.setSystemTime(Date.now() + 120_000);
|
||||
const agentId = randomUUID(), runId = randomUUID();
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Legacy", adapterType: "codex_local", status: "running" });
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", invocationSource: "on_demand",
|
||||
runtimeMode: "legacy", controllerBootId: currentControllerBootId() });
|
||||
await db.update(controllerInstances).set({ leaseExpiresAt: sql`now()` });
|
||||
currentControllerProcessStamp(); vi.setSystemTime(Date.now() + 120_000);
|
||||
await tickControllerInstanceLease();
|
||||
expect(await proveControllerInstance(db, currentControllerBootId(), null)).toBe("live");
|
||||
await db.delete(heartbeatRuns);
|
||||
await db.delete(agents);
|
||||
} finally { vi.useRealTimers(); }
|
||||
});
|
||||
|
||||
it("proves a cleanly stopped boot gone at once", async () => {
|
||||
const stopped = await otherBoot({ liveForMs: 60_000 });
|
||||
await db.update(controllerInstances).set({ stoppedAt: sql`now()` }).where(eq(controllerInstances.bootId, stopped));
|
||||
registerControllerInstanceDb(db);
|
||||
expect(await proveOwnerGone(stampOf(stopped))).toEqual({ verdict: "gone", reason: "controller_stopped" });
|
||||
});
|
||||
|
||||
it("keeps a record that names a process group unprovable: agent processes may outlive their server", async () => {
|
||||
const replaced = await otherBoot();
|
||||
await startHere();
|
||||
expect(await proveOwnerGone(stampOf(replaced, { processGroupId: 7 }))).toEqual({ verdict: "unprovable", reason: "different_process_space" });
|
||||
});
|
||||
|
||||
it("issues no statements while idle, and renews in one statement only while it holds stamped work", async () => {
|
||||
let now = 0;
|
||||
vi.useFakeTimers({ toFake: ["Date"], now: new Date() });
|
||||
try {
|
||||
await startHere({ now: () => now }, false);
|
||||
await tickControllerInstanceLease();
|
||||
expect(await db.select().from(controllerInstances)).toEqual([]);
|
||||
currentControllerProcessStamp();
|
||||
await tickControllerInstanceLease();
|
||||
const [created] = await db.select().from(controllerInstances);
|
||||
expect(created).toMatchObject({ bootId: currentControllerBootId(), processSpace: currentProcessSpaceHash() });
|
||||
// Past the stamp window, a record stamped with this boot still holds the lease.
|
||||
now = 120_000;
|
||||
vi.setSystemTime(Date.now() + 120_000);
|
||||
const leaseId = randomUUID();
|
||||
await db.insert(environmentLeases).values({ id: leaseId, companyId, heartbeatRunId: null, status: "active", leasePolicy: "ephemeral",
|
||||
metadata: { [ENVIRONMENT_LEASE_CONTROLLER_KEY]: { bootId: currentControllerBootId(), pid: process.pid, hostHash: currentProcessSpaceHash(), processStartedAt: null } } });
|
||||
await db.update(controllerInstances).set({ leaseExpiresAt: sql`now()` });
|
||||
await tickControllerInstanceLease();
|
||||
expect((await proveControllerInstance(db, currentControllerBootId(), null))).toBe("live");
|
||||
// Released, and the server's work settled (every stamped write committed): the next renewal
|
||||
// lets the lease lapse and the worker goes quiet.
|
||||
await db.delete(environmentLeases);
|
||||
beginIdleTrackedWork()();
|
||||
await db.update(controllerInstances).set({ leaseExpiresAt: sql`now()` });
|
||||
await tickControllerInstanceLease();
|
||||
expect((await proveControllerInstance(db, currentControllerBootId(), null))).toBe("expired");
|
||||
const renew = vi.fn();
|
||||
resetControllerInstanceLeaseForTests();
|
||||
registerControllerInstanceLeaseWork(db, coordinator, { renew });
|
||||
await tickControllerInstanceLease();
|
||||
expect(renew).not.toHaveBeenCalled();
|
||||
} finally { vi.useRealTimers(); }
|
||||
});
|
||||
|
||||
it("re-checks holdings when other work settles, not on its own run's settle, at most every 5 s", async () => {
|
||||
vi.useFakeTimers({ toFake: ["Date"], now: new Date() });
|
||||
try {
|
||||
const wake = vi.fn();
|
||||
const capturing = { register: () => ({ ready: Promise.resolve(), wake, stop: async () => {} }) } as unknown as
|
||||
ReturnType<typeof createDeliveryWorkCoordinator>;
|
||||
registerControllerInstanceDb(db);
|
||||
registerControllerInstanceLeaseWork(db, capturing, { renew: async () => "holding" as const });
|
||||
currentControllerProcessStamp();
|
||||
expect(wake).toHaveBeenCalledTimes(1);
|
||||
await tickControllerInstanceLease();
|
||||
beginIdleTrackedWork()();
|
||||
expect(wake).toHaveBeenCalledTimes(1);
|
||||
beginIdleTrackedWork()();
|
||||
expect(wake).toHaveBeenCalledTimes(1);
|
||||
vi.setSystemTime(Date.now() + 6_000);
|
||||
beginIdleTrackedWork()();
|
||||
expect(wake).toHaveBeenCalledTimes(2);
|
||||
} finally { vi.useRealTimers(); }
|
||||
});
|
||||
|
||||
it("never loses a stamp taken while a renewal is in flight", async () => {
|
||||
const wake = vi.fn();
|
||||
const capturing = { register: () => ({ ready: Promise.resolve(), wake, stop: async () => {} }) } as unknown as
|
||||
ReturnType<typeof createDeliveryWorkCoordinator>;
|
||||
registerControllerInstanceDb(db);
|
||||
registerControllerInstanceLeaseWork(db, capturing, { renew: async () => { currentControllerProcessStamp(); return "idle" as const; } });
|
||||
currentControllerProcessStamp();
|
||||
wake.mockClear();
|
||||
await tickControllerInstanceLease();
|
||||
expect(wake).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("counts a cleanup claim as held only while its attempt is in flight", async () => {
|
||||
vi.useFakeTimers({ toFake: ["Date"], now: new Date() });
|
||||
try {
|
||||
await startHere();
|
||||
vi.setSystemTime(Date.now() + 60_000);
|
||||
const leaseId = randomUUID();
|
||||
const claim = { bootId: currentControllerBootId(), pid: process.pid, hostHash: currentProcessSpaceHash(), processStartedAt: null };
|
||||
await db.insert(environmentLeases).values({ id: leaseId, companyId, heartbeatRunId: null, status: "pending_cleanup", leasePolicy: "ephemeral",
|
||||
metadata: { pendingCleanupController: claim, pendingCleanupInFlight: false } });
|
||||
await db.update(controllerInstances).set({ leaseExpiresAt: sql`now()` });
|
||||
// A stamp whose record no renewal saw holds the lease until the 10-minute cap.
|
||||
currentControllerProcessStamp(); vi.setSystemTime(Date.now() + 120_000);
|
||||
await tickControllerInstanceLease();
|
||||
expect(await proveControllerInstance(db, currentControllerBootId(), null)).toBe("live");
|
||||
vi.setSystemTime(Date.now() + 11 * 60_000);
|
||||
await db.update(controllerInstances).set({ leaseExpiresAt: sql`now()` });
|
||||
await tickControllerInstanceLease();
|
||||
expect(await proveControllerInstance(db, currentControllerBootId(), null)).toBe("expired");
|
||||
await db.update(environmentLeases).set({ metadata: { pendingCleanupController: claim, pendingCleanupInFlight: true } });
|
||||
currentControllerProcessStamp(); vi.setSystemTime(Date.now() + 120_000);
|
||||
await tickControllerInstanceLease();
|
||||
expect(await proveControllerInstance(db, currentControllerBootId(), null)).toBe("live");
|
||||
} finally { vi.useRealTimers(); }
|
||||
});
|
||||
|
||||
it("holds an unobserved stamp only until the server's work settles", async () => {
|
||||
vi.useFakeTimers({ toFake: ["Date"], now: new Date() });
|
||||
try {
|
||||
await startHere();
|
||||
vi.setSystemTime(Date.now() + 120_000);
|
||||
await db.update(controllerInstances).set({ leaseExpiresAt: sql`now()` });
|
||||
await tickControllerInstanceLease();
|
||||
expect(await proveControllerInstance(db, currentControllerBootId(), null)).toBe("live");
|
||||
beginIdleTrackedWork()();
|
||||
await db.update(controllerInstances).set({ leaseExpiresAt: sql`now()` });
|
||||
await tickControllerInstanceLease();
|
||||
expect(await proveControllerInstance(db, currentControllerBootId(), null)).toBe("expired");
|
||||
} finally { vi.useRealTimers(); }
|
||||
});
|
||||
|
||||
it("keeps a pending stamped write held when a renewal sees another held record", async () => {
|
||||
vi.useFakeTimers({ toFake: ["Date"], now: new Date() });
|
||||
try {
|
||||
await startHere();
|
||||
const agentId = randomUUID(), runId = randomUUID();
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Legacy", adapterType: "codex_local", status: "running" });
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", invocationSource: "on_demand",
|
||||
runtimeMode: "legacy", controllerBootId: currentControllerBootId() });
|
||||
// A second stamp whose ownership write has not committed yet.
|
||||
currentControllerProcessStamp();
|
||||
await tickControllerInstanceLease();
|
||||
// The other record finishes; the pending write is still uncommitted past the stamp window.
|
||||
await db.delete(heartbeatRuns);
|
||||
await db.delete(agents);
|
||||
vi.setSystemTime(Date.now() + 120_000);
|
||||
await db.update(controllerInstances).set({ leaseExpiresAt: sql`now()` });
|
||||
await tickControllerInstanceLease();
|
||||
expect(await proveControllerInstance(db, currentControllerBootId(), null)).toBe("live");
|
||||
beginIdleTrackedWork()();
|
||||
await db.update(controllerInstances).set({ leaseExpiresAt: sql`now()` });
|
||||
await tickControllerInstanceLease();
|
||||
expect(await proveControllerInstance(db, currentControllerBootId(), null)).toBe("expired");
|
||||
} finally { vi.useRealTimers(); }
|
||||
});
|
||||
|
||||
it("rides out a 45 s database outage: quiesces dispatch, never exits, and resumes", async () => {
|
||||
let now = 0, down = true;
|
||||
const renew = vi.fn(async () => { if (down) throw new Error("connection refused"); return "holding" as const; });
|
||||
const { lost } = await startHere({ now: () => now, renew }, false);
|
||||
currentControllerProcessStamp();
|
||||
for (now = 0; now <= 45_000; now += 5_000) {
|
||||
await tickControllerInstanceLease();
|
||||
expect(readTaskDrain(new Date())).not.toBeNull();
|
||||
}
|
||||
down = false;
|
||||
await tickControllerInstanceLease();
|
||||
expect(lost).not.toHaveBeenCalled();
|
||||
expect(readTaskDrain(new Date())).toBeNull();
|
||||
});
|
||||
|
||||
it("bounds a renewal that never answers, counting from the attempt's start", async () => {
|
||||
let now = 0;
|
||||
vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] });
|
||||
try {
|
||||
const { lost } = await startHere({ now: () => now, renew: () => new Promise(() => {}) }, false);
|
||||
currentControllerProcessStamp();
|
||||
const attempt = tickControllerInstanceLease();
|
||||
now = 10_000;
|
||||
await vi.advanceTimersByTimeAsync(10_000);
|
||||
await attempt;
|
||||
expect(readTaskDrain(new Date())).not.toBeNull();
|
||||
for (now = 15_000; now <= 60_000; now += 5_000) {
|
||||
const next = tickControllerInstanceLease();
|
||||
await vi.advanceTimersByTimeAsync(10_000);
|
||||
await next;
|
||||
}
|
||||
expect(lost).toHaveBeenCalledWith("lease_lost");
|
||||
} finally { vi.useRealTimers(); }
|
||||
});
|
||||
|
||||
it("stops after a 70 s outage while holding work, and the successor recovers its work exactly once", async () => {
|
||||
let now = 0;
|
||||
const { lost } = await startHere({ now: () => now, renew: async () => { throw new Error("down"); } }, false);
|
||||
currentControllerProcessStamp();
|
||||
for (now = 0; now <= 70_000 && !lost.mock.calls.length; now += 5_000) await tickControllerInstanceLease();
|
||||
expect(lost).toHaveBeenCalledWith("lease_lost");
|
||||
expect(now).toBeGreaterThanOrEqual(60_000);
|
||||
expect(now).toBeLessThanOrEqual(70_000);
|
||||
|
||||
// The stopped boot as its successor sees it: lease expired long ago, work last written under it.
|
||||
resetControllerInstanceLeaseForTests();
|
||||
await db.insert(controllerInstances).values({ bootId: "exited-boot", processSpace: "old-container", pid: 7,
|
||||
leaseExpiresAt: sql`now() - interval '3 minutes'`, epoch: 1 });
|
||||
await startHere();
|
||||
const leaseId = randomUUID();
|
||||
await db.insert(environmentLeases).values({ id: leaseId, companyId, heartbeatRunId: null, status: "active",
|
||||
leasePolicy: "ephemeral", provider: "fake", providerLeaseId: `sandbox://fake/${leaseId}`, updatedAt: new Date(minutesAgo(5)),
|
||||
metadata: { [ENVIRONMENT_LEASE_CONTROLLER_KEY]: { bootId: "exited-boot", pid: 7, hostHash: "old-container", processStartedAt: null } } });
|
||||
const passes = [];
|
||||
for (let pass = 0; pass < 2; pass += 1) {
|
||||
const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState() });
|
||||
passes.push(report.kinds.find(kind => kind.kind === "environment_login_probe_lease")?.settled);
|
||||
}
|
||||
expect(passes).toEqual([1, 0]);
|
||||
expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, leaseId)))[0]).toMatchObject({ status: "pending_cleanup" });
|
||||
});
|
||||
|
||||
it("stops at once when a newer boot fenced it", async () => {
|
||||
const { lost } = await startHere();
|
||||
await db.update(controllerInstances).set({ fencedAt: sql`now()` }).where(eq(controllerInstances.bootId, currentControllerBootId()));
|
||||
currentControllerProcessStamp();
|
||||
await tickControllerInstanceLease();
|
||||
expect(lost).toHaveBeenCalledWith("fenced");
|
||||
});
|
||||
|
||||
it("records a graceful stop, and is off on a volume-exclusive Railway deployment or when disabled", async () => {
|
||||
await startHere();
|
||||
await stopControllerInstanceLease(db);
|
||||
expect(await proveControllerInstance(db, currentControllerBootId(), null)).toBe("stopped");
|
||||
vi.stubEnv("PAPERCLIP_CONTROLLER_INSTANCE_LEASE", "off");
|
||||
expect(controllerInstanceLeaseEnabled()).toBe(false);
|
||||
vi.unstubAllEnvs();
|
||||
for (const [key, value] of [["RAILWAY_PROJECT_ID", "p"], ["RAILWAY_ENVIRONMENT_ID", "e"], ["RAILWAY_SERVICE_ID", "s"], ["RAILWAY_DEPLOYMENT_ID", "d"]]) {
|
||||
vi.stubEnv(key!, value!);
|
||||
}
|
||||
// A stateless Railway service has no deployment-replacement proof, so it keeps the lease.
|
||||
expect(controllerInstanceLeaseEnabled()).toBe(true);
|
||||
vi.stubEnv("RAILWAY_VOLUME_ID", "v");
|
||||
vi.stubEnv("RAILWAY_VOLUME_MOUNT_PATH", "/data");
|
||||
expect(controllerInstanceLeaseEnabled()).toBe(false);
|
||||
vi.stubEnv("PAPERCLIP_CONTROLLER_INSTANCE_LEASE", "on");
|
||||
expect(controllerInstanceLeaseEnabled()).toBe(true);
|
||||
vi.unstubAllEnvs();
|
||||
});
|
||||
});
|
||||
@@ -161,7 +161,12 @@ describeDatabase("assembled maintenance server production wiring", () => {
|
||||
await server.updateIssue({ id: issue!.id, status: "done" }); await settleClock(server);
|
||||
const quiet = server.proxy.queries.length;
|
||||
for (let step = 0; step < 4; step++) { await server.advance(30_000); await settleClock(server); }
|
||||
expect(server.proxy.queries.length).toBe(quiet);
|
||||
// The controller instance lease may check once more that the boot no longer holds the
|
||||
// run's stamped records, then stops (it renews only while the boot holds work).
|
||||
expect(server.proxy.queries.slice(quiet).filter(query => !query.sql.includes("controller_instances"))).toEqual([]);
|
||||
const leaseIdle = server.proxy.queries.length;
|
||||
for (let step = 0; step < 2; step++) { await server.advance(30_000); await settleClock(server); }
|
||||
expect(server.proxy.queries.length).toBe(leaseIdle);
|
||||
await writeFile(path.join(server.directory, "receipt-measurement.json"), JSON.stringify({ phase: "ordinary-held-task-retry", companyId: company.id, issueId: issue!.id,
|
||||
requestId: retry.requestId, waiting, grace, parked, executed, count: count!.count, parkedRealMs: 30_000, parkedQueries: 0, settledClockMs: 120_000, settledQueries: 0,
|
||||
boundary: "actual Retry and subtree release HTTP, wakeAgents false, real clock through process/controller authority, real local process, no outcome seeded", frames: server.proxy.frames }, null, 2));
|
||||
|
||||
+12
-1
@@ -78,6 +78,8 @@ import { chatCompletionDeliveryService } from "./services/chat-completion-delive
|
||||
// instrumentationReady before opening DB connections or constructing the
|
||||
// HTTP server, so trace coverage does not depend on incidental timing.
|
||||
import { instrumentationReady, shutdownInstrumentation } from "./instrumentation.js";
|
||||
import { controllerInstanceLeaseEnabled, registerControllerInstanceDb, registerControllerInstanceLeaseWork,
|
||||
stopControllerInstanceLease } from "./services/controller-instance-lease.js";
|
||||
import { sentryReady, shutdownSentry, captureException } from "./sentry.js";
|
||||
import { setSentryRunLinkResolver } from "./sentry-links.js";
|
||||
import { createSentryRunLinkResolver } from "./services/sentry-run-links.js";
|
||||
@@ -1097,6 +1099,13 @@ async function startServerWithDatabaseTeardown(
|
||||
logger.error({ err }, "startup reconciliation of managed runtime control operations failed");
|
||||
}
|
||||
|
||||
// Records this boot owns are stamped with its boot id; the lease lets a server on
|
||||
// another host or container prove this boot gone after it is replaced.
|
||||
registerControllerInstanceDb(db as any);
|
||||
const controllerInstanceLeaseWork = controllerInstanceLeaseEnabled()
|
||||
? registerControllerInstanceLeaseWork(db as any, deliveryWork) : null;
|
||||
if (controllerInstanceLeaseWork) server.on("close", () => { void controllerInstanceLeaseWork.stop(); });
|
||||
|
||||
// These jobs need not delay readiness, but their writes must finish before
|
||||
// an idle report can authorize sleep. Count the whole detached promise.
|
||||
void trackIdleWork(reconcilePersistedRuntimeServicesOnStartup(db as any)
|
||||
@@ -2167,7 +2176,9 @@ async function startServerWithDatabaseTeardown(
|
||||
// Keep the finalizer's defensive listener close for partial teardown paths.
|
||||
await finalizeServerShutdown({
|
||||
signal,
|
||||
shutdownAppServices: appShutdown,
|
||||
shutdownAppServices: async () => {
|
||||
try { await appShutdown?.(); } finally { await stopControllerInstanceLease(); }
|
||||
},
|
||||
closeHttpListener: () =>
|
||||
closeHttpListenerForShutdown({ server, signal, log: logger }),
|
||||
drainPendingRunFailureReports: waitForPendingRunFailureReports,
|
||||
|
||||
@@ -0,0 +1,298 @@
|
||||
import { and, eq, isNull, sql } from "drizzle-orm";
|
||||
import { controllerInstances, type Db } from "@paperclipai/db";
|
||||
import { logger } from "../middleware/logger.js";
|
||||
import {
|
||||
currentControllerBootId, currentControllerDeployment, currentProcessSpaceHash, onControllerStamp,
|
||||
} from "./controller-process-identity.js";
|
||||
import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
|
||||
import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
|
||||
import { setControllerLeaseQuiesce, subscribeIdleWorkSettled } from "./task-admission.js";
|
||||
|
||||
export const CONTROLLER_LEASE_TTL_MS = 90_000;
|
||||
export const CONTROLLER_LEASE_RENEW_MS = 30_000;
|
||||
/** Renewal attempts start failing from their own start; this boot stops after this long while holding work. */
|
||||
export const CONTROLLER_LEASE_SELF_FENCE_MS = 60_000;
|
||||
/** A prover fences a boot only this long after the boot's last sign of life. */
|
||||
export const CONTROLLER_LEASE_FENCE_GRACE_MS = 30_000;
|
||||
/** A stamp counts as held work for a full lease, so a slow ownership write commits before the lease can lapse. */
|
||||
const STAMP_COMMIT_GRACE_MS = CONTROLLER_LEASE_TTL_MS;
|
||||
/** A settle of other work re-checks holdings at most this often. */
|
||||
const SETTLE_RECHECK_MIN_MS = 5_000;
|
||||
/** A renewal that has not answered within this bound counts as failed. */
|
||||
const RENEW_TIMEOUT_MS = 10_000;
|
||||
const RETRY_MS = 5_000;
|
||||
|
||||
export type ControllerInstanceState = "live" | "grace" | "expired" | "fenced" | "stopped";
|
||||
/** "holding" saw an in-progress record held by this boot; "holding_stamp" held only for a recent stamp. */
|
||||
type RenewResult = "holding" | "holding_stamp" | "idle" | "fenced";
|
||||
/** A stamp whose record no renewal has seen keeps the lease held until the server's tracked work next settles
|
||||
* (its write then committed or rolled back), and at most this long. */
|
||||
const UNOBSERVED_STAMP_CAP_MS = 10 * 60_000;
|
||||
|
||||
type LeaseState = {
|
||||
onLost: (reason: "fenced" | "lease_lost") => void;
|
||||
now: () => number;
|
||||
renew: (stampedRecently: boolean) => Promise<RenewResult>;
|
||||
lastStampAt: number;
|
||||
/** Start of the first renewal attempt in the current failing streak. */
|
||||
failingSince: number | null;
|
||||
/** Renewal is due: this boot stamped work, or its last renewal found work it still holds. */
|
||||
active: boolean;
|
||||
epoch: number | null;
|
||||
stopped: boolean;
|
||||
/** Wall-clock time of the last stamp (the stamp window follows the server's clock). */
|
||||
lastStampWallAt: number;
|
||||
lastTickAt: number;
|
||||
/** Counts stamps, so a stamp taken while a renewal is in flight is never lost. */
|
||||
stampSeq: number;
|
||||
/** Wall time of the oldest stamp no renewal has yet seen a held record for. */
|
||||
unobservedStampSince: number | null;
|
||||
unsubscribe: () => void;
|
||||
wake: () => void;
|
||||
};
|
||||
|
||||
let state: LeaseState | null = null;
|
||||
let registeredDb: Db | null = null;
|
||||
|
||||
/** A Railway deployment with an exclusive volume proves a replaced deployment on its own; every
|
||||
* other host (Docker, Kubernetes, Fly, a stateless Railway service, a rebooted machine) needs the
|
||||
* lease to prove a replaced process. PAPERCLIP_CONTROLLER_INSTANCE_LEASE=on|off overrides this. */
|
||||
export function controllerInstanceLeaseEnabled(env: NodeJS.ProcessEnv = process.env): boolean {
|
||||
const setting = env.PAPERCLIP_CONTROLLER_INSTANCE_LEASE;
|
||||
if (setting === "off") return false;
|
||||
if (setting === "on") return true;
|
||||
return currentControllerDeployment(env)?.exclusiveVolume !== true;
|
||||
}
|
||||
|
||||
function exitOnLostLease(reason: "fenced" | "lease_lost") {
|
||||
logger.fatal({ reason }, "controller instance lease lost while holding work; stopping so a newer server can recover it");
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
/** In-progress records stamped with this boot id: the process-owned kinds of the ownership registry. */
|
||||
function holdsStampedWorkSql(bootId: string) {
|
||||
return sql`(
|
||||
exists (select 1 from environment_leases where
|
||||
(status = 'active' and metadata->'leaseController'->>'bootId' = ${bootId})
|
||||
or (status = 'pending_cleanup' and metadata->>'pendingCleanupInFlight' = 'true'
|
||||
and metadata->'pendingCleanupController'->>'bootId' = ${bootId}))
|
||||
or exists (select 1 from sandbox_creation_operations where state = 'pending' and metadata->'creationController'->>'bootId' = ${bootId})
|
||||
or exists (select 1 from workspace_operations where status in ('running', 'pending') and metadata->'finalizeOwner'->>'bootId' = ${bootId})
|
||||
or exists (select 1 from agents where lifecycle_operation->'leaseHolder'->>'bootId' = ${bootId})
|
||||
or exists (select 1 from native_run_finalizations where lease_owner is not null
|
||||
and lease_holder->>'token' = lease_owner and lease_holder->>'bootId' = ${bootId})
|
||||
or exists (select 1 from native_run_finalizations where controller_boot_id = ${bootId}
|
||||
and phase not in ('committed', 'terminal_failure', 'applied'))
|
||||
or exists (select 1 from heartbeat_runs where status = 'running' and controller_boot_id::text = ${bootId}))`;
|
||||
}
|
||||
|
||||
/**
|
||||
* One statement: create this boot's row on first use, and extend its lease only while the boot
|
||||
* holds work (a stamp within the TTL, or an in-progress record stamped with its boot id).
|
||||
* Returns whether it still holds work, or that a newer boot fenced it.
|
||||
*/
|
||||
function renewInDatabase(db: Db): (stampedRecently: boolean) => Promise<RenewResult> {
|
||||
return async stampedRecently => {
|
||||
const bootId = currentControllerBootId();
|
||||
const records = holdsStampedWorkSql(bootId);
|
||||
const holds = sql`(${stampedRecently} or ${records})`;
|
||||
const expires = sql`now() + ${`${CONTROLLER_LEASE_TTL_MS} milliseconds`}::interval`;
|
||||
const rows = await db.execute(sql`
|
||||
insert into controller_instances (boot_id, process_space, pid, lease_expires_at)
|
||||
values (${bootId}, ${currentProcessSpaceHash()}, ${process.pid}, ${expires})
|
||||
on conflict (boot_id) do update set
|
||||
lease_expires_at = case when ${holds} then ${expires} else controller_instances.lease_expires_at end,
|
||||
renewed_at = case when ${holds} then now() else controller_instances.renewed_at end
|
||||
where controller_instances.fenced_at is null and controller_instances.stopped_at is null
|
||||
returning epoch, ${holds} as holds, ${records} as records`) as unknown as Array<{ epoch: string | number; holds: boolean; records: boolean }>;
|
||||
const row = rows[0];
|
||||
if (!row) return "fenced";
|
||||
if (state) state.epoch = Number(row.epoch);
|
||||
return row.records ? "holding" : row.holds ? "holding_stamp" : "idle";
|
||||
};
|
||||
}
|
||||
|
||||
function withTimeout<T>(work: Promise<T>, ms: number): Promise<T> {
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
return Promise.race([work, new Promise<never>((_, reject) => {
|
||||
timer = setTimeout(() => reject(new Error("controller instance lease renewal timed out")), ms);
|
||||
})]).finally(() => clearTimeout(timer));
|
||||
}
|
||||
|
||||
/**
|
||||
* Registers this boot's lease with the shared delivery coordinator. The lease is renewed only
|
||||
* while the boot holds work, so an idle boot issues no statements and its database can sleep.
|
||||
*/
|
||||
export function registerControllerInstanceLeaseWork(db: Db, coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
|
||||
options: {
|
||||
onLost?: (reason: "fenced" | "lease_lost") => void;
|
||||
now?: () => number;
|
||||
renew?: (stampedRecently: boolean) => Promise<RenewResult>;
|
||||
} = {}) {
|
||||
registeredDb = db;
|
||||
const current: LeaseState = {
|
||||
onLost: options.onLost ?? exitOnLostLease, now: options.now ?? (() => performance.now()),
|
||||
renew: options.renew ?? renewInDatabase(db), lastStampAt: Number.NEGATIVE_INFINITY, failingSince: null,
|
||||
active: false, epoch: null, stopped: false, unsubscribe: () => {}, wake: () => {},
|
||||
lastStampWallAt: Number.NEGATIVE_INFINITY, lastTickAt: Number.NEGATIVE_INFINITY, stampSeq: 0,
|
||||
unobservedStampSince: null,
|
||||
};
|
||||
state = current;
|
||||
const worker = coordinator.register(DELIVERY_QUEUES.controllerInstanceLease, {
|
||||
retryMs: RETRY_MS,
|
||||
runDuringDrain: true,
|
||||
run: () => tickControllerInstanceLease(),
|
||||
hasPending: async () => false,
|
||||
nextRunAt: () => nextLeaseDeadline(current),
|
||||
});
|
||||
current.wake = worker.wake;
|
||||
const unsubscribeStamp = onControllerStamp(() => {
|
||||
const wasActive = current.active;
|
||||
current.lastStampAt = current.now();
|
||||
current.lastStampWallAt = Date.now();
|
||||
current.stampSeq += 1;
|
||||
current.unobservedStampSince ??= Date.now();
|
||||
current.active = true;
|
||||
// The first stamp after an idle lapse renews at once, without delaying the stamping write.
|
||||
if (!wasActive && !current.stopped) worker.wake();
|
||||
});
|
||||
// When the server's work settles, check holdings now rather than one renewal later,
|
||||
// so an idle server issues no statement after its last work.
|
||||
// The rate limit also absorbs the settle of this worker's own run.
|
||||
const unsubscribeSettle = subscribeIdleWorkSettled(() => {
|
||||
// All tracked work finished: any ownership write it made has committed or rolled back,
|
||||
// so the holdings query is authoritative from here on.
|
||||
current.unobservedStampSince = null;
|
||||
if (current.active && !current.stopped && Date.now() - current.lastTickAt >= SETTLE_RECHECK_MIN_MS) worker.wake();
|
||||
});
|
||||
current.unsubscribe = () => { unsubscribeStamp(); unsubscribeSettle(); };
|
||||
return {
|
||||
ready: worker.ready,
|
||||
stop: async () => {
|
||||
current.unsubscribe();
|
||||
await worker.stop();
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function nextLeaseDeadline(current: LeaseState): number | null {
|
||||
if (current.stopped) return null;
|
||||
if (current.failingSince !== null) return Date.now() + RETRY_MS;
|
||||
return current.active ? Date.now() + CONTROLLER_LEASE_RENEW_MS : null;
|
||||
}
|
||||
|
||||
/** Renew while holding work; quiesce on failure; stop only if the failure outlasts the self-fence window with work held. */
|
||||
export async function tickControllerInstanceLease(): Promise<void> {
|
||||
const current = state;
|
||||
if (!current || current.stopped || (!current.active && current.failingSince === null)) return;
|
||||
const attemptStartedAt = current.now();
|
||||
const stampSeqAtStart = current.stampSeq;
|
||||
current.lastTickAt = Date.now();
|
||||
// Held until a renewal sees the stamped record committed (or rolled back past the cap).
|
||||
const stampedRecently = Date.now() - current.lastStampWallAt < STAMP_COMMIT_GRACE_MS
|
||||
|| (current.unobservedStampSince !== null && Date.now() - current.unobservedStampSince < UNOBSERVED_STAMP_CAP_MS);
|
||||
let result: RenewResult | "failed";
|
||||
try { result = await withTimeout(current.renew(stampedRecently), RENEW_TIMEOUT_MS); } catch (error) {
|
||||
logger.warn({ err: error }, "controller instance lease renewal failed");
|
||||
result = "failed";
|
||||
}
|
||||
if (result === "holding" || result === "holding_stamp" || result === "idle") {
|
||||
current.failingSince = null;
|
||||
// Idle: the lease lapses with no further statements until the next stamp wakes it.
|
||||
// A stamp taken during this attempt was not seen by its snapshot: check again now.
|
||||
const stampedDuringAttempt = current.stampSeq !== stampSeqAtStart;
|
||||
// A held record does not show that every stamped write has committed: another stamp's
|
||||
// write may still be pending. Only the work-settled signal (or the cap) clears it.
|
||||
current.active = result !== "idle" || stampedDuringAttempt;
|
||||
if (result === "idle" && stampedDuringAttempt) current.wake();
|
||||
setControllerLeaseQuiesce(false);
|
||||
return;
|
||||
}
|
||||
// Fenced or unreachable: dispatch nothing new while the outcome is open.
|
||||
setControllerLeaseQuiesce(true);
|
||||
if (result === "fenced") { current.stopped = true; current.onLost("fenced"); return; }
|
||||
current.failingSince ??= attemptStartedAt;
|
||||
if (current.now() - current.failingSince >= CONTROLLER_LEASE_SELF_FENCE_MS && (current.active || stampedRecently)) {
|
||||
current.stopped = true;
|
||||
current.onLost("lease_lost");
|
||||
}
|
||||
}
|
||||
|
||||
/** Graceful shutdown: this boot no longer owns anything, so its stamps are gone at once. */
|
||||
export async function stopControllerInstanceLease(db: Db | null = registeredDb): Promise<void> {
|
||||
const current = state;
|
||||
if (current) { current.stopped = true; current.unsubscribe(); }
|
||||
state = null;
|
||||
if (!db || current?.epoch == null) return;
|
||||
await db.update(controllerInstances).set({ stoppedAt: sql`now()` })
|
||||
.where(eq(controllerInstances.bootId, currentControllerBootId()))
|
||||
.catch(error => logger.warn({ err: error }, "controller instance stop record failed"));
|
||||
}
|
||||
|
||||
/**
|
||||
* The lease state of another boot, for a record it last wrote at `touchedAt`. The boot's last
|
||||
* sign of life is the later of its lease expiry and that write plus the TTL (a boot that resumed
|
||||
* after an idle lapse renews on its first stamp). Within the grace after it, the owner is still
|
||||
* treated as alive so the record stays watched; past it, the boot is fenced (only by a newer
|
||||
* epoch) and its records are gone. A boot that held work exits before this point.
|
||||
*/
|
||||
export async function proveControllerInstance(db: Db, bootId: string, touchedAt: Date | null,
|
||||
options: { readOnly?: boolean } = {}): Promise<ControllerInstanceState | null> {
|
||||
const [row] = await db.select({
|
||||
fenced: sql<boolean>`${controllerInstances.fencedAt} is not null`,
|
||||
stopped: sql<boolean>`${controllerInstances.stoppedAt} is not null`,
|
||||
live: sql<boolean>`${controllerInstances.leaseExpiresAt} > now()`,
|
||||
pastGrace: sql<boolean>`greatest(${controllerInstances.leaseExpiresAt},
|
||||
${touchedAt?.toISOString() ?? null}::timestamptz + ${`${CONTROLLER_LEASE_TTL_MS} milliseconds`}::interval)
|
||||
< now() - ${`${CONTROLLER_LEASE_FENCE_GRACE_MS} milliseconds`}::interval`,
|
||||
}).from(controllerInstances).where(eq(controllerInstances.bootId, bootId));
|
||||
if (!row) return null;
|
||||
if (row.stopped) return "stopped";
|
||||
if (row.fenced) return "fenced";
|
||||
if (row.live) return "live";
|
||||
if (!touchedAt) return "expired";
|
||||
if (!row.pastGrace) return "grace";
|
||||
// A read-only pass reports; only a mutating pass fences (or creates this boot's row).
|
||||
if (options.readOnly) return "expired";
|
||||
// Only a boot that holds a lease row of its own may fence, and only an older epoch.
|
||||
const epoch = await ownEpoch();
|
||||
if (epoch === null) return "expired";
|
||||
const [fenced] = await db.update(controllerInstances).set({ fencedAt: sql`now()`, fencedByEpoch: epoch })
|
||||
.where(and(eq(controllerInstances.bootId, bootId), isNull(controllerInstances.fencedAt), isNull(controllerInstances.stoppedAt),
|
||||
sql`greatest(${controllerInstances.leaseExpiresAt}, ${touchedAt.toISOString()}::timestamptz
|
||||
+ ${`${CONTROLLER_LEASE_TTL_MS} milliseconds`}::interval)
|
||||
< now() - ${`${CONTROLLER_LEASE_FENCE_GRACE_MS} milliseconds`}::interval`,
|
||||
sql`${controllerInstances.epoch} < ${epoch}`))
|
||||
.returning({ bootId: controllerInstances.bootId });
|
||||
if (fenced) return "fenced";
|
||||
const [after] = await db.select({ fenced: sql<boolean>`${controllerInstances.fencedAt} is not null` })
|
||||
.from(controllerInstances).where(eq(controllerInstances.bootId, bootId));
|
||||
return after?.fenced ? "fenced" : "expired";
|
||||
}
|
||||
|
||||
/** This boot's epoch, creating its row on first need (one statement). Null when the lease is off or unreachable. */
|
||||
async function ownEpoch(): Promise<number | null> {
|
||||
const current = state;
|
||||
if (!current || current.stopped) return null;
|
||||
if (current.epoch === null) {
|
||||
const result = await withTimeout(current.renew(false), RENEW_TIMEOUT_MS).catch(() => null);
|
||||
if (result === "fenced") return null;
|
||||
}
|
||||
return current.epoch;
|
||||
}
|
||||
|
||||
/** Lets the owner proof read other boots' leases, whether or not this boot holds one. */
|
||||
export function registerControllerInstanceDb(db: Db): void {
|
||||
registeredDb = db;
|
||||
}
|
||||
|
||||
export function registeredControllerInstanceDb(): Db | null {
|
||||
return registeredDb;
|
||||
}
|
||||
|
||||
export function resetControllerInstanceLeaseForTests(): void {
|
||||
state?.unsubscribe();
|
||||
state = null;
|
||||
registeredDb = null;
|
||||
setControllerLeaseQuiesce(false);
|
||||
}
|
||||
@@ -119,10 +119,23 @@ export function currentControllerBootId(): string {
|
||||
return bootId;
|
||||
}
|
||||
|
||||
const stampListeners = new Set<() => void>();
|
||||
/** Called (synchronously, without I/O) whenever this process stamps ownership of a record. */
|
||||
export function onControllerStamp(listener: () => void): () => void {
|
||||
stampListeners.add(listener);
|
||||
return () => { stampListeners.delete(listener); };
|
||||
}
|
||||
function notifyStamp() {
|
||||
for (const listener of stampListeners) {
|
||||
try { listener(); } catch { /* a listener never blocks a stamp */ }
|
||||
}
|
||||
}
|
||||
|
||||
export async function currentControllerProcess(): Promise<ControllerProcess> {
|
||||
const startedAt = await currentProcessStartedAtMs();
|
||||
const iso = startedAt === null ? null : new Date(startedAt).toISOString();
|
||||
resolvedProcessStartedAt = iso;
|
||||
notifyStamp();
|
||||
return { bootId, pid: process.pid, hostHash, processStartedAt: iso, ...deploymentField() };
|
||||
}
|
||||
|
||||
@@ -133,6 +146,7 @@ export async function currentControllerProcess(): Promise<ControllerProcess> {
|
||||
* a live PID then stays ambiguous, which fails closed.
|
||||
*/
|
||||
export function currentControllerProcessStamp(): ControllerProcess {
|
||||
notifyStamp();
|
||||
return { bootId, pid: process.pid, hostHash, processStartedAt: resolvedProcessStartedAt, ...deploymentField() };
|
||||
}
|
||||
|
||||
|
||||
@@ -88,6 +88,7 @@ export const DELIVERY_QUEUES = {
|
||||
budgetEnforcement: "budget-enforcement",
|
||||
accounting: "cost-accounting",
|
||||
taskDrainExpiry: "task-drain-expiry",
|
||||
controllerInstanceLease: "controller-instance-lease",
|
||||
agentLifecycle: "agent-lifecycle",
|
||||
chatBoardMessages: "chat-board-messages",
|
||||
chatTaskStarts: "chat-task-starts",
|
||||
|
||||
@@ -12,6 +12,7 @@ import {
|
||||
} from "./controller-process-identity.js";
|
||||
import type { WorkspaceWorker } from "./native-runtime/native-workspace-platform-proof.js";
|
||||
import { ephemeralServerEnabled } from "./ephemeral-server-mode.js";
|
||||
import { proveControllerInstance, registeredControllerInstanceDb, type ControllerInstanceState } from "./controller-instance-lease.js";
|
||||
|
||||
/** `processSpace` value written when a board operator confirmed that this exact
|
||||
* recorded owner stopped (native-controller-host-recovery.ts). The attestation
|
||||
@@ -57,6 +58,8 @@ export type LocalOwnerStamp = {
|
||||
* An ephemeral (multi-host) server never judges them.
|
||||
*/
|
||||
processSpaceUnrecorded?: boolean;
|
||||
/** When the owned record was last written, if the caller knows (for the controller lease proof). */
|
||||
touchedAt?: string | null;
|
||||
};
|
||||
|
||||
export type RemoteOwnerStamp = {
|
||||
@@ -89,6 +92,9 @@ export type OwnerProofReason =
|
||||
| "provider_stopped"
|
||||
| "deployment_replaced"
|
||||
| "operator_attested"
|
||||
/** The owner's server boot stopped cleanly, or a newer boot fenced it after its lease expired. */
|
||||
| "controller_stopped"
|
||||
| "controller_fenced"
|
||||
/** This process owns the record and is not executing its run. Only a
|
||||
* caller that knows this process's executions may conclude it
|
||||
* (`settleOwnableKind` with `currentProcessIdle`); the prover never does. */
|
||||
@@ -98,6 +104,8 @@ export type OwnerProofReason =
|
||||
| "pid_alive"
|
||||
| "process_group_alive"
|
||||
| "fence_held"
|
||||
| "controller_lease_held"
|
||||
| "controller_lease_grace"
|
||||
| "provider_running"
|
||||
// unprovable
|
||||
| "missing_identity"
|
||||
@@ -135,6 +143,10 @@ export type ProcessProbe = {
|
||||
* the platform stops every earlier deployment of the service before it
|
||||
* starts this one (a Railway service with an attached volume). */
|
||||
deployment?: { worker: WorkspaceWorker; exclusiveVolume: boolean; volumeId?: string | null } | null;
|
||||
/** The lease state of another server boot (controller_instances), when this server can read it. */
|
||||
controllerInstance?: (bootId: string, touchedAt: Date | null) => Promise<ControllerInstanceState | null>;
|
||||
/** Report without changing any lease row (a read-only reconciliation pass). */
|
||||
readOnly?: boolean;
|
||||
};
|
||||
|
||||
function processGroupExists(processGroupId: number): boolean {
|
||||
@@ -163,6 +175,10 @@ export function currentProcessProbe(): ProcessProbe {
|
||||
processSpace: currentProcessSpaceHash(),
|
||||
pidExists: processExists,
|
||||
processGroupExists,
|
||||
controllerInstance: async function (this: ProcessProbe, bootId, touchedAt) {
|
||||
const db = registeredControllerInstanceDb();
|
||||
return db ? proveControllerInstance(db, bootId, touchedAt, { readOnly: this?.readOnly === true }) : null;
|
||||
},
|
||||
readStartedAt: async pid => {
|
||||
const startedAt = await observedProcessStartedAt(pid);
|
||||
return startedAt === null ? null : new Date(startedAt);
|
||||
@@ -230,6 +246,23 @@ export async function proveOwnerGone(
|
||||
}
|
||||
|
||||
async function proveLocalOwnerGone(stamp: LocalOwnerStamp, probe: ProcessProbe): Promise<OwnerProof> {
|
||||
const proof = await proveLocalProcessGone(stamp, probe);
|
||||
if (proof.verdict !== "unprovable" || !CROSS_SPACE_REASONS.has(proof.reason) || !stamp.bootId || !probe.controllerInstance
|
||||
|| validPid(stamp.processGroupId ?? null)) return proof;
|
||||
// Another host or PID space: the owner's server boot answers through its lease. Only a record the
|
||||
// server process itself held qualifies; agent processes in a recorded group may outlive it.
|
||||
const instance = await probe.controllerInstance(stamp.bootId, stamp.touchedAt ? new Date(stamp.touchedAt) : null);
|
||||
if (instance === "stopped") return { verdict: "gone", reason: "controller_stopped" };
|
||||
if (instance === "fenced") return { verdict: "gone", reason: "controller_fenced" };
|
||||
if (instance === "live") return { verdict: "alive", reason: "controller_lease_held" };
|
||||
// Just past its last sign of life: kept as alive so the settle watch re-checks it after the grace.
|
||||
if (instance === "grace") return { verdict: "alive", reason: "controller_lease_grace" };
|
||||
return proof;
|
||||
}
|
||||
|
||||
const CROSS_SPACE_REASONS = new Set<OwnerProofReason>(["different_process_space", "different_host", "unknown_process_space"]);
|
||||
|
||||
async function proveLocalProcessGone(stamp: LocalOwnerStamp, probe: ProcessProbe): Promise<OwnerProof> {
|
||||
if (stamp.bootId && stamp.bootId === probe.bootId) return { verdict: "alive", reason: "current_process" };
|
||||
// The platform stopped the owner's whole deployment, with every process and
|
||||
// orphaned child in it, before it started this one. Host and PID are moot.
|
||||
|
||||
@@ -10,7 +10,7 @@ import { taskClosedByAgentSql, TEMPORARY_REFUSAL_TRIGGERS } from "../work-intent
|
||||
import { deliveredQueuedCommentIds } from "../queued-comment-delivery.js";
|
||||
import { logger } from "../../middleware/logger.js";
|
||||
import { persistActivity } from "../activity-log.js";
|
||||
import { proveOwnerGone, type OwnerProof, type ProcessProbe } from "../owner-proof.js";
|
||||
import { currentProcessProbe, proveOwnerGone, type OwnerProof, type ProcessProbe } from "../owner-proof.js";
|
||||
import {
|
||||
OWNABLE_KINDS,
|
||||
resolveFenceHolders,
|
||||
@@ -176,9 +176,10 @@ async function classifyRunOwners(db: Db, records: InProgressRecord[]): Promise<M
|
||||
}
|
||||
|
||||
async function classify(owner: RecordOwner, runProofs: Map<string, Classification>, probe: ProcessProbe | undefined,
|
||||
now: Date): Promise<Classification> {
|
||||
now: Date, touchedAt: Date | null = null): Promise<Classification> {
|
||||
switch (owner.kind) {
|
||||
case "process": return proveOwnerGone(owner.stamp, probe);
|
||||
// The record's last write time lets a replaced server boot be fenced only for work it touched under a valid lease.
|
||||
case "process": return proveOwnerGone(touchedAt ? { ...owner.stamp, touchedAt: touchedAt.toISOString() } : owner.stamp, probe);
|
||||
case "fence": return proveOwnerGone(owner.stamp, owner.fence, now);
|
||||
case "run": return runProofs.get(owner.runId) ?? { verdict: "unprovable", reason: "owner_run_unread" };
|
||||
case "issue": return runProofs.get(`issue:${owner.issueId}`) ?? { verdict: "unprovable", reason: "owner_issue_unread" };
|
||||
@@ -272,7 +273,7 @@ async function settleKind(db: Db, kindId: string, options: {
|
||||
const settled: SettledRecord[] = [];
|
||||
let watched = 0;
|
||||
for (const record of unique) {
|
||||
let proof = await classify(record.owner, runProofs, options.probe, now);
|
||||
let proof = await classify(record.owner, runProofs, options.probe, now, record.updatedAt);
|
||||
if (options.currentProcessIdle && record.runId && proof.verdict === "alive" && proof.reason === "current_process") {
|
||||
proof = { verdict: "gone", reason: "current_process_idle" };
|
||||
}
|
||||
@@ -588,6 +589,8 @@ export async function runOwnershipReconciliationPass(db: Db, options: {
|
||||
const goneThisPass = new Set<string>();
|
||||
let queueDueAt: Date | null = null;
|
||||
const settleRecords = options.settle ?? options.record !== false;
|
||||
// A pass that settles nothing must not fence another boot's lease either.
|
||||
const passProbe = settleRecords ? options.probe : { ...(options.probe ?? currentProcessProbe()), readOnly: true };
|
||||
let retryNeeded = false;
|
||||
|
||||
for (const kind of kinds) {
|
||||
@@ -618,7 +621,7 @@ export async function runOwnershipReconciliationPass(db: Db, options: {
|
||||
report.inProgress = unique.length;
|
||||
const runProofs = await classifyRunOwners(db, unique);
|
||||
for (const record of unique) {
|
||||
const proof = await classify(record.owner, runProofs, options.probe, now);
|
||||
const proof = await classify(record.owner, runProofs, passProbe, now, record.updatedAt);
|
||||
report.verdicts[proof.verdict] += 1;
|
||||
report.reasons[proof.reason] = (report.reasons[proof.reason] ?? 0) + 1;
|
||||
const base = { kind: kind.id, recordId: record.recordId, companyId: record.companyId,
|
||||
|
||||
@@ -45,6 +45,16 @@ export function registerTaskDrainExpiryWork(
|
||||
};
|
||||
}
|
||||
|
||||
// Set while this server cannot renew its controller instance lease: it stops
|
||||
// dispatching new work until the lease is renewed (or it exits). It never
|
||||
// replaces an operator's drain and has no idle owner, so HTTP stays open.
|
||||
let controllerLeaseQuiesce: TaskDrainState | null = null;
|
||||
export function setControllerLeaseQuiesce(quiesced: boolean): void {
|
||||
if (quiesced === (controllerLeaseQuiesce !== null)) return;
|
||||
controllerLeaseQuiesce = quiesced ? { startedAt: new Date(), expiresAt: null } : null;
|
||||
notifyDrainChanged();
|
||||
}
|
||||
|
||||
export function readTaskDrain(now: Date): TaskDrainState | null {
|
||||
if (taskDrainState?.expiresAt && taskDrainState.expiresAt.getTime() <= now.getTime()) {
|
||||
taskDrainState = null;
|
||||
@@ -52,7 +62,7 @@ export function readTaskDrain(now: Date): TaskDrainState | null {
|
||||
try { listener(); } catch (error) { process.emitWarning(`Task drain expiry notification failed: ${String(error)}`); }
|
||||
}
|
||||
}
|
||||
return taskDrainState;
|
||||
return taskDrainState ?? controllerLeaseQuiesce;
|
||||
}
|
||||
|
||||
export function computeTaskDrain(opts: { ttlMs?: number | null; purpose?: "idle" } = {}): TaskDrainState {
|
||||
@@ -71,7 +81,7 @@ export function startTaskDrain(opts: { ttlMs?: number | null; purpose?: "idle" }
|
||||
return drain;
|
||||
}
|
||||
export function stopTaskDrain(): { wasActive: boolean } {
|
||||
const wasActive = readTaskDrain(new Date()) !== null;
|
||||
const wasActive = readTaskDrain(new Date()) !== null && taskDrainState !== null;
|
||||
taskDrainState = null;
|
||||
notifyDrainChanged();
|
||||
return { wasActive };
|
||||
@@ -83,6 +93,12 @@ export function isIdleTaskDrainActive(): boolean { return Boolean(readTaskDrain(
|
||||
// The generation also catches work which starts AND finishes during a scan.
|
||||
let generation = 0;
|
||||
let active = 0;
|
||||
const settledListeners = new Set<() => void>();
|
||||
/** Called (synchronously) whenever the last tracked work finishes. */
|
||||
export function subscribeIdleWorkSettled(listener: () => void): () => void {
|
||||
settledListeners.add(listener);
|
||||
return () => { settledListeners.delete(listener); };
|
||||
}
|
||||
export function beginIdleTrackedWork(): () => void {
|
||||
generation++;
|
||||
active++;
|
||||
@@ -92,6 +108,9 @@ export function beginIdleTrackedWork(): () => void {
|
||||
finished = true;
|
||||
active--;
|
||||
generation++;
|
||||
if (active === 0) for (const listener of settledListeners) {
|
||||
try { listener(); } catch { /* a listener never blocks work accounting */ }
|
||||
}
|
||||
};
|
||||
}
|
||||
export function trackIdleWork<T>(work: Promise<T>): Promise<T> {
|
||||
|
||||
Reference in new issue
Block a user