fix: protect starting runs during overlapping deployments (#13285)
## Thinking Path > - Paperclip controls agent work across service deployments. > - A run can provision a remote sandbox before a process or invocation event exists. > - Each container previously treated its own missing process handle as proof that the run was orphaned. > - Overlapping deployments could therefore fail a run owned by another container. > - This pull request records and renews a controller lease before provisioning. > - A recovery worker must revoke an expired owner before it finalizes the run. ## Linked Issues or Issue Description Merged PR #13272 records startup adapter identity and restores explicit user continuation. This PR adds controller ownership on top of current master. Refs #7997 and #10442 for related replica and ownership problems. Related #13138 addresses silence and detached local processes; this change does not infer death from silence. **What happened?** During an overlapping hosted service deployment, a new container reaped a legacy conversation run that another container was provisioning. The run had no PID or adapter invocation yet. **Expected behavior** A live controller keeps its run. After controller loss, one recovery worker takes cleanup authority and the old controller cannot dispatch further work. **Steps to reproduce** Claim a legacy run in controller A. Start controller B against the same database before A finishes provisioning. Run the startup reaper in B. **Paperclip version or commit** Observed on `663c44cb2b9c28336d38d0b4a6971f4f1964bce6` in a hosted Railway deployment with a Daytona environment. ## What Changed - Add nullable controller boot ID, lease deadline, and execution stage columns. Claim ownership in the queued-to-running update. - Renew ownership independently of run output. Abort and reject dispatch if renewal fails. - Serialize reaper revocation against renewal. Let unfinished recovery claims expire after a restart. - Restrict graceful shutdown to legacy runs owned by the current controller. - Hand ownership back to the existing native coordinator when runtime selection becomes native. - Add twelve database regressions and document the lease contract. Update the task-drain regression to require controller expiry before reaping. ## Verification - `pnpm exec vitest run server/src/services/legacy-controller-lease.test.ts server/src/__tests__/heartbeat-task-drain-admission-release.test.ts`: 14 passed after rebasing onto master (`f12b647ae`). - Queue-interruption regressions in `heartbeat-process-recovery.test.ts`: 2 passed after preserving the new cleanup promotion from #13275. - `pnpm --filter @paperclipai/server exec tsc --noEmit`: passed after rebuilding runner TypeScript outputs for the updated master. Broad local tests are omitted at the maintainer’s request; CI owns broad coverage. - Latest-head CI passed on `f255e8e4d5ab2b24b12638a02434e6aa8a2285c5`: [run 34658248569](https://github.com/paperclipai/paperclip/actions/runs/34658248569). All test shards, browser suites, typecheck, build, canary, and security checks passed. Greptile is 5/5 with no unresolved review threads. ## Risks - Additive, idempotent migration; historical rows retain the previous recovery behavior. - Database unavailability aborts new dispatch rather than permitting an unfenced controller to continue. - Lease expiry is permission to clean up, not evidence that remote inference stopped. Follow-up PRs add persistent cleanup and automatic continuation. - Mixed-version deployment still includes old binaries whose reapers do not understand controller leases. ## Model Used OpenAI GPT-6 through Codex, using reasoning, repository inspection, code execution, and test tools. The precise backend revision and context-window size are not exposed in this session. ## 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 - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Nicky Leach
GitHub
parent
f12b647ae8
commit
dbf5ea432d
|
|
@ -386,3 +386,14 @@ pnpm secrets:migrate-inline-env --apply
|
|||
```
|
||||
|
||||
Hosted AWS provider notes live in [SECRETS-AWS-PROVIDER.md](./SECRETS-AWS-PROVIDER.md).
|
||||
|
||||
## Legacy controller ownership
|
||||
|
||||
Legacy run claims atomically record `controller_boot_id`, a database-clock
|
||||
`controller_lease_expires_at`, and `execution_stage` before workspace provisioning.
|
||||
The lease renews independently of output. A different container must not infer
|
||||
controller death from its own process map or numeric PIDs. Expiration grants
|
||||
cleanup authority; it does not prove that remote inference has stopped. Recovery
|
||||
revokes the previous boot identity with a conditional update. Its own claim also
|
||||
expires so another sweep can finish cleanup after a restart. Historical rows keep
|
||||
null ownership fields and follow the previous recovery path.
|
||||
|
|
|
|||
|
|
@ -1261,6 +1261,11 @@ Scheduler must skip invocation when:
|
|||
- an existing run is active
|
||||
- hard budget limit has been hit
|
||||
|
||||
Legacy execution records a renewable controller lease when claiming a queued run,
|
||||
before provisioning. A live lease protects the run during overlapping service
|
||||
deployments. An expired controller loses dispatch authority; a recovery worker
|
||||
must establish that the previous execution stopped before starting a successor.
|
||||
|
||||
## 11.7 Durable agent session goals
|
||||
|
||||
Runner Protocol v2 negotiates a required `sessionGoals` capability and typed
|
||||
|
|
|
|||
|
|
@ -0,0 +1,3 @@
|
|||
ALTER TABLE "heartbeat_runs" ADD COLUMN IF NOT EXISTS "controller_boot_id" uuid;--> statement-breakpoint
|
||||
ALTER TABLE "heartbeat_runs" ADD COLUMN IF NOT EXISTS "controller_lease_expires_at" timestamp with time zone;--> statement-breakpoint
|
||||
ALTER TABLE "heartbeat_runs" ADD COLUMN IF NOT EXISTS "execution_stage" text;
|
||||
File diff suppressed because it is too large
Load Diff
|
|
@ -1898,6 +1898,13 @@
|
|||
"when": 1789137216452,
|
||||
"tag": "0272_light_kate_bishop",
|
||||
"breakpoints": true
|
||||
},
|
||||
{
|
||||
"idx": 273,
|
||||
"version": "7",
|
||||
"when": 1789164595203,
|
||||
"tag": "0273_aromatic_moondragon",
|
||||
"breakpoints": true
|
||||
}
|
||||
]
|
||||
}
|
||||
|
|
@ -67,6 +67,10 @@ export const heartbeatRuns = pgTable(
|
|||
stderrExcerpt: text("stderr_excerpt"),
|
||||
errorCode: text("error_code"),
|
||||
externalRunId: text("external_run_id"),
|
||||
// Legacy controller lease. A PID alone is not an identity across containers.
|
||||
controllerBootId: uuid("controller_boot_id"),
|
||||
controllerLeaseExpiresAt: timestamp("controller_lease_expires_at", { withTimezone: true }),
|
||||
executionStage: text("execution_stage"),
|
||||
processPid: integer("process_pid"),
|
||||
processGroupId: integer("process_group_id"),
|
||||
processStartedAt: timestamp("process_started_at", { withTimezone: true }),
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
import { randomUUID } from "node:crypto";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
|
||||
import {
|
||||
activityLog,
|
||||
|
|
@ -311,8 +311,14 @@ describeEmbeddedPostgres("heartbeat task-drain admission release", () => {
|
|||
expect(status.activeRuns).toBe(0);
|
||||
expect(status.quiescent).toBe(true);
|
||||
|
||||
// The run's row is still "running", so the orphan reaper finds it,
|
||||
// finalizes it, and releases the issue lock on its own cycle.
|
||||
// Missing local tracking cannot override the durable controller lease.
|
||||
// Once that unrenewed lease expires, the reaper finalizes the orphan and
|
||||
// releases the issue lock on its own cycle.
|
||||
const beforeExpiry = await heartbeat.reapOrphanedRuns();
|
||||
expect(beforeExpiry.runIds).not.toContain(runId);
|
||||
await db.update(heartbeatRuns).set({
|
||||
controllerLeaseExpiresAt: sql`clock_timestamp() - interval '1 second'`,
|
||||
}).where(eq(heartbeatRuns.id, runId));
|
||||
const reapResult = await heartbeat.reapOrphanedRuns();
|
||||
expect(reapResult.runIds).toContain(runId);
|
||||
|
||||
|
|
|
|||
|
|
@ -1,4 +1,5 @@
|
|||
import { PROCESS_IDENTITY_RECORDED, recordNativeLocalProcessStop } from "./native-local-process-stop.js";
|
||||
import { legacyControllerBootId, legacyControllerClaim, hasLiveLegacyController, revokeExpiredLegacyController, watchLegacyControllerLease } from "./legacy-controller-lease.js";
|
||||
import { completeTerminatedRemoteNativeSessionCleanup } from "../vendor/paperclip-runner/index.js";
|
||||
import { remoteExecutionHasStopped, remoteTerminationReceipt, stoppedRemoteCleanupScopes } from "./remote-execution-termination.js";
|
||||
import { applyConnectorSkills, prepareConnectorSkillDelivery, resolveConnectorAssignments } from "./connector-runtime.js";
|
||||
|
|
@ -14560,6 +14561,10 @@ export function heartbeatService(
|
|||
const restartSuspendedRunIds: string[] = [];
|
||||
|
||||
for (const { run, agent } of activeRuns) {
|
||||
// Shutdown owns only this boot's legacy executions. Expired foreign
|
||||
// owners belong to the reaper, not another container's drain.
|
||||
if (run.runtimeMode === "legacy" && run.controllerBootId &&
|
||||
run.controllerBootId !== legacyControllerBootId) continue;
|
||||
if (isNativeRunnerOwnershipHeld(run)) continue;
|
||||
if (
|
||||
run.runtimeMode === "native" &&
|
||||
|
|
@ -16921,6 +16926,7 @@ export function heartbeatService(
|
|||
.set({
|
||||
status: "running",
|
||||
runnerProfileJson: sql`(case when jsonb_typeof(${heartbeatRuns.runnerProfileJson}) = 'object' then ${heartbeatRuns.runnerProfileJson} else '{}'::jsonb end) || ${JSON.stringify({ adapterDispatch: { adapterType: agent.adapterType } })}::jsonb`,
|
||||
...legacyControllerClaim(run.runtimeMode),
|
||||
responsibleUserId,
|
||||
startedAt: lockedRun.startedAt ?? claimedAt,
|
||||
updatedAt: claimedAt,
|
||||
|
|
@ -17018,6 +17024,7 @@ export function heartbeatService(
|
|||
.set({
|
||||
status: "running",
|
||||
runnerProfileJson: sql`(case when jsonb_typeof(${heartbeatRuns.runnerProfileJson}) = 'object' then ${heartbeatRuns.runnerProfileJson} else '{}'::jsonb end) || ${JSON.stringify({ adapterDispatch: { adapterType: agent.adapterType } })}::jsonb`,
|
||||
...legacyControllerClaim(run.runtimeMode),
|
||||
responsibleUserId,
|
||||
startedAt: lockedRun.startedAt ?? claimedAt,
|
||||
contextSnapshot: withQueuedCommentIdsInRunContext(
|
||||
|
|
@ -17085,6 +17092,7 @@ export function heartbeatService(
|
|||
.set({
|
||||
status: "running",
|
||||
runnerProfileJson: sql`(case when jsonb_typeof(${heartbeatRuns.runnerProfileJson}) = 'object' then ${heartbeatRuns.runnerProfileJson} else '{}'::jsonb end) || ${JSON.stringify({ adapterDispatch: { adapterType: agent.adapterType } })}::jsonb`,
|
||||
...legacyControllerClaim(run.runtimeMode),
|
||||
responsibleUserId,
|
||||
startedAt: run.startedAt ?? claimedAt,
|
||||
updatedAt: claimedAt,
|
||||
|
|
@ -18371,6 +18379,7 @@ export function heartbeatService(
|
|||
}
|
||||
if (resumedRunIds.has(run.id)) continue;
|
||||
if (locallyTracked) continue;
|
||||
if (await hasLiveLegacyController(db, run)) continue;
|
||||
|
||||
// Apply staleness threshold to avoid false positives
|
||||
if (staleThresholdMs > 0) {
|
||||
|
|
@ -18442,6 +18451,7 @@ export function heartbeatService(
|
|||
((tracksLegacyLocalChild &&
|
||||
(!!run.processPid || !!run.processGroupId)) ||
|
||||
monitorDispatchLostWithoutFutureWake);
|
||||
if (!(await revokeExpiredLegacyController(db, run))) continue;
|
||||
const baseMessage = buildProcessLossMessage(run);
|
||||
const conversationContinuationEligible = await runUsedConversationAdapter(db, run);
|
||||
|
||||
|
|
@ -19275,8 +19285,11 @@ export function heartbeatService(
|
|||
}
|
||||
}
|
||||
|
||||
if (run.runtimeMode === "legacy" && run.controllerBootId &&
|
||||
run.controllerBootId !== legacyControllerBootId) return;
|
||||
activeRunExecutions.add(run.id);
|
||||
const executionControl = createAdapterExecutionControl();
|
||||
const controllerLease = watchLegacyControllerLease(db, run, executionControl.controller);
|
||||
let runScratch: HeartbeatRunScratch | null = null;
|
||||
let githubLauncherLocation:
|
||||
Parameters<typeof cleanupGitHubOperationLaunchers>[0] | null = null;
|
||||
|
|
@ -21100,6 +21113,7 @@ export function heartbeatService(
|
|||
ReturnType<typeof envOrchestrator.acquireForRun>
|
||||
>;
|
||||
try {
|
||||
await controllerLease.assertOwned();
|
||||
acquiredEnvironment = await envOrchestrator.acquireForRun({
|
||||
companyId: agent.companyId,
|
||||
selectedEnvironmentId,
|
||||
|
|
@ -21111,6 +21125,7 @@ export function heartbeatService(
|
|||
persistedExecutionWorkspace,
|
||||
executionWorkspaceSettings: environmentExecutionWorkspaceSettings,
|
||||
});
|
||||
await controllerLease.assertOwned();
|
||||
nativeRunnerPreparationSpans.push({
|
||||
name: "environment.acquire",
|
||||
parentName: "task.run",
|
||||
|
|
@ -21250,6 +21265,7 @@ export function heartbeatService(
|
|||
): Promise<
|
||||
{ dispatched: true; resultPromise: Promise<T> } | { dispatched: false }
|
||||
> => {
|
||||
await controllerLease.assertOwned("dispatching");
|
||||
// Recheck after workspace/credential preparation, immediately before the
|
||||
// provider handoff. Never hold validation locks while adapter code runs.
|
||||
await authorizeFailedChatRetryExecution();
|
||||
|
|
@ -22631,6 +22647,7 @@ export function heartbeatService(
|
|||
})
|
||||
.onConflictDoNothing();
|
||||
});
|
||||
controllerLease.stop();
|
||||
nativeWorkspaceSync = await prepareNativeWorkspaceSync({
|
||||
db,
|
||||
runId: run.id,
|
||||
|
|
@ -24922,6 +24939,7 @@ export function heartbeatService(
|
|||
logger.error({ err, runId: run.id }, "failed to promote interrupted comment queue after cleanup");
|
||||
});
|
||||
}
|
||||
controllerLease.stop();
|
||||
activeRunExecutions.delete(run.id);
|
||||
// A failed owned Stop remains visible until this exact executor settles,
|
||||
// including a graceful exit result arriving after the cancellation error.
|
||||
|
|
|
|||
|
|
@ -0,0 +1,120 @@
|
|||
import { randomUUID } from "node:crypto";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { agents, companies, createDb, heartbeatRuns } from "@paperclipai/db";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js";
|
||||
import { heartbeatService } from "./heartbeat.js";
|
||||
import { hasLiveLegacyController, legacyControllerBootId, legacyControllerClaim,
|
||||
renewLegacyControllerLease, revokeExpiredLegacyController, watchLegacyControllerLease } from "./legacy-controller-lease.js";
|
||||
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
(support.supported ? describe : describe.skip)("durable legacy controller ownership", () => {
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
beforeAll(async () => {
|
||||
database = await startEmbeddedPostgresTestDatabase("legacy-controller-");
|
||||
db = createDb(database.connectionString);
|
||||
}, 30000);
|
||||
afterAll(async () => { await database?.cleanup(); });
|
||||
async function seed() {
|
||||
const companyId = randomUUID(), agentId = randomUUID();
|
||||
await db.insert(companies).values({ id: companyId, name: "Controller test", issuePrefix: `C${companyId.slice(0, 7)}` });
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Agent", role: "general", adapterType: "claude_local", status: "idle" });
|
||||
const [queued] = await db.insert(heartbeatRuns).values({ companyId, agentId }).returning();
|
||||
const [run] = await db.update(heartbeatRuns).set({ status: "running", ...legacyControllerClaim("legacy") })
|
||||
.where(eq(heartbeatRuns.id, queued.id)).returning();
|
||||
return run;
|
||||
}
|
||||
async function expire(id: string) {
|
||||
await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: sql`clock_timestamp() - interval '1 second'` })
|
||||
.where(eq(heartbeatRuns.id, id));
|
||||
}
|
||||
it("commits ownership with the queued claim before provisioning or logs exist", async () => {
|
||||
const run = await seed();
|
||||
expect(run).toMatchObject({ status: "running", controllerBootId: legacyControllerBootId, executionStage: "preparing", processPid: null });
|
||||
expect(await hasLiveLegacyController(db, run)).toBe(true);
|
||||
expect(await revokeExpiredLegacyController(db, run)).toBe(false);
|
||||
});
|
||||
it("another deployment's startup reaper preserves an unexpired controller", async () => {
|
||||
const run = await seed();
|
||||
await db.update(heartbeatRuns).set({ controllerBootId: randomUUID() }).where(eq(heartbeatRuns.id, run.id));
|
||||
await heartbeatService(db).reapOrphanedRuns({ staleThresholdMs: 0 });
|
||||
const [saved] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
|
||||
expect(saved.status).toBe("running");
|
||||
expect(saved.errorCode).toBeNull();
|
||||
});
|
||||
it.each([false, true])("shutdown preserves a foreign controller (expired: %s)", async expired => {
|
||||
const run = await seed();
|
||||
await db.update(heartbeatRuns).set({ controllerBootId: randomUUID() }).where(eq(heartbeatRuns.id, run.id));
|
||||
if (expired) await expire(run.id);
|
||||
const result = await heartbeatService(db).drainRunningRunsForShutdown("SIGTERM", new Date(), [run.id]);
|
||||
expect(result.interruptedRunIds).toEqual([]);
|
||||
const [saved] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
|
||||
expect(saved.status).toBe("running");
|
||||
});
|
||||
it("a current controller renews and records the dispatch boundary", async () => {
|
||||
const run = await seed();
|
||||
expect(await renewLegacyControllerLease(db, run, "dispatching")).toBe(true);
|
||||
const [saved] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
|
||||
expect(saved.executionStage).toBe("dispatching");
|
||||
expect(await revokeExpiredLegacyController(db, run)).toBe(false);
|
||||
});
|
||||
it("an expired controller cannot renew or dispatch even before a reaper claims it", async () => {
|
||||
const run = await seed();
|
||||
await expire(run.id);
|
||||
expect(await renewLegacyControllerLease(db, run, "dispatching")).toBe(false);
|
||||
const controller = new AbortController();
|
||||
const watch = watchLegacyControllerLease(db, run, controller);
|
||||
try {
|
||||
await expect(watch.assertOwned("dispatching")).rejects.toThrow("lease lost");
|
||||
expect(controller.signal.aborted).toBe(true);
|
||||
} finally { watch.stop(); }
|
||||
});
|
||||
it("only one competing recovery revokes the observed expired owner", async () => {
|
||||
const run = await seed();
|
||||
await expire(run.id);
|
||||
const attempts = await Promise.all([revokeExpiredLegacyController(db, run), revokeExpiredLegacyController(db, run)]);
|
||||
expect(attempts.filter(Boolean)).toHaveLength(1);
|
||||
expect(await renewLegacyControllerLease(db, run)).toBe(false);
|
||||
});
|
||||
it("a crash after revocation permits a later recovery claim", async () => {
|
||||
const run = await seed();
|
||||
await expire(run.id);
|
||||
expect(await revokeExpiredLegacyController(db, run)).toBe(true);
|
||||
const [claimed] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
|
||||
expect(await hasLiveLegacyController(db, claimed)).toBe(true);
|
||||
expect(await revokeExpiredLegacyController(db, claimed)).toBe(false);
|
||||
await expire(run.id);
|
||||
const [saved] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
|
||||
expect(await revokeExpiredLegacyController(db, saved)).toBe(true);
|
||||
});
|
||||
it("rejects foreign company renewal and revocation", async () => {
|
||||
const run = await seed();
|
||||
expect(await renewLegacyControllerLease(db, { ...run, companyId: randomUUID() })).toBe(false);
|
||||
await expire(run.id);
|
||||
expect(await revokeExpiredLegacyController(db, { ...run, companyId: randomUUID() })).toBe(false);
|
||||
});
|
||||
it("never turns a terminal run back into owned execution", async () => {
|
||||
const run = await seed();
|
||||
await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, run.id));
|
||||
expect(await renewLegacyControllerLease(db, run)).toBe(false);
|
||||
expect(await revokeExpiredLegacyController(db, run)).toBe(false);
|
||||
});
|
||||
it("rejects dispatch at the lease deadline even if the database query never settles", async () => {
|
||||
const run = await seed();
|
||||
const hungDb = { update: () => ({ set: () => ({ where: () => ({ returning: () => new Promise(() => {}) }) }) }) } as unknown as typeof db;
|
||||
vi.useFakeTimers();
|
||||
const controller = new AbortController();
|
||||
const watch = watchLegacyControllerLease(hungDb, { ...run, controllerLeaseExpiresAt: new Date(Date.now() + 100) }, controller);
|
||||
try {
|
||||
const checked = expect(watch.assertOwned("dispatching")).rejects.toThrow("lease lost");
|
||||
await vi.advanceTimersByTimeAsync(101);
|
||||
await checked;
|
||||
expect(controller.signal.aborted).toBe(true);
|
||||
} finally { watch.stop(); vi.useRealTimers(); }
|
||||
});
|
||||
|
||||
it("leaves native controller ownership to the native coordinator", () => {
|
||||
expect(legacyControllerClaim("native")).toEqual({});
|
||||
});
|
||||
});
|
||||
|
|
@ -0,0 +1,111 @@
|
|||
import { randomUUID } from "node:crypto";
|
||||
import { and, eq, gt, lte, sql } from "drizzle-orm";
|
||||
import { heartbeatRuns, type Db } from "@paperclipai/db";
|
||||
|
||||
// A boot UUID has meaning across containers; a numeric PID does not.
|
||||
export const legacyControllerBootId = randomUUID();
|
||||
export const LEGACY_CONTROLLER_LEASE_MS = 60_000;
|
||||
export const LEGACY_CONTROLLER_RENEW_MS = 10_000;
|
||||
|
||||
type Run = typeof heartbeatRuns.$inferSelect;
|
||||
|
||||
/** Commit these fields in the same UPDATE that claims a queued run. */
|
||||
export function legacyControllerClaim(runtimeMode: string) {
|
||||
if (runtimeMode === "native") return {};
|
||||
return {
|
||||
controllerBootId: legacyControllerBootId,
|
||||
controllerLeaseExpiresAt: sql`clock_timestamp() + interval '60 seconds'`,
|
||||
executionStage: "preparing",
|
||||
};
|
||||
}
|
||||
|
||||
export async function renewLegacyControllerLease(
|
||||
db: Db,
|
||||
run: Pick<Run, "id" | "companyId" | "controllerBootId">,
|
||||
stage?: "dispatching",
|
||||
): Promise<boolean> {
|
||||
const [renewed] = await db.update(heartbeatRuns).set({
|
||||
controllerLeaseExpiresAt: sql`clock_timestamp() + interval '60 seconds'`,
|
||||
...(stage ? { executionStage: stage } : {}),
|
||||
}).where(and(
|
||||
eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId),
|
||||
eq(heartbeatRuns.runtimeMode, "legacy"), eq(heartbeatRuns.status, "running"),
|
||||
eq(heartbeatRuns.controllerBootId, legacyControllerBootId),
|
||||
gt(heartbeatRuns.controllerLeaseExpiresAt, sql`clock_timestamp()`),
|
||||
)).returning({ id: heartbeatRuns.id });
|
||||
return Boolean(renewed);
|
||||
}
|
||||
|
||||
export async function hasLiveLegacyController(db: Db, run: Run): Promise<boolean> {
|
||||
if (run.runtimeMode === "native" || !run.controllerBootId) return false;
|
||||
const [owner] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId),
|
||||
eq(heartbeatRuns.status, "running"),
|
||||
gt(heartbeatRuns.controllerLeaseExpiresAt, sql`clock_timestamp()`),
|
||||
));
|
||||
return Boolean(owner);
|
||||
}
|
||||
|
||||
/** Atomically revoke an expired controller. Renewal and revocation serialize on
|
||||
* the run row. Expiry permits cleanup, never dispatch of a replacement agent. */
|
||||
export async function revokeExpiredLegacyController(db: Db, run: Run): Promise<boolean> {
|
||||
if (run.runtimeMode === "native" || !run.controllerBootId) return true;
|
||||
const [revoked] = await db.update(heartbeatRuns).set({
|
||||
controllerBootId: randomUUID(),
|
||||
controllerLeaseExpiresAt: sql`clock_timestamp() + interval '60 seconds'`,
|
||||
}).where(and(
|
||||
eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId),
|
||||
eq(heartbeatRuns.status, "running"),
|
||||
eq(heartbeatRuns.controllerBootId, run.controllerBootId),
|
||||
lte(heartbeatRuns.controllerLeaseExpiresAt, sql`clock_timestamp()`),
|
||||
)).returning({ id: heartbeatRuns.id });
|
||||
return Boolean(revoked);
|
||||
}
|
||||
|
||||
/** Abort the adapter if the controller cannot renew. Bound each check by the
|
||||
* lease duration even when the database connection never settles. */
|
||||
export function watchLegacyControllerLease(db: Db, run: Run, controller: AbortController) {
|
||||
if (run.runtimeMode === "native" || !run.controllerBootId) {
|
||||
return { stop() {}, async assertOwned(_stage?: "dispatching") {} };
|
||||
}
|
||||
let stopped = false;
|
||||
let pending = false;
|
||||
const lost = () => { if (!stopped) controller.abort(new Error("Legacy controller lease lost")); };
|
||||
let deadline = setTimeout(lost, Math.max(0,
|
||||
(run.controllerLeaseExpiresAt?.getTime() ?? 0) - Date.now()));
|
||||
deadline.unref();
|
||||
const assertOwned = async (stage?: "dispatching") => {
|
||||
if (stopped) return;
|
||||
controller.signal.throwIfAborted();
|
||||
const startedAt = Date.now();
|
||||
let onAbort!: () => void;
|
||||
const aborted = new Promise<never>((_, reject) => {
|
||||
onAbort = () => reject(controller.signal.reason);
|
||||
controller.signal.addEventListener("abort", onAbort, { once: true });
|
||||
});
|
||||
let renewed: boolean;
|
||||
try {
|
||||
renewed = await Promise.race([renewLegacyControllerLease(db, run, stage), aborted]);
|
||||
} finally {
|
||||
controller.signal.removeEventListener("abort", onAbort);
|
||||
}
|
||||
if (stopped) return;
|
||||
if (!renewed) {
|
||||
lost();
|
||||
controller.signal.throwIfAborted();
|
||||
}
|
||||
controller.signal.throwIfAborted();
|
||||
if (!stopped) {
|
||||
clearTimeout(deadline);
|
||||
deadline = setTimeout(lost, Math.max(0, LEGACY_CONTROLLER_LEASE_MS - (Date.now() - startedAt)));
|
||||
deadline.unref();
|
||||
}
|
||||
};
|
||||
const timer = setInterval(() => {
|
||||
if (pending || stopped) return;
|
||||
pending = true;
|
||||
void assertOwned().catch(lost).finally(() => { pending = false; });
|
||||
}, LEGACY_CONTROLLER_RENEW_MS);
|
||||
timer.unref();
|
||||
return { assertOwned, stop() { stopped = true; clearInterval(timer); clearTimeout(deadline); } };
|
||||
}
|
||||
Loading…
Reference in New Issue