mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-09 07:10:34 +01:00
fix(heartbeat): finalize an accepted native result after a graceful shutdown (#19591)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Native runner runs go through a durable coordinator row (`native_run_finalizations`) that holds the controller's execution lease, the accepted result, and the finalization phase > - A graceful server shutdown (SIGTERM) drains running native runs: it detaches the executor and hands the run to the next server boot > - When SIGTERM lands after `run.result.accepted` / `run.terminal` but before finalization, the drain did not release the execution lease, because it only updated coordinator rows without a result > - Startup restart recovery skips accepted results, and both the finalization reconciler and the orphan reaper wait for an unexpired coordinator lease (20 minutes for a local run). The run stayed `running` and the task stayed `in_progress` with no Retry > - This pull request makes the drain release the execution lease of an accepted result too, so the next boot finalizes the durable result > - The benefit is that a completed task finishes after a restart instead of stalling silently ## Linked Issues or Issue Description Refs #19525, #19530, #19513 **What happened?** A restart-chaos wave sent SIGTERM to the server 0.75 s after a native OpenCode run emitted `run.terminal` (result accepted, `runTerminalState: succeeded`). The graceful drain listed the run in `restartSuspendedRunIds`. On the next boot, startup recovery claimed only another run. The accepted run stayed `running` for more than 10 minutes. Its task stayed `in_progress` with `interruption.state=working`, and no Retry appeared. **Expected behavior** After a restart, a run whose result was accepted and whose terminal state is known is finalized as succeeded, and its task is completed. It must never stay silently `running`. **Root cause** `drainRunningRunsForShutdown` (`server/src/services/heartbeat/recovery.ts`) updated the coordinator only `where result_id is null`. For an accepted result, the update did nothing, so the dead controller's execution lease (`NATIVE_SESSION_EXECUTION_LEASE_TTL_MS`, 20 min) stayed valid. The detached executor does not finalize (by design, it hands off to the next controller). Then: - `claimNativeRestartRecoveries` only claims rows with `result_id is null`. - `reconcileNativeFinalizations` only selects rows whose lease is null or expired. - The orphan reaper skips a native run with pending workspace finalization or an unexpired controller lease, and defers to the lease deadline. The 5-minute stale threshold is not involved. **Steps to reproduce** See the new regression test: accept a result through `PaperclipControlPlanePort.completeRun` under a held 20-minute execution lease, run the SIGTERM drain, then run startup recovery on a new heartbeat service. ## What Changed - `drainRunningRunsForShutdown` now releases the coordinator execution lease (`leaseExpiresAt = now`) for an accepted result as well. It keeps `recoveryState` unchanged for an accepted result, because the run needs finalization, not a runner reattach. Rows without a result keep the old behavior (`awaiting_runner_reattach`). - A `native-finalizer:` lease on an accepted result is kept. That lease belongs to a finalization that is in progress in the draining process, and the shutdown waits for it. - The drain event for an accepted result now says that the result was released for finalization (`runnerDisposition: awaiting_result_finalization`). - New regression tests in `server/src/__tests__/heartbeat-native-restart-suspension.test.ts`: - An accepted terminal result under a held execution lease, then SIGTERM drain, then next-boot recovery: the run becomes `succeeded`, the coordinator becomes `committed`, the task becomes `done`, there is one result row and one run, and no provider turn starts. - A `native-finalizer:` lease survives the drain unchanged. ## Verification - Red before the fix: `cd server && pnpm exec vitest run src/__tests__/heartbeat-native-restart-suspension.test.ts -t "accepted terminal"` failed with `{ status: 'running', phase: 'workspace_finalizing' }` instead of `{ status: 'succeeded', phase: 'committed' }`. - Green after the fix: the same file passes 5/5. - Related suites pass: `src/services/heartbeat/recovery.test.ts`, `src/services/legacy-controller-lease.test.ts`, `src/services/native-runtime/native-runner-restart-recovery.integration.test.ts` (80 passed, 14 skipped), and `src/__tests__/heartbeat-process-recovery.test.ts` except one case. - The one failing case in `heartbeat-process-recovery.test.ts` ("dispatches interrupted CLI input and cleans up early fixture exits (true)") also fails on unchanged `origin/master`. It is not related to this change. - `server` typecheck (`tsc --noEmit`) passes. - Restart-chaos replay of the original cell: in progress. Results follow in a PR comment. ## Risks - Low to medium. The change only affects the graceful SIGTERM drain of native runs owned by this boot. - A rolling deploy can overlap the old and new servers. The old process detaches the executor before it releases the lease, so it does not finalize the result. A finalizer that already started keeps its `native-finalizer:` lease, so the new server cannot finalize the same result at the same time. - If a `native-finalizer:` lease is in progress and the process exits before it finishes, that run still waits for the 5-minute finalizer lease. This was already the behavior and is out of scope. - No schema, API, or UI change. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used through Claude Code with tool use (shell, file edit, test runs). Extended reasoning was used. ## 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 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
1c9767d323
commit
5a5bfe689f
2 files changed
+121
-10
No files matched your search
@@ -1,9 +1,12 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { authUsers, budgetReservations, companyMemberships, heartbeatRuns, issues, nativeRunFinalizations, workspaceOperations } from "@paperclipai/db";
|
||||
import { authUsers, budgetReservations, companyMemberships, completionContracts, heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, workspaceOperations } from "@paperclipai/db";
|
||||
import { expect, it, vi } from "vitest";
|
||||
import { heartbeatService } from "../services/heartbeat.js";
|
||||
import { instanceSettingsService } from "../services/instance-settings.js";
|
||||
import { currentNativeControllerIdentity } from "../services/native-runtime/native-restart-recovery.js";
|
||||
import { PaperclipControlPlanePort } from "../services/native-runtime/paperclip-control-plane-port.js";
|
||||
import { CONTROL_PLANE_CONFORMANCE_OPEN, CONTROL_PLANE_CONFORMANCE_RESULT, CONTROL_PLANE_CONFORMANCE_TERMINAL } from "../vendor/paperclip-runner/testing.js";
|
||||
import { describeEmbeddedPostgres, nativeRestartFixture } from "./helpers/native-restart.js";
|
||||
|
||||
|
||||
@@ -92,4 +95,99 @@ describeEmbeddedPostgres("heartbeat native recovery interrupted by graceful shut
|
||||
const [run] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
expect(run.status).toBe("failed");
|
||||
}, 60_000);
|
||||
|
||||
|
||||
async function shutdownDispositions(db: Awaited<ReturnType<typeof seedRecoverableRun>>["db"], runId: string) {
|
||||
const events = await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, runId),
|
||||
eq(heartbeatRunEvents.eventType, "native.recovery.transition")));
|
||||
return events.map((event) => (event.payload as Record<string, unknown> | null)?.runnerDisposition);
|
||||
}
|
||||
|
||||
/** A run whose provider turn ended and whose result this boot's controller accepted. */
|
||||
async function seedAcceptedResult(label: string, leaseOwner: (runnerInstanceId: string) => string) {
|
||||
const f = await seedRecoverableRun(label);
|
||||
const [contract] = await f.db.select().from(completionContracts).where(eq(completionContracts.issueId, f.issueId));
|
||||
await f.db.update(heartbeatRuns).set({ completionContractId: contract.id,
|
||||
completionContractSha256: contract.canonicalSha256 }).where(eq(heartbeatRuns.id, f.runId));
|
||||
const controller = await currentNativeControllerIdentity();
|
||||
const leaseExpiresAt = new Date(Date.now() + 20 * 60_000);
|
||||
await f.db.update(nativeRunFinalizations).set({
|
||||
leaseOwner: leaseOwner(f.native.runnerInstanceId),
|
||||
leaseExpiresAt,
|
||||
controllerBootId: controller.bootId,
|
||||
controllerPid: controller.pid,
|
||||
controllerProcessStartedAt: controller.processStartedAt,
|
||||
controllerGeneration: 1,
|
||||
}).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
const [prepared] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
const identity = { companyId: fixture.current.companyId, issueId: f.issueId, runId: f.runId,
|
||||
agentId: fixture.current.agentId, sessionId: prepared.nativeSessionId! };
|
||||
const port = new PaperclipControlPlanePort(f.db, {
|
||||
...identity,
|
||||
completionContractId: prepared.completionContractId!,
|
||||
completionContractSha256: prepared.completionContractSha256!,
|
||||
sourceInstanceId: prepared.runnerInstanceId!,
|
||||
controlPlaneSourceInstanceId: `${label}-control`,
|
||||
});
|
||||
await port.openRun({ ...CONTROL_PLANE_CONFORMANCE_OPEN, identity, sourceInstanceId: prepared.runnerInstanceId! });
|
||||
const { revision, criterionIds } = f.native.completionContract;
|
||||
await port.completeRun({
|
||||
result: { ...CONTROL_PLANE_CONFORMANCE_RESULT, completionClaim: {
|
||||
...CONTROL_PLANE_CONFORMANCE_RESULT.completionClaim, contractRevision: revision,
|
||||
criteria: criterionIds.map((criterionId) => ({ criterionId, status: "satisfied" as const, evidenceRefs: ["event:2"] })),
|
||||
} },
|
||||
terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL,
|
||||
callerResultId: `${label}-result`,
|
||||
});
|
||||
// The detached controller had begun local workspace finalization.
|
||||
await f.db.insert(workspaceOperations).values({ companyId: fixture.current.companyId, issueId: f.issueId,
|
||||
heartbeatRunId: f.runId, phase: "workspace_finalize", status: "running", cwd: fixture.current.runtimeRoot });
|
||||
const [accepted] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
expect(accepted).toMatchObject({ phase: "workspace_finalizing", resultId: expect.any(String) });
|
||||
return { ...f, leaseExpiresAt };
|
||||
}
|
||||
|
||||
it("finalizes an accepted terminal result whose controller was suspended by a graceful shutdown", async () => {
|
||||
// RC-30: SIGTERM lands after run.result.accepted/run.terminal but before the
|
||||
// controller finalizes. The run must not stay `running` behind the old
|
||||
// controller's execution lease once the next boot starts recovery.
|
||||
const f = await seedAcceptedResult("ACCEPTED-THEN-SIGTERM", (runner) => `${runner}:${randomUUID()}`);
|
||||
const dying = heartbeatService(f.db);
|
||||
const drained = await dying.drainRunningRunsForShutdown("SIGTERM", new Date(), [f.runId]);
|
||||
expect(drained).toMatchObject({ restartSuspendedRunIds: [f.runId] });
|
||||
const [released] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
// Finalization, not a runner reattach, is what remains.
|
||||
expect(released.recoveryState).toBeNull();
|
||||
expect(await shutdownDispositions(f.db, f.runId)).toEqual(["awaiting_result_finalization"]);
|
||||
|
||||
const providerDispatch = vi.fn(() => { throw new Error("an accepted result must not open another provider turn"); });
|
||||
const next = heartbeatService(f.db, { nativeSessionBackendFactory: providerDispatch });
|
||||
try {
|
||||
await next.recoverNativeRunsAfterRestart(new Date());
|
||||
} finally { await next.drainActiveRunExecutions(); }
|
||||
expect(providerDispatch).not.toHaveBeenCalled();
|
||||
|
||||
const [run] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
const [owner] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
const [task] = await f.db.select().from(issues).where(eq(issues.id, f.issueId));
|
||||
expect({ status: run.status, phase: owner.phase }).toEqual({ status: "succeeded", phase: "committed" });
|
||||
expect(task.status).toBe("done");
|
||||
expect(await f.db.select({ id: nativeRunResults.id }).from(nativeRunResults)
|
||||
.where(eq(nativeRunResults.runId, f.runId))).toHaveLength(1);
|
||||
expect(await f.db.select({ id: heartbeatRuns.id }).from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.nativeIssueId, f.issueId))).toHaveLength(1);
|
||||
}, 90_000);
|
||||
|
||||
it("keeps a finalizer lease already settling the accepted result through a graceful shutdown", async () => {
|
||||
// The shutdown drain waits for in-process finalizers; releasing their
|
||||
// lease would let the next controller finalize the same result concurrently.
|
||||
const f = await seedAcceptedResult("FINALIZER-THEN-SIGTERM", () => `native-finalizer:${randomUUID()}`);
|
||||
const dying = heartbeatService(f.db);
|
||||
await dying.drainRunningRunsForShutdown("SIGTERM", new Date(), [f.runId]);
|
||||
const [owner] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
expect(owner).toMatchObject({ phase: "workspace_finalizing", leaseExpiresAt: f.leaseExpiresAt, recoveryState: null });
|
||||
expect(await shutdownDispositions(f.db, f.runId)).toEqual(["finalizer_in_progress"]);
|
||||
const [run] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
||||
expect(run.status).toBe("running");
|
||||
}, 60_000);
|
||||
});
|
||||
@@ -1188,11 +1188,18 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
|
||||
processPid: run.processPid,
|
||||
processStartedAt: run.processStartedAt?.toISOString() ?? null,
|
||||
});
|
||||
await db.transaction(async tx => {
|
||||
// An accepted result needs finalization, not a runner reattach. Its
|
||||
// execution lease must still be released: the detached executor will
|
||||
// not finalize, and the next controller's reconciler skips leased rows.
|
||||
// A finalizer lease belongs to finalization already in progress here.
|
||||
const released = await db.transaction(async tx => {
|
||||
await notifyLiveRunRecovery(tx, { companyId: run.companyId, runId: run.id });
|
||||
await tx.update(nativeRunFinalizations).set({
|
||||
leaseExpiresAt: now,
|
||||
recoveryState: "awaiting_runner_reattach",
|
||||
return tx.update(nativeRunFinalizations).set({
|
||||
leaseExpiresAt: sql`case when ${nativeRunFinalizations.resultId} is not null
|
||||
and ${nativeRunFinalizations.leaseOwner} like 'native-finalizer:%'
|
||||
then ${nativeRunFinalizations.leaseExpiresAt} else ${now.toISOString()}::timestamptz end`,
|
||||
recoveryState: sql`case when ${nativeRunFinalizations.resultId} is null
|
||||
then 'awaiting_runner_reattach' else ${nativeRunFinalizations.recoveryState} end`,
|
||||
recoveryHistory: sql`(
|
||||
select coalesce(jsonb_agg(item order by ordinal), '[]'::jsonb)
|
||||
from jsonb_array_elements(
|
||||
@@ -1214,20 +1221,26 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
|
||||
eq(nativeRunFinalizations.runId, run.id),
|
||||
owner?.bootId ? eq(nativeRunFinalizations.controllerBootId, owner.bootId)
|
||||
: isNull(nativeRunFinalizations.controllerBootId),
|
||||
isNull(nativeRunFinalizations.resultId),
|
||||
),
|
||||
);
|
||||
)
|
||||
.returning({ resultId: nativeRunFinalizations.resultId, leaseOwner: nativeRunFinalizations.leaseOwner });
|
||||
});
|
||||
const resultAccepted = Boolean(released[0]?.resultId);
|
||||
const finalizerHeld = resultAccepted && Boolean(released[0]?.leaseOwner?.startsWith("native-finalizer:"));
|
||||
await appendRunEvent(run, {
|
||||
eventType: "native.recovery.transition",
|
||||
stream: "system",
|
||||
level: "info",
|
||||
message:
|
||||
"Server shutdown suspended native controller ownership without cancelling provider work",
|
||||
message: finalizerHeld
|
||||
? "Server shutdown kept the in-progress finalizer lease of the accepted native result"
|
||||
: resultAccepted
|
||||
? "Server shutdown released the accepted native result for finalization by the next controller"
|
||||
: "Server shutdown suspended native controller ownership without cancelling provider work",
|
||||
payload: {
|
||||
restartKind: "graceful",
|
||||
signal,
|
||||
runnerDisposition: "awaiting_runner_reattach",
|
||||
runnerDisposition: finalizerHeld ? "finalizer_in_progress"
|
||||
: resultAccepted ? "awaiting_result_finalization" : "awaiting_runner_reattach",
|
||||
processPid: run.processPid,
|
||||
processGroupId: run.processGroupId,
|
||||
processStartedAt: run.processStartedAt?.toISOString() ?? null,
|
||||
|
||||
Reference in new issue
Block a user