perf(run-events): stop blocking checkpoint FK checks during event appends (#20670)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - A native run sends a stream of runner events. The server appends
each event to the run log. The runner controller then saves a checkpoint
for recovery.
> - Each checkpoint save inserts the new `event:N` row and updates the
`cursor` row in `native_runner_checkpoints`. The new row has a foreign
key to the run row in `heartbeat_runs`.
> - The event append locks the same run row `FOR UPDATE`. That lock
blocks the `FOR KEY SHARE` lock that the foreign key check takes. So
each checkpoint save waits for the event append that is in progress.
> - On a hosted database, the two-row checkpoint upsert had the largest
total SQL time. The one-row (cursor-only) upsert, which has no foreign
key check, was much cheaper.
> - This pull request changes the append lock to `FOR NO KEY UPDATE`.
Appenders still serialize. Child foreign key checks no longer wait.
> - The benefit is shorter checkpoint saves, shorter waits before each
runner ACK, and less lock wait on the database.

## Linked Issues or Issue Description

**What existing behavior does this improve?**
The controller checkpoint save for native runs
(`nativeRunnerCheckpointWriter`), and the native run-event append
(`appendHeartbeatRunEvent`).

**Subsystem affected**
Server run-event log and native runtime checkpoint persistence.

**Current behavior**
On one hosted database, Neon query performance showed this statement as
the largest total: `insert into native_runner_checkpoints ... values
(...), (...) on conflict (run_id, part) do update`. It ran 6,582 times
at 62.6 ms (412 s) in the first 38 minutes after a compute restart, and
9,435 times at 31.8 ms (300 s) about two hours later. An earlier day
showed 100,650 calls at 102 ms. The one-row form of the same upsert ran
965 times at 13.4 ms.
The runner already writes only changed records (`captureCheckpoint`
compares plaintext), so unchanged material is not rewritten. The two-row
form is a new event row plus the cursor. Only the new row fires the
foreign key check. `appendHeartbeatRunEvent` locks the run row `FOR
UPDATE` for each native event, in a transaction with several statements.
`FOR UPDATE` conflicts with `FOR KEY SHARE`, so the check waits for that
transaction.

**Proposed behavior**
`appendHeartbeatRunEvent` locks the run row `FOR NO KEY UPDATE`. This
mode still conflicts with other appenders, `FOR SHARE`, `FOR UPDATE`,
deletes and updates. It does not conflict with `FOR KEY SHARE`. The
append never changes key columns: `runUpdate` is limited to process
fields, and the native and retry-exhaustion paths reject it. The fast
path already uses a plain `UPDATE`, which takes the same lock mode. The
execution-admission adapter already locks this row `FOR NO KEY UPDATE`.

**Reason and benefit**
The wait goes in both directions. A checkpoint save waits for an append.
An append also waits for an open checkpoint transaction that holds `FOR
KEY SHARE`. Both waits stop. The controller sends each ACK only after
its checkpoint commits, so shorter saves also shorten the event loop.

## What Changed

- `server/src/services/heartbeat-run-events.ts`: lock the run row `FOR
NO KEY UPDATE` instead of `FOR UPDATE` in the transactional append path.
Add a comment that names the contract.
- `heartbeat-run-event-atomic-append.test.ts`: add a regression test. An
append transaction stays open. A checkpoint row insert for the same run
must finish under a 2 s `lock_timeout`. A second appender must still
wait. After release, the sequence is 1, 2 with no gap and no duplicate.
The existing test with 40 concurrent appenders on two connections still
proves contiguous sequence numbers. Update the existing lock-statement
assertion to the new clause.
- Add `scripts/benchmark-run-event-checkpoint-lock.ts`. It replays a
native event stream with the real append and the real checkpoint writer
on separate connections. It times each checkpoint save for a baseline
append module and the current one.

## Verification

- Red: with the old `FOR UPDATE` clause, the new test fails with
`canceling statement due to lock timeout`. With the change it passes.
- `npx vitest run
src/__tests__/heartbeat-run-event-atomic-append.test.ts
src/__tests__/heartbeat-run-event-sequencing.test.ts
src/services/native-runtime/native-runner-checkpoint-authority.integration.test.ts
src/__tests__/native-recovery-notice-work.test.ts` (server): 4 files, 42
tests passed.
- `pnpm --filter @paperclipai/server typecheck`: passed.
- `pnpm verify` exits 2, because a new script under `scripts/` is a
global CI input. Required CI supplies the broad coverage.

Local benchmark. Embedded PostgreSQL 18.1 on a laptop. 400 events per
pass, 2 KiB event payload, 6 KiB header, real encryption. Order:
warm-up, baseline, current, baseline, current. `HOLD_MS` keeps each
append transaction open longer. It stands in for hosted round trips.

| Hold | Pass | Median save | p90 | Max | Saves > 2 ms | Total save time
|
| --- | --- | --- | --- | --- | --- | --- |
| 5 ms | Baseline (1) | 8.16 ms | 9.50 ms | 129.48 ms | 396 / 400 |
3,551 ms |
| 5 ms | Current (2) | 1.44 ms | 2.08 ms | 12.66 ms | 44 / 400 | 677 ms
|
| 5 ms | Baseline (3) | 8.12 ms | 9.64 ms | 22.08 ms | 398 / 400 | 3,453
ms |
| 5 ms | Current (4) | 1.50 ms | 2.37 ms | 10.54 ms | 80 / 400 | 700 ms
|
| 0 ms | Baseline (1) | 1.71 ms | 2.50 ms | 11.05 ms | 105 / 400 | 780
ms |
| 0 ms | Current (2) | 0.97 ms | 1.42 ms | 10.28 ms | 22 / 400 | 458 ms
|
| 0 ms | Baseline (3) | 1.49 ms | 2.02 ms | 6.45 ms | 41 / 400 | 667 ms
|
| 0 ms | Current (4) | 0.91 ms | 1.24 ms | 9.80 ms | 19 / 400 | 420 ms |

- Event order: the controller frame chain awaits each event's append and
its local commit, but not its checkpoint save. The ACK goes through
`#sendJson`, which queues the save and does not block the next frame.
Only semantic inputs and command results await the save. So the save for
event N overlaps the append of event N+1. Streaming events
(`item.delta`) also go through the same locked append, with no
checkpoint save, so a save can also overlap them. The benchmark uses
this order.
- Control without overlap (`SERIAL=1`, each save finishes before the
next append, 5 ms hold): baseline median 1.15 / 1.07 ms, current 1.03 /
1.00 ms. There is no material difference. The gain comes only from the
removed lock wait.
- Results are the same in every pass. The script asserts that all event
rows, the header and the cursor are stored, and that the run-log
sequence is contiguous.
- Command: see the script header. It needs a baseline copy of
`heartbeat-run-events.ts` from the base revision.

## Risks

- Low risk. The SQL change is one lock clause. Appenders still serialize
with each other, with `FOR UPDATE` and `FOR SHARE` lockers, and with
updates and deletes of the run row. Run deletion and key changes take
`FOR UPDATE`, so they still wait for an append, and an append still
waits for them.
- The only new concurrency is with `FOR KEY SHARE` lockers. In this code
base those are foreign key checks of child inserts. The one explicit
`FOR KEY SHARE` read is on `issues`, not on `heartbeat_runs`. A foreign
key check only needs the run row and its key to exist. The append does
not delete the row or change its key.
- The hosted cause is inferred. Neon shows calls and mean time only. We
did not read lock waits on the hosted database. The local proof shows
the mechanism and the size of the effect. A new query-performance window
after deployment must confirm the result.
- Persisted run state: no saved run, session, event or checkpoint shape
changes. State saved by the previous release carries forward unchanged.
The lock mode applies only inside one transaction, so a deploy during a
run needs no special case.

## Model Used

- Claude Opus 5.5 (Anthropic), model ID `claude-opus-5-5`, used through
Claude Code with tool use (shell, file edits, local PostgreSQL).

## 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>

scripts/benchmark-run-event-checkpoint-lock.ts

@@ -0,0 +1,123 @@
+/**
+ * Local proof for the run-event append lock and controller checkpoint saves.
+ *
+ * A native run appends every PRP event through `appendHeartbeatRunEvent`,
+ * which locks the run row. The controller then saves its checkpoint: a new
+ * `event:N` row plus the `cursor` row. The new row's foreign key check takes
+ * FOR KEY SHARE on the same run row. This script replays that pattern with the
+ * real append and the real checkpoint writer, on separate connections, and
+ * times each checkpoint save. It compares a baseline append module (FOR UPDATE)
+ * with the current one (FOR NO KEY UPDATE), in alternating order.
+ *
+ * Overlap is the normal controller order. The frame chain awaits each event's
+ * append (onCommittedEvent) and local commit, but not its checkpoint save: the
+ * ACK goes through #sendJson, which queues the save and does not block the
+ * next frame. Only semantic inputs and command results await the save. So the
+ * save for event N runs while event N+1 is appended. SERIAL=1 awaits every
+ * save before the next append, as a control without overlap.
+ *
+ * HOLD_MS keeps each append transaction open that much longer. It stands in
+ * for hosted round trips, which make the real append transaction last longer
+ * than on loopback.
+ *
+ *   git show <base>:server/src/services/heartbeat-run-events.ts \
+ *     > server/src/services/heartbeat-run-events.baseline.ts
+ *   pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-run-event-checkpoint-lock.ts
+ *
+ * Optional: EVENTS (default 400 per pass), PASSES (4), HOLD_MS (5), EVENT_KIB (2), SERIAL (0).
+ * Delete the baseline copy afterwards.
+ */
+import assert from "node:assert/strict";
+import { randomBytes, randomUUID } from "node:crypto";
+import { existsSync } from "node:fs";
+import { createRequire } from "node:module";
+import { fileURLToPath } from "node:url";
+import { agents, companies, createDb, heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations,
+  nativeRunnerCheckpoints, startEmbeddedPostgresTestDatabase } from "../packages/db/src/index.js";
+import * as candidate from "../server/src/services/heartbeat-run-events.js";
+import { nativeRunnerCheckpointWriter } from "../server/src/services/native-runtime/native-runner-checkpoint.js";
+
+const { eq, sql: drizzleSql } = createRequire(new URL("../server/package.json", import.meta.url))("drizzle-orm") as typeof import("drizzle-orm");
+
+const baselinePath = fileURLToPath(new URL("../server/src/services/heartbeat-run-events.baseline.ts", import.meta.url));
+assert(existsSync(baselinePath), "Write the baseline module first (see the header)");
+const baseline = await import(baselinePath) as typeof candidate;
+const size = (name: string, fallback: number) => Number(process.env[name] ?? fallback);
+const EVENTS = size("EVENTS", 400), PASSES = size("PASSES", 4), HOLD_MS = size("HOLD_MS", 5), EVENT_KIB = size("EVENT_KIB", 2);
+const SERIAL = process.env.SERIAL === "1";
+process.env.PAPERCLIP_SECRETS_MASTER_KEY ??= randomBytes(32).toString("base64");
+
+const keepAlive = setInterval(() => {}, 1_000);
+const temporary = await startEmbeddedPostgresTestDatabase("run-event-checkpoint-lock-");
+const appendDb = createDb(temporary.connectionString);
+const checkpointDb = createDb(temporary.connectionString);
+try {
+  const [company] = await appendDb.insert(companies).values({ name: "Lock", issuePrefix: "LCK" }).returning();
+  const [agent] = await appendDb.insert(agents).values({ companyId: company!.id, name: "Native" }).returning();
+  const [issue] = await appendDb.insert(issues).values({ companyId: company!.id, title: "Streaming task" }).returning();
+  const quantile = (values: number[], q: number) => [...values].sort((a, b) => a - b)[Math.min(values.length - 1, Math.floor(values.length * q))]!;
+  const round = (value: number) => Number(value.toFixed(2));
+
+  async function pass(label: "baseline" | "candidate") {
+    const append = (label === "baseline" ? baseline : candidate).appendHeartbeatRunEvent;
+    const runId = randomUUID(), bootId = randomUUID(), leaseOwner = randomUUID();
+    await appendDb.insert(heartbeatRuns).values({ id: runId, companyId: company!.id, agentId: agent!.id, status: "running",
+      runtimeMode: "native", nativeIssueId: issue!.id, issueId: issue!.id,
+      runnerProfileJson: { padding: randomBytes(3 * 1024).toString("base64") } });
+    await appendDb.insert(nativeRunFinalizations).values({ runId, companyId: company!.id, issueId: issue!.id, phase: "observed",
+      controllerBootId: bootId, controllerGeneration: 1, leaseOwner, leaseExpiresAt: new Date(Date.now() + 3_600_000) });
+    const write = nativeRunnerCheckpointWriter(checkpointDb, { companyId: company!.id, issueId: issue!.id, runId,
+      authorityRunId: runId, bootId, generation: 1, leaseOwner });
+    const checkpoint = (records: Map<string, string>, changed: Map<string, string>, reset: boolean) => ({
+      runId, records, changedRecords: changed, removedRecords: [], reset, workCommandIds: [], predecessor: null,
+      runnerVersion: "bench", runnerDigest: `sha256:${"a".repeat(64)}`,
+      recoveryEvidence: { runnerInstanceId: "bench", normalizedSessionId: "bench", authenticatedConnectionCount: 1, highestRecordedSourceSeq: 0 },
+      restoreSessionDirectory() {},
+    }) as unknown as Parameters<typeof write>[0];
+    const records = new Map([["header", JSON.stringify({ padding: randomBytes(6 * 1024).toString("base64") })],
+      ["cursor", JSON.stringify({ ackedSourceSeq: 0 })]]);
+    await write(checkpoint(records, new Map(records), true));
+    const saves: number[] = [];
+    let tail = Promise.resolve();
+    const started = performance.now();
+    for (let seq = 1; seq <= EVENTS; seq++) {
+      const event = { sourceSeq: seq, eventType: "item.completed", text: randomBytes(EVENT_KIB * 768).toString("base64") };
+      // The append commits before the controller ACKs, like onCommittedEvent.
+      await appendDb.transaction(async tx => {
+        await append(tx as unknown as typeof appendDb, { companyId: company!.id, agentId: agent!.id, runId,
+          eventType: "item.completed", stream: "system", level: "info", payload: { prpEvent: event },
+          nativeSource: { sourceInstanceId: "bench-runner", sourceEventId: `event-${seq}`, sourceSeq: seq,
+            protocolSchemaVersion: 1, canonicalPayload: event } });
+        if (HOLD_MS) await tx.execute(drizzleSql`select pg_sleep(${HOLD_MS / 1000})`);
+      });
+      const changed = new Map([[`event:${seq}`, JSON.stringify(event)], ["cursor", JSON.stringify({ ackedSourceSeq: seq })]]);
+      for (const [key, value] of changed) records.set(key, value);
+      // Checkpoint saves are serialized per controller and overlap the next append.
+      tail = tail.then(async () => {
+        const start = performance.now();
+        await write(checkpoint(new Map(records), changed, false));
+        saves.push(performance.now() - start);
+      });
+      if (SERIAL) await tail;
+    }
+    await tail;
+    const elapsed = performance.now() - started;
+    const [stored] = await checkpointDb.select({ count: drizzleSql<number>`count(*)::int` }).from(nativeRunnerCheckpoints)
+      .where(eq(nativeRunnerCheckpoints.runId, runId));
+    const events = await appendDb.select({ seq: heartbeatRunEvents.seq }).from(heartbeatRunEvents)
+      .where(eq(heartbeatRunEvents.runId, runId)).orderBy(heartbeatRunEvents.seq);
+    assert.equal(stored!.count, EVENTS + 2, "Every event row, the header and the cursor are stored");
+    assert.deepEqual(events.map(row => row.seq), Array.from({ length: EVENTS }, (_, index) => index + 1));
+    return { label, saves: saves.length, saveMedianMs: round(quantile(saves, 0.5)), saveP90Ms: round(quantile(saves, 0.9)),
+      saveMaxMs: round(Math.max(...saves)), savesOver2Ms: saves.filter(value => value > 2).length,
+      saveTotalMs: round(saves.reduce((sum, value) => sum + value, 0)), elapsedMs: round(elapsed) };
+  }
+
+  await pass("candidate"); // Warm caches and the prepared-statement path once.
+  const results = [];
+  for (let index = 0; index < PASSES; index++) results.push(await pass(index % 2 === 0 ? "baseline" : "candidate"));
+  console.log(JSON.stringify({ events: EVENTS, serial: SERIAL, holdMs: HOLD_MS, eventKiB: EVENT_KIB, results }, null, 2));
+} finally {
+  clearInterval(keepAlive);
+  await temporary.cleanup();
+}

server/src/__tests__/heartbeat-run-event-atomic-append.test.ts

@@ -3,7 +3,7 @@ import { randomBytes } from "node:crypto";
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
 import { eq, inArray, sql } from "drizzle-orm";
 import { drizzle } from "drizzle-orm/postgres-js";
-import { agents, companies, createDb, heartbeatRunEvents, heartbeatRuns } from "@paperclipai/db";
+import { agents, companies, createDb, heartbeatRunEvents, heartbeatRuns, issues, nativeRunnerCheckpoints } from "@paperclipai/db";
 import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js";
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
 
@@ -51,6 +51,7 @@ describeEmbeddedPostgres("atomic ordinary run events", () => {
         const fixtureCompanies = [companyId, otherCompanyId];
         await db.delete(heartbeatRunEvents).where(inArray(heartbeatRunEvents.companyId, fixtureCompanies));
         await db.delete(heartbeatRuns).where(inArray(heartbeatRuns.companyId, fixtureCompanies));
+        await db.delete(issues).where(inArray(issues.companyId, fixtureCompanies));
         await db.delete(agents).where(inArray(agents.companyId, fixtureCompanies));
         await db.delete(companies).where(inArray(companies.id, fixtureCompanies));
       }
@@ -145,7 +146,7 @@ describeEmbeddedPostgres("atomic ordinary run events", () => {
     expect((await readRun(runId)).nextEventSeq).toBe(2);
     expect(await readEvents(runId)).toEqual([committed.row]);
     const lockReads = statements.filter(query => query.startsWith("select ") &&
-      query.includes('from "heartbeat_runs"') && query.endsWith("for update"));
+      query.includes('from "heartbeat_runs"') && query.endsWith("for no key update"));
     expect(lockReads).toHaveLength(2);
     for (const query of lockReads) {
       expect(query).toMatch(/^select "company_id", "agent_id" from "heartbeat_runs" /);
@@ -153,6 +154,53 @@ describeEmbeddedPostgres("atomic ordinary run events", () => {
     }
   });
 
+  it("serializes appenders without blocking child foreign key checks on the run", async () => {
+    const runId = await makeRun();
+    const [issue] = await db.insert(issues).values({ companyId, title: "Checkpointed task" }).returning();
+    await db.update(heartbeatRuns).set({ nativeIssueId: issue!.id, runtimeMode: "native" })
+      .where(eq(heartbeatRuns.id, runId));
+    const native = (ordinal: number) => ({
+      companyId, agentId, runId, eventType: "item.completed",
+      nativeSource: { sourceInstanceId: "native-writer", sourceEventId: `held-${ordinal}`,
+        sourceSeq: ordinal, protocolSchemaVersion: 1, canonicalPayload: { ordinal } },
+    });
+    let releaseHolder!: () => void;
+    const held = new Promise<void>(resolve => { releaseHolder = resolve; });
+    let holderLocked!: () => void;
+    const locked = new Promise<void>(resolve => { holderLocked = resolve; });
+    // An append whose transaction is still open holds the run lock, as an
+    // event append does while its remaining statements run.
+    const holder = otherDb.transaction(async tx => {
+      await appendHeartbeatRunEvent(tx as unknown as typeof db, native(1));
+      holderLocked();
+      await held;
+    });
+    await locked;
+    let secondSettled = false;
+    const second = appendHeartbeatRunEvent(db, native(2)).finally(() => { secondSettled = true; });
+    try {
+      // A controller checkpoint row references the run row by foreign key.
+      // Its insert must not wait for the open append.
+      await db.transaction(async tx => {
+        await tx.execute(sql`set local lock_timeout = '2s'`);
+        await tx.insert(nativeRunnerCheckpoints).values({ companyId, issueId: issue!.id, runId,
+          part: "event:1", material: { fixture: true } });
+      });
+      // Another appender still waits for the open append.
+      await new Promise(resolve => setTimeout(resolve, 200));
+      expect(secondSettled).toBe(false);
+    } finally {
+      releaseHolder();
+      await holder;
+    }
+    expect((await second).row.seq).toBe(2);
+    expect((await readEvents(runId)).map(row => [row.seq, row.sourceSeq])).toEqual([[1, 1], [2, 2]]);
+    expect((await readRun(runId)).nextEventSeq).toBe(3);
+    expect(await db.select({ part: nativeRunnerCheckpoints.part }).from(nativeRunnerCheckpoints)
+      .where(eq(nativeRunnerCheckpoints.runId, runId))).toEqual([{ part: "event:1" }]);
+    await db.delete(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, runId));
+  });
+
   it("rejects mismatched authority without changing either run", async () => {
     const runId = await makeRun();
     const otherRunId = "30000000-0000-4000-8000-000000000099";

server/src/services/heartbeat-run-events.ts

@@ -134,7 +134,11 @@ export async function appendHeartbeatRunEvent(
         ...(projectsProgress ? { issueId: boundRunIssueIdSql() } : {}) })
       .from(heartbeatRuns)
       .where(eq(heartbeatRuns.id, input.runId))
-      .for("update")
+      // Appenders, Stop, recovery and deletion still serialize on this row.
+      // The append never changes key columns, so it must not block foreign
+      // key checks (FOR KEY SHARE) of child rows, such as controller
+      // checkpoints, that are written for the same run while events stream.
+      .for("no key update")
       .limit(1)
       .then((rows) => rows[0] ?? null);
     if (!run || run.companyId !== input.companyId || run.agentId !== input.agentId) {