mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
fix(heartbeat): write spawn identity and event sequence in one run update (#19954)
## 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 = <issue> 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 "<the four tests that call persistHeartbeatRunProcessMetadata>"`: 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 <noreply@anthropic.com>
This commit is contained in:
DottaandClaude Opus 5.5 authored and GitHub committed
1 parent
779913333a
commit
7213f64962
4 files changed
+169
-23
No files matched your search
@@ -36,8 +36,19 @@ export interface AppendHeartbeatRunEventInput {
|
||||
protocolSchemaVersion: number;
|
||||
canonicalPayload: Record<string, unknown>;
|
||||
};
|
||||
/**
|
||||
* 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<Pick<typeof heartbeatRuns.$inferInsert,
|
||||
"legacyProcessSettlement" | "processPid" | "processGroupId" | "processHostHash" | "processStartedAt">>;
|
||||
|
||||
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<number> {
|
||||
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,
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<typeof createDb>;
|
||||
const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
|
||||
|
||||
describe.skipIf(!support.supported)("spawn process identity persistence", () => {
|
||||
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
let other: ReturnType<typeof createDb>;
|
||||
|
||||
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<void>((resolve) => { release = resolve; });
|
||||
let locked!: () => void;
|
||||
const holding = new Promise<void>((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");
|
||||
});
|
||||
});
|
||||
@@ -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<T>(write: () => Promise<T>, attempts = 4): Promise<T> {
|
||||
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) {
|
||||
|
||||
Reference in new issue
Block a user