From b294abb9eb96c33048b4f53a2a037e04886b7407 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Fri, 16 Oct 2026 08:51:01 -0500 Subject: [PATCH] perf(server): append ordinary run events in one statement (#16569) Ordinary run events now validate their binding, allocate a sequence and insert the decoded receipt in one atomic PostgreSQL statement. Native replay and retry-exhaustion transactions keep their existing behavior. Verification: exact-head full CI/E2E, Superagent, legacy security and Apex 5/5 passed. PostgreSQL concurrency, rollback and payload controls pass. Controlled network-delay comparison reduced median event persistence from about192ms to48ms; no staging response-time claim is made before deployment. Co-Authored-By: Paperclip --- doc/run-log-events.md | 10 ++ .../heartbeat-run-event-atomic-append.test.ts | 145 ++++++++++++++++++ .../recovery-stale-issue-lock-sweep.test.ts | 23 ++- server/src/services/heartbeat-run-events.ts | 26 +++- .../native-session-executor.test.ts | 32 ++-- 5 files changed, 217 insertions(+), 19 deletions(-) create mode 100644 server/src/__tests__/heartbeat-run-event-atomic-append.test.ts diff --git a/doc/run-log-events.md b/doc/run-log-events.md index 1c9e8658f57..40fb9dcb733 100644 --- a/doc/run-log-events.md +++ b/doc/run-log-events.md @@ -5,6 +5,16 @@ Run-log events write to the `heartbeat_run_events` table Paperclip Telemetry events, and they are not OpenTelemetry exports. A run-log event needs no operator endpoint. +## Ordinary event persistence + +Events without a native source identity or retry-exhaustion receipt use one +atomic PostgreSQL statement. The statement checks the stored run's company and +agent, allocates its next sequence, and inserts the event. An insertion failure +also rolls back the sequence allocation. The returned row uses the existing +column decoders, including the lossless payload codec. Callers still wait for +the durable write before they publish or acknowledge the event. Native-source +replay and retry-exhaustion deduplication retain their existing transaction path. + ## Provider-pack preparation timing The existing `run.performance.span` event also records `provider_pack.select`, diff --git a/server/src/__tests__/heartbeat-run-event-atomic-append.test.ts b/server/src/__tests__/heartbeat-run-event-atomic-append.test.ts new file mode 100644 index 00000000000..ad4a40ce410 --- /dev/null +++ b/server/src/__tests__/heartbeat-run-event-atomic-append.test.ts @@ -0,0 +1,145 @@ +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { eq, sql } from "drizzle-orm"; +import { drizzle } from "drizzle-orm/postgres-js"; +import { agents, companies, createDb, heartbeatRunEvents, heartbeatRuns } from "@paperclipai/db"; +import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; + +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; + +describeEmbeddedPostgres("atomic ordinary run events", () => { + let temporary: Awaited>; + let db: ReturnType; + let otherDb: ReturnType; + const companyId = "30000000-0000-4000-8000-000000000001"; + const agentId = "30000000-0000-4000-8000-000000000002"; + const otherCompanyId = "30000000-0000-4000-8000-000000000003"; + const otherAgentId = "30000000-0000-4000-8000-000000000004"; + let ordinal = 10; + + const makeRun = async () => { + const id = `30000000-0000-4000-8000-${String(ordinal++).padStart(12, "0")}`; + await db.insert(heartbeatRuns).values({ id, companyId, agentId, status: "running" }); + return id; + }; + const readRun = async (id: string) => (await db.select().from(heartbeatRuns) + .where(eq(heartbeatRuns.id, id)))[0]!; + const readEvents = (id: string) => db.select().from(heartbeatRunEvents) + .where(eq(heartbeatRunEvents.runId, id)).orderBy(heartbeatRunEvents.seq); + + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("paperclip-atomic-events-"); + db = createDb(temporary.connectionString); + otherDb = createDb(temporary.connectionString); + await db.insert(companies).values([ + { id: companyId, name: "Atomic events", issuePrefix: "ATM" }, + { id: otherCompanyId, name: "Other events", issuePrefix: "OTH" }, + ]); + await db.insert(agents).values([ + { id: agentId, companyId, name: "Event agent" }, + { id: otherAgentId, companyId: otherCompanyId, name: "Other agent" }, + ]); + }, 60_000); + afterAll(async () => { await temporary?.cleanup(); }); + + it("persists a decoded receipt in one executed statement", async () => { + const runId = await makeRun(); + const statements: string[] = []; + const recorded = drizzle(db.$client, { + schema: db._.fullSchema, + logger: { logQuery: query => statements.push(query) }, + }) as typeof db; + const payload = { nested: ["before\u0000after"], $paperclipRunEventJsonV1: "literal" }; + const receipt = await appendHeartbeatRunEvent(recorded, { + companyId, agentId, runId, eventType: "run.performance.span", stream: "system", + level: "info", color: "green", message: "runner span", payload, + }); + expect(receipt.disposition).toBe("committed"); + expect(receipt.highestContiguousSourceSeq).toBe(0); + expect(receipt.row).toEqual(expect.objectContaining({ + companyId, agentId, runId, seq: 1, eventType: "run.performance.span", stream: "system", + level: "info", color: "green", message: "runner span", payload, + sourceInstanceId: null, sourceEventId: null, sourceSeq: null, + sourcePayloadSha256: null, protocolSchemaVersion: null, + })); + expect(receipt.row.id).toEqual(expect.any(Number)); + expect(receipt.row.createdAt).toBeInstanceOf(Date); + expect(await readEvents(runId)).toEqual([receipt.row]); + expect(statements).toHaveLength(1); + }); + + it("shares the run lock and sequence with native writers on separate connections", async () => { + const runId = await makeRun(); + const receipts = await Promise.all(Array.from({ length: 40 }, (_, index) => + appendHeartbeatRunEvent(index % 2 ? db : otherDb, { + companyId, agentId, runId, eventType: index % 2 ? "run.performance.span" : "item.completed", + message: `event-${index}`, + ...(index % 2 ? {} : { nativeSource: { + sourceInstanceId: "native-writer", sourceEventId: `event-${index}`, + sourceSeq: index / 2 + 1, protocolSchemaVersion: 1, + canonicalPayload: { ordinal: index }, + } }), + }))); + expect(new Set(receipts.map(receipt => receipt.row.id)).size).toBe(40); + expect((await readEvents(runId)).map(row => row.seq)).toEqual( + Array.from({ length: 40 }, (_, index) => index + 1), + ); + expect((await readRun(runId)).nextEventSeq).toBe(41); + const replay = await appendHeartbeatRunEvent(db, { + companyId, agentId, runId, eventType: "item.completed", message: "event-0", + nativeSource: { sourceInstanceId: "native-writer", sourceEventId: "event-0", + sourceSeq: 1, protocolSchemaVersion: 1, canonicalPayload: { ordinal: 0 } }, + }); + expect(replay.disposition).toBe("duplicate"); + expect(replay.highestContiguousSourceSeq).toBe(20); + expect((await readRun(runId)).nextEventSeq).toBe(41); + }); + + it("rejects mismatched authority without changing either run", async () => { + const runId = await makeRun(); + const otherRunId = "30000000-0000-4000-8000-000000000099"; + await db.insert(heartbeatRuns).values({ id: otherRunId, companyId: otherCompanyId, + agentId: otherAgentId, status: "running" }); + const before = [await readRun(runId), await readRun(otherRunId)]; + for (const authority of [ + { companyId: otherCompanyId, agentId, runId }, + { companyId, agentId: otherAgentId, runId }, + { companyId, agentId, runId: otherRunId }, + { companyId, agentId, runId: "30000000-0000-4000-8000-000000000098" }, + ]) { + await expect(appendHeartbeatRunEvent(db, { ...authority, eventType: "stdout" })) + .rejects.toThrow("heartbeat_run_event_binding_mismatch"); + } + expect([await readRun(runId), await readRun(otherRunId)]).toEqual(before); + expect(await readEvents(runId)).toEqual([]); + expect(await readEvents(otherRunId)).toEqual([]); + }); + + it("rolls back allocation on insertion failure and in an outer transaction", async () => { + const runId = await makeRun(); + const before = await readRun(runId); + await db.execute(sql`alter table heartbeat_run_events add constraint atomic_event_failure + check (message is distinct from 'force-event-failure') not valid`); + try { + await expect(appendHeartbeatRunEvent(db, { + companyId, agentId, runId, eventType: "stdout", message: "force-event-failure", + })).rejects.toThrow(); + expect(await readRun(runId)).toEqual(before); + expect(await readEvents(runId)).toEqual([]); + } finally { + await db.execute(sql`alter table heartbeat_run_events drop constraint atomic_event_failure`); + } + await expect(db.transaction(async tx => { + await appendHeartbeatRunEvent(tx as unknown as typeof db, { + companyId, agentId, runId, eventType: "stdout", + }); + throw new Error("outer rollback"); + })).rejects.toThrow("outer rollback"); + expect(await readRun(runId)).toEqual(before); + expect(await readEvents(runId)).toEqual([]); + const receipt = await appendHeartbeatRunEvent(db, { companyId, agentId, runId, eventType: "stdout" }); + expect(receipt.row.seq).toBe(1); + expect(receipt.row.payload).toBeNull(); + }); +}); diff --git a/server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts b/server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts index 04f6b69165f..a6168189b0b 100644 --- a/server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts +++ b/server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts @@ -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, vi } from "vitest"; import { costEvents, @@ -1049,27 +1049,34 @@ describeEmbeddedPostgres("recovery sweepStaleIssueLocks", () => { // Make only the audit-event insert fail. The run update commits the // terminal status first, so the audit write is best-effort. The sweep must // catch the failure and still clear the lock. - const transactionSpy = vi - .spyOn(db, "transaction") - .mockRejectedValueOnce(new Error("simulated audit write failure")); + await db.execute(sql` + create function reject_stale_lock_audit() returns trigger language plpgsql as $$ + begin raise exception 'simulated audit write failure'; end; + $$ + `); try { + await db.execute(sql` + create trigger reject_stale_lock_audit before insert on heartbeat_run_events + for each row execute function reject_stale_lock_audit() + `); const heartbeat = heartbeatService(db); const result = await heartbeat.sweepStaleIssueLocks(); expect(result.terminalizedRunIds).toEqual([runningRunId]); expect(result.cleared).toBe(1); } finally { - transactionSpy.mockRestore(); + await db.execute(sql`drop trigger if exists reject_stale_lock_audit on heartbeat_run_events`); + await db.execute(sql`drop function reject_stale_lock_audit()`); } // The run reached its terminal status even though the audit write failed. const runStatus = await db - .select({ status: heartbeatRuns.status }) + .select({ status: heartbeatRuns.status, nextEventSeq: heartbeatRuns.nextEventSeq }) .from(heartbeatRuns) .where(eq(heartbeatRuns.id, runningRunId)) - .then((rows) => rows[0]?.status); - expect(runStatus).toBe("interrupted"); + .then((rows) => rows[0]); + expect(runStatus).toEqual({ status: "interrupted", nextEventSeq: 1 }); // The sweep cleared the lock in the same pass. const lock = await db diff --git a/server/src/services/heartbeat-run-events.ts b/server/src/services/heartbeat-run-events.ts index e90f5c4e7d0..85175dc1da1 100644 --- a/server/src/services/heartbeat-run-events.ts +++ b/server/src/services/heartbeat-run-events.ts @@ -1,4 +1,4 @@ -import { and, asc, eq, or, sql } from "drizzle-orm"; +import { and, asc, eq, getTableColumns, or, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { heartbeatRunEvents, heartbeatRuns } from "@paperclipai/db"; import { nativeSha256 } from "./native-runtime/canonical.js"; @@ -68,6 +68,30 @@ export async function appendHeartbeatRunEvent( db: Db, input: AppendHeartbeatRunEventInput, ): Promise { + if (!input.nativeSource && !input.retryExhaustion) { + const allocated = db.$with("allocated_run_event", {}).as(sql` + update ${heartbeatRuns} + set next_event_seq = ${heartbeatRuns.nextEventSeq} + 1, + updated_at = ${sql.param(new Date(), heartbeatRuns.updatedAt)} + where ${heartbeatRuns.id} = ${input.runId} + and ${heartbeatRuns.companyId} = ${input.companyId} + and ${heartbeatRuns.agentId} = ${input.agentId} + returning company_id, id as run_id, agent_id, next_event_seq - 1 as seq + `); + const appended = db.$with("appended_run_event", getTableColumns(heartbeatRunEvents)).as(sql` + insert into ${heartbeatRunEvents} + (company_id, run_id, agent_id, seq, event_type, stream, level, color, message, payload) + select company_id, run_id, agent_id, seq, + ${input.eventType}, ${input.stream ?? null}, ${input.level ?? null}, + ${input.color ?? null}, ${input.message ?? null}, + ${sql.param(input.payload ?? null, heartbeatRunEvents.payload)} + from ${allocated} + returning * + `); + const [row] = await db.with(allocated, appended).select().from(appended); + if (!row) throw new Error("heartbeat_run_event_binding_mismatch"); + return { row, disposition: "committed", highestContiguousSourceSeq: 0 }; + } return db.transaction(async (tx) => { const run = await tx .select() diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index d81b50e23bc..d10effb82bf 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -66,6 +66,7 @@ import { NativeSessionProtocolIntegrityError, } from "../../vendor/paperclip-runner/index.js"; import * as issueServiceModule from "../issues.js"; +import * as runEventWriter from "../heartbeat-run-events.js"; import { NativePermissionDeclinedError } from "./native-permission-decline.js"; import { resolvePaperclipRunnerNativeProviderInput } from "./provider-profile.js"; import { @@ -9678,12 +9679,24 @@ describe("native process ownership", () => { startedAt: "2026-08-18T18:00:00.000Z", }; const onSpawn = vi.fn(async () => undefined); + const db = leaseDb(execution); + const admissionWriter = vi.spyOn(runEventWriter, "appendHeartbeatRunEvent").mockResolvedValueOnce({ + row: { + id: 1, companyId: execution.binding.companyId, runId: execution.binding.runId, + agentId: execution.binding.agentId, seq: 1, eventType: "native.process_start_requested", + stream: "system", level: "info", color: null, message: null, payload: null, + sourceInstanceId: null, sourceEventId: null, sourceSeq: null, + sourcePayloadSha256: null, protocolSchemaVersion: null, createdAt: new Date(), + }, + disposition: "committed", highestContiguousSourceSeq: 0, + }); state.createBackend.mockClear(); state.execute.mockReset().mockImplementation(async (options) => { await options.onSessionAdmission(); - expect(updates).toContainEqual({ table: heartbeatRunEvents, values: expect.objectContaining({ + expect(admissionWriter).toHaveBeenCalledWith(db, expect.objectContaining({ eventType: "native.process_start_requested", runId: execution.binding.runId, - }) }); + companyId: execution.binding.companyId, agentId: execution.binding.agentId, + })); await options.backend.onSpawn(processMetadata); return { result: { summary: "completed" }, @@ -9697,18 +9710,17 @@ describe("native process ownership", () => { highestContiguousSourceSeq: 1, }; }); - const updates: Array<{ table: unknown; values: Record }> = []; state.createBackend.mockImplementationOnce((_input, options) => { - expect(updates.some(update => update.values.eventType === "native.process_start_requested")).toBe(false); + expect(admissionWriter).not.toHaveBeenCalled(); return { kind: "test", onSpawn: options.onSpawn }; }); - await executePaperclipNativeSession({ - db: leaseDb(execution, {}, {}, updates), - execution, - runnerInstanceId: "runner", - onSpawn, - }); + try { + await executePaperclipNativeSession({ db, execution, runnerInstanceId: "runner", onSpawn }); + expect(admissionWriter).toHaveBeenCalledOnce(); + } finally { + admissionWriter.mockRestore(); + } expect(state.createBackend).toHaveBeenCalledWith( execution,