From 7213f649621e5f232da0bb2ba3bf79488f0617b1 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sat, 7 Nov 2026 21:39:43 -0600 Subject: [PATCH] fix(heartbeat): write spawn identity and event sequence in one run update (#19954) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - The heartbeat records the provider process identity when a native or legacy run spawns its process > - This write updated the run row and then reserved the event sequence with a second UPDATE of the same row > - PostgreSQL re-checks every foreign key on a row that the current transaction already changed, so the second UPDATE took KEY SHARE on the issue, the predecessor run, and the other parent rows > - A transaction that held one of those rows FOR UPDATE blocked spawn persistence, and the failed write ended a user's Retry as `native_session_interrupted` > - This pull request writes the identity and reserves the sequence in one UPDATE, and it retries deadlock and serialization failures of the spawn write > - The benefit is that a user's Retry no longer fails because an unrelated transaction holds the issue or the failed predecessor run ## Linked Issues or Issue Description **What happened** A user clicked Retry on a failed native run in the `runs-adapters.inspect-failed-run` chat eval (embedded PostgreSQL, one server process). The retry run failed with `native_session_interrupted`. Its error was `Failed query: update "heartbeat_runs" set "next_event_seq" = "heartbeat_runs"."next_event_seq" + 1 ...`. The query started at 01:51:04 and failed at 01:51:16. The OpenCode driver awaits `onSpawn` before its health check, so `native.session.execute` failed after 12002 ms. The stack showed `allocateHeartbeatRunEventSeq` inside `appendHeartbeatRunEvent`, called from `persistHeartbeatRunProcessMetadata`. The process identity was lost (`processPid` stayed null). The server log does not include the PostgreSQL cause code. **Expected behavior** Spawn persistence must not wait on locks for unrelated parent rows. A transient deadlock or serialization failure must be retried. It must not end the run. **Steps to reproduce** 1. Create a run whose `issue_id` and `retry_of_run_id` point at an issue and a predecessor run. 2. In a second connection, begin a transaction and run `SELECT ... FROM issues WHERE id = FOR UPDATE` (or lock the predecessor run). Keep the transaction open. 3. Call `persistHeartbeatRunProcessMetadata` for the run. 4. Before this change, the call blocks on the `next_event_seq` UPDATE until the second transaction ends. If that transaction then waits on the run row, PostgreSQL reports a deadlock. `server/src/services/heartbeat/process-identity-persistence.test.ts` reproduces step 4. Root cause: `server/src/services/heartbeat/run-lifecycle.ts` (`persistHeartbeatRunProcessMetadata`) updated the run row, then `appendHeartbeatRunEvent` called `allocateHeartbeatRunEventSeq` (`server/src/services/heartbeat-run-events.ts`), which updated the same row again. PostgreSQL's `RI_FKey_fk_upd_check_required` always fires the foreign-key checks when the old row version was written by the current transaction. The checks take KEY SHARE on each referenced row, and KEY SHARE conflicts with FOR UPDATE. ## What Changed - `appendHeartbeatRunEvent` accepts a narrow `runUpdate` option (process identity columns only). `allocateHeartbeatRunEventSeq` applies these columns in the same UPDATE that reserves the sequence. Deduplicated receipts (native source, retry exhaustion) reject `runUpdate`, because they do not allocate. - `persistHeartbeatRunProcessMetadata` reads the run binding, appends `native.process_identity_recorded` with the identity as `runUpdate`, and returns the updated row. The run row now gets one UPDATE in this transaction. - New `retryTransientRunWrite` retries SQLSTATE `40P01` and `40001` up to four times. The lifecycle `persistRunProcessMetadata` and the legacy cleanup-intent wrapper in `heartbeat.ts` use it. The legacy wrapper still updates the run row a second time when it records the cleanup intent, so the retry covers a deadlock there. - New regression test `server/src/services/heartbeat/process-identity-persistence.test.ts`. ## Verification - `npx vitest run src/services/heartbeat/process-identity-persistence.test.ts` (from `server/`): 3 passed. Without the `run-lifecycle.ts` change, both lock cases fail with `expected 'blocked' not to be 'blocked'` after 3 s. - `npx vitest run src/__tests__/heartbeat-run-event-sequencing.test.ts src/services/heartbeat/run-lifecycle.test.ts`: 42 passed. - `npx vitest run src/__tests__/heartbeat-process-recovery.test.ts -t ""`: 5 passed. - `tsc --noEmit -p server`: no errors in the changed files. The only error comes from a local workspace-link mix in `packages/adapter-utils`, which this change does not touch. - I did not run `pnpm verify` or the full suite locally. CI supplies that coverage. ## Risks - Low risk. The written columns, the event type, the payload, and the sequence order do not change. - The event payload `processStartedAt` now comes from the value that the transaction writes, not from a re-read of the row. The value is the same. - An event insert still checks its own `agent_id` and `company_id` foreign keys. A transaction that holds the agent row FOR UPDATE can still delay the spawn write. That wait is the same for every event writer, and this change does not alter it. - Other paths that update a run row and then append an event in the same transaction keep the second-UPDATE foreign-key checks. This change fixes the spawn identity path that lost the run. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - Anthropic Claude Opus 5.5 (`claude-opus-5-5`), used through Claude Code with tool use (shell, file edits, embedded PostgreSQL experiments). ## 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 --- server/src/services/heartbeat-run-events.ts | 18 +++- server/src/services/heartbeat.ts | 5 +- .../process-identity-persistence.test.ts | 99 +++++++++++++++++++ .../src/services/heartbeat/run-lifecycle.ts | 70 +++++++++---- 4 files changed, 169 insertions(+), 23 deletions(-) create mode 100644 server/src/services/heartbeat/process-identity-persistence.test.ts diff --git a/server/src/services/heartbeat-run-events.ts b/server/src/services/heartbeat-run-events.ts index cbc23b0e7a7..1c3325ebd05 100644 --- a/server/src/services/heartbeat-run-events.ts +++ b/server/src/services/heartbeat-run-events.ts @@ -36,8 +36,19 @@ export interface AppendHeartbeatRunEventInput { protocolSchemaVersion: number; canonicalPayload: Record; }; + /** + * Run columns to change in the same UPDATE that reserves the sequence. A + * second UPDATE of a run row this transaction already changed makes + * PostgreSQL re-check every foreign key on heartbeat_runs (KEY SHARE on the + * agent, issue, predecessor run, ...), so it can wait behind, or deadlock + * with, an unrelated transaction that holds one of those rows FOR UPDATE. + */ + runUpdate?: HeartbeatRunEventRunUpdate; } +export type HeartbeatRunEventRunUpdate = Partial>; + export interface AppendHeartbeatRunEventResult { row: typeof heartbeatRunEvents.$inferSelect; disposition: "committed" | "duplicate"; @@ -61,10 +72,12 @@ export class HeartbeatRunEventConflictError extends Error { export async function allocateHeartbeatRunEventSeq( db: Db, runId: string, + runUpdate: HeartbeatRunEventRunUpdate = {}, ): Promise { const [updated] = await db .update(heartbeatRuns) .set({ + ...runUpdate, nextEventSeq: sql`${heartbeatRuns.nextEventSeq} + 1`, updatedAt: new Date(), }) @@ -86,7 +99,9 @@ export async function appendHeartbeatRunEvent( const projectsReassessment = readNativeEventEvidenceOutcome(input.payload?.prpEvent) !== null; const projectsProgress = isSafeNativeChatProgressEventType(input.eventType); const projectsResponse = isNativeResponseEvidence(input.eventType, input.payload?.prpEvent); - if (!input.nativeSource && !input.retryExhaustion && !projectsProgress && !projectsResponse && !projectsReassessment && !projectsRecovery && !projectsRetention) { + // Deduplicated receipts return without allocating, so they cannot carry run changes. + if (input.runUpdate && (input.nativeSource || input.retryExhaustion)) throw new Error("heartbeat_run_event_run_update_unsupported"); + if (!input.runUpdate && !input.nativeSource && !input.retryExhaustion && !projectsProgress && !projectsResponse && !projectsReassessment && !projectsRecovery && !projectsRetention) { const allocated = db.$with("allocated_run_event", {}).as(sql` update ${heartbeatRuns} set next_event_seq = ${heartbeatRuns.nextEventSeq} + 1, @@ -202,6 +217,7 @@ export async function appendHeartbeatRunEvent( const seq = await allocateHeartbeatRunEventSeq( tx as unknown as Db, input.runId, + input.runUpdate, ); const [row] = await tx.insert(heartbeatRunEvents).values({ companyId: input.companyId, diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 52f70ed5ac2..a9fd5f8837d 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -82,6 +82,7 @@ import { decorateHeartbeatRunRuntimeStatus, createHeartbeatLifecycle, persistHeartbeatRunProcessMetadata, + retryTransientRunWrite, } from "./heartbeat/run-lifecycle.js"; export { redactDetectedSuccessfulRunProgressSummaryForBoard, @@ -3547,13 +3548,13 @@ export function heartbeatService( } // Bind cleanup intent to this exact process generation in the // same commit as its host-observed PID/group/start identity. - return db.transaction(async transaction => { + return retryTransientRunWrite(() => db.transaction(async transaction => { const tx = transaction as unknown as Db; const persisted = await persistHeartbeatRunProcessMetadata(tx, processRunId, meta); await recordLegacyCaseHandoffCleanupIntent(tx, { runId: processRunId, issueId, controllerBootId: legacyControllerBootId, scratch: runScratch }); return persisted; - }); + })); }, recordCurrentHeartbeatRunRuntimeProgress, dispatchPendingNativeStatusWakeups, diff --git a/server/src/services/heartbeat/process-identity-persistence.test.ts b/server/src/services/heartbeat/process-identity-persistence.test.ts new file mode 100644 index 00000000000..2571acf24a3 --- /dev/null +++ b/server/src/services/heartbeat/process-identity-persistence.test.ts @@ -0,0 +1,99 @@ +import { and, eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { agents, companies, createDb, heartbeatRunEvents, heartbeatRuns, issues } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { PROCESS_IDENTITY_RECORDED } from "../native-local-process-stop.js"; +import { persistHeartbeatRunProcessMetadata, retryTransientRunWrite } from "./run-lifecycle.js"; + +const support = await getEmbeddedPostgresTestSupport(); +const companyId = "30000000-0000-4000-8000-000000000001"; +const agentId = "30000000-0000-4000-8000-000000000002"; +const issueId = "30000000-0000-4000-8000-000000000003"; +const predecessorRunId = "30000000-0000-4000-8000-000000000004"; +const runId = "30000000-0000-4000-8000-000000000005"; +type Tx = ReturnType; +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +describe.skipIf(!support.supported)("spawn process identity persistence", () => { + let temporary: Awaited>; + let db: ReturnType; + let other: ReturnType; + + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("paperclip-process-identity-"); + db = createDb(temporary.connectionString); + other = createDb(temporary.connectionString); + await db.insert(companies).values({ id: companyId, name: "Identity", issuePrefix: "PID" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Identity agent" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Retry me" }); + await db.insert(heartbeatRuns).values([ + { id: predecessorRunId, companyId, agentId, issueId, scopeKind: "issue", status: "failed" }, + { id: runId, companyId, agentId, issueId, scopeKind: "issue", status: "running", retryOfRunId: predecessorRunId }, + ]); + }); + + afterAll(async () => { + await temporary?.cleanup(); + }); + + // A user Retry spawned the provider while another transaction held a parent + // row (the issue or predecessor run) FOR UPDATE. Writing the identity + // and then reserving the event sequence updated the run row twice, and the + // second UPDATE re-checked heartbeat_runs' foreign keys (KEY SHARE on those + // parents; the event row's own agent/company keys are checked either way). It waited behind the holder and then failed, so the live run + // ended as native_session_interrupted. + it.each([ + ["issue", (tx: Tx) => tx.select({ id: issues.id }).from(issues).where(eq(issues.id, issueId))], + ["predecessor run", (tx: Tx) => tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(eq(heartbeatRuns.id, predecessorRunId))], + ] as const)("records identity without waiting on a transaction holding the %s FOR UPDATE", async (_label, lockParent) => { + let release!: () => void; + const released = new Promise((resolve) => { release = resolve; }); + let locked!: () => void; + const holding = new Promise((resolve) => { locked = resolve; }); + const holder = other.transaction(async (tx) => { + await lockParent(tx as unknown as Tx).for("update"); + locked(); + await released; + }); + await holding; + try { + const pid = 40_000 + Math.floor(Math.random() * 1000); + const persisted = persistHeartbeatRunProcessMetadata(db, runId, { + pid, processGroupId: null, startedAt: new Date().toISOString(), targetKind: "remote", + }); + const outcome = await Promise.race([ + persisted.then((run) => ({ run }), (error: unknown) => ({ error })), + sleep(3_000).then(() => "blocked" as const), + ]); + expect(outcome).not.toBe("blocked"); + expect(outcome).not.toHaveProperty("error"); + const run = (outcome as { run: typeof heartbeatRuns.$inferSelect | null }).run; + expect(run).toMatchObject({ id: runId, processPid: pid, processHostHash: "remote" }); + // The cases share one run; select this write's event by its reserved sequence. + const [event] = await db.select().from(heartbeatRunEvents).where(and( + eq(heartbeatRunEvents.runId, runId), + eq(heartbeatRunEvents.seq, Number(run!.nextEventSeq) - 1), + )); + expect(event).toMatchObject({ eventType: PROCESS_IDENTITY_RECORDED }); + expect(event?.payload).toMatchObject({ processPid: pid, targetKind: "remote" }); + } finally { + release(); + await holder; + } + }, 30_000); +}); + +describe("transient run write retry", () => { + it("retries a deadlock victim and rethrows other failures", async () => { + let calls = 0; + await expect(retryTransientRunWrite(async () => { + calls += 1; + if (calls < 3) throw Object.assign(new Error("Failed query"), { cause: { code: "40P01" } }); + return "persisted"; + })).resolves.toBe("persisted"); + expect(calls).toBe(3); + await expect(retryTransientRunWrite(async () => { + throw Object.assign(new Error("Failed query"), { cause: { code: "23505" } }); + })).rejects.toThrow("Failed query"); + }); +}); diff --git a/server/src/services/heartbeat/run-lifecycle.ts b/server/src/services/heartbeat/run-lifecycle.ts index 1a97b33057b..650f7a504a4 100644 --- a/server/src/services/heartbeat/run-lifecycle.ts +++ b/server/src/services/heartbeat/run-lifecycle.ts @@ -576,6 +576,32 @@ function buildRunEventRuntimeProgress(input: { }; } +function isTransientRunWriteConflict(error: unknown) { + let current = error; + for (let depth = 0; current && typeof current === "object" && depth < 4; depth++) { + const code = (current as { code?: unknown }).code; + if (code === "40P01" || code === "40001") return true; + current = (current as { cause?: unknown }).cause; + } + return false; +} + +/** + * Spawn identity is the run's ownership proof. A deadlock victim or + * serialization failure rolled back cleanly, so the whole write is retried + * instead of failing a live provider session. + */ +export async function retryTransientRunWrite(write: () => Promise, attempts = 4): Promise { + for (let attempt = 1; ; attempt++) { + try { + return await write(); + } catch (error) { + if (attempt >= attempts || !isTransientRunWriteConflict(error)) throw error; + await new Promise((resolve) => setTimeout(resolve, 20 * attempt)); + } + } +} + export async function persistHeartbeatRunProcessMetadata( db: Db, runId: string, @@ -586,32 +612,36 @@ export async function persistHeartbeatRunProcessMetadata( () => null, ); const startedAt = new Date(observedStartedAt ?? meta.startedAt); + const processStartedAt = Number.isNaN(startedAt.getTime()) ? new Date() : startedAt; return db.transaction(async tx => { - const run = await tx - .update(heartbeatRuns) - .set({ + const [binding] = await tx + .select({ companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .limit(1); + if (!binding) return null; + // One UPDATE writes the identity and reserves the event sequence. Updating + // the row twice here re-checks its foreign keys and can wait behind (or + // deadlock with) any transaction holding the agent, issue, or predecessor + // run FOR UPDATE, which failed spawn persistence and lost the run. + await appendHeartbeatRunEvent(tx as unknown as Db, { + companyId: binding.companyId, runId, agentId: binding.agentId, + eventType: PROCESS_IDENTITY_RECORDED, stream: "system", level: "info", + message: "Process identity recorded; prior stop evidence no longer applies.", + payload: { targetKind: meta.targetKind ?? "local", hostname: os.hostname(), + processPid: meta.pid, processGroupId: meta.processGroupId, + processStartedAt: processStartedAt.toISOString() }, + runUpdate: { legacyProcessSettlement: null, processPid: meta.pid, processGroupId: meta.processGroupId, // A sandbox PID never names a process in this PID space. processHostHash: meta.targetKind === "remote" ? "remote" : currentProcessSpaceHash(), - processStartedAt: Number.isNaN(startedAt.getTime()) - ? new Date() - : startedAt, - updatedAt: new Date(), - }) - .where(eq(heartbeatRuns.id, runId)) - .returning() - .then((rows) => rows[0] ?? null); - if (run) await appendHeartbeatRunEvent(tx as unknown as Db, { - companyId: run.companyId, runId, agentId: run.agentId, - eventType: PROCESS_IDENTITY_RECORDED, stream: "system", level: "info", - message: "Process identity recorded; prior stop evidence no longer applies.", - payload: { targetKind: meta.targetKind ?? "local", hostname: os.hostname(), - processPid: meta.pid, processGroupId: meta.processGroupId, - processStartedAt: run.processStartedAt?.toISOString() ?? null }, + processStartedAt, + }, }); - return run; + return tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)) + .then((rows) => rows[0] ?? null); }); } @@ -2245,7 +2275,7 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl runId: string, meta: { pid: number; processGroupId: number | null; startedAt: string; targetKind: "local" | "remote" }, ) { - return persistHeartbeatRunProcessMetadata(db, runId, meta); + return retryTransientRunWrite(() => persistHeartbeatRunProcessMetadata(db, runId, meta)); } async function clearDetachedRunWarning(runId: string) {