From f9581e533e899ce6c67ad7b74ee1f38321a348e5 Mon Sep 17 00:00:00 2001 From: Naeem Akhtar <32741588+naeemakhtar110@users.noreply.github.com> Date: Mon, 26 Oct 2026 01:35:36 +0300 Subject: [PATCH] perf(server): find the native event cursor through the source seq index (#16140) Find the contiguous native cursor in PostgreSQL while preserving gaps, late arrivals, source isolation, and event replay. Verify a long-history response budget against current master. Co-Authored-By: Paperclip --- doc/DATABASE.md | 6 ++ .../heartbeat-run-event-sequencing.test.ts | 101 +++++++++++++++++- server/src/services/heartbeat-run-events.ts | 42 +++++--- 3 files changed, 136 insertions(+), 13 deletions(-) diff --git a/doc/DATABASE.md b/doc/DATABASE.md index ed99988523b..e35cf130381 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -49,6 +49,12 @@ and leaves all existing cases passive. Source identities are not grants. ### Task query performance +Native event acknowledgements read a count and maximum through the existing +source-sequence index. A stream with gaps uses indexed lookups for its first +missing successor. The acknowledgement keeps its contiguous-prefix rule without +returning every historical sequence number. All events stay stored. Counting the +index entries still grows with the stream length. + The attention feed checks for newer runs with company-scoped existence probes. The probes match the failed run's agent and issue or task context. They preserve the existing millisecond comparison and empty-context rules. Each batch holds diff --git a/server/src/__tests__/heartbeat-run-event-sequencing.test.ts b/server/src/__tests__/heartbeat-run-event-sequencing.test.ts index 28c2dd00d24..d8e7bd0b9cb 100644 --- a/server/src/__tests__/heartbeat-run-event-sequencing.test.ts +++ b/server/src/__tests__/heartbeat-run-event-sequencing.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it } from "vitest"; -import { eq } from "drizzle-orm"; +import { eq, sql } from "drizzle-orm"; +import { createDatabaseResponseCapture, installDatabaseResponseCapture } from "@paperclipai/db/response-capture"; import { agents, companies, @@ -15,6 +16,62 @@ import { import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; describe("P6-11..13 / P6-17 canonical event allocator", () => { + it("keeps native cursor responses bounded with long histories and a late gap fill", async () => { + const temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-cursor-budget-"); + const db = createDb(temporary.connectionString); + const companyId = "20000000-0000-4000-8000-000000000001"; + const agentId = "20000000-0000-4000-8000-000000000002"; + const runId = "20000000-0000-4000-8000-000000000003"; + const otherRunId = "20000000-0000-4000-8000-000000000004"; + try { + await db.insert(companies).values({ id: companyId, name: "Cursor budget", issuePrefix: "CBG" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Cursor agent" }); + await db.insert(heartbeatRuns).values([ + { id: runId, companyId, agentId, status: "running", nextEventSeq: 5005 }, + { id: otherRunId, companyId, agentId, status: "running" }, + ]); + for (let batch = 0; batch < 10; batch += 1) { + await db.insert(heartbeatRunEvents).values(Array.from({ length: 500 }, (_, index) => { + const seq = batch * 500 + index + 1; + return { companyId, agentId, runId, seq, eventType: "item.delta", sourceInstanceId: "runner-a", + sourceSeq: seq, sourceEventId: `seed:${seq}` }; + })); + } + await db.insert(heartbeatRunEvents).values([ + ...[-1, 0, null].map((sourceSeq, index) => ({ + companyId, agentId, runId, seq: 5001 + index, eventType: "item.delta", + sourceInstanceId: "runner-a", sourceSeq, sourceEventId: `legacy:${index}`, + })), + { companyId, agentId, runId, seq: 5004, eventType: "item.delta", sourceInstanceId: "runner-b", + sourceSeq: 5001, sourceEventId: "other-source:5001" }, + { companyId, agentId, runId: otherRunId, seq: 1, eventType: "item.delta", sourceInstanceId: "runner-a", + sourceSeq: 5001, sourceEventId: "other-run:5001" }, + ]); + const append = (sourceSeq: number) => appendHeartbeatRunEvent(db, { + companyId, agentId, runId, eventType: "item.delta", payload: { sourceSeq }, + nativeSource: { sourceInstanceId: "runner-a", sourceSeq, sourceEventId: `runner-a:${sourceSeq}`, + protocolSchemaVersion: 1, canonicalPayload: { sourceSeq } }, + }); + const events: Array> = [], now = Date.now(); + const capture = createDatabaseResponseCapture({ start: now - 100, end: now + 59_000 }, event => events.push(event)); + installDatabaseResponseCapture(db, { heartbeatRunEvents }, capture); + try { expect((await append(5002)).highestContiguousSourceSeq).toBe(5000); } + finally { capture.flush(); } + const reads = events.filter(event => event.event === "paperclip.db_response_capture.entry.v1" + && event.table === "heartbeat_run_events" && (event.operation === "select" || event.operation === "execute")); + expect(reads.reduce((sum, event) => sum + Number(event.calls), 0)).toBeGreaterThan(0); + expect(reads.reduce((sum, event) => sum + Number(event.rows), 0)).toBeLessThan(16); + expect(reads.reduce((sum, event) => sum + Number(event.responseValueBytes), 0)).toBeLessThan(2048); + expect((await append(5001)).highestContiguousSourceSeq).toBe(5002); + const replay = await append(5002); + expect(replay.disposition).toBe("duplicate"); + expect(replay.highestContiguousSourceSeq).toBe(5002); + const [remaining] = await db.select({ count: sql`count(*)::int` }).from(heartbeatRunEvents) + .where(eq(heartbeatRunEvents.runId, runId)); + expect(remaining?.count).toBe(5006); + } finally { await temporary.cleanup(); } + }, 60_000); + it.each([false, true])("deduplicates concurrent retry exhaustion (historical receipt: %s)", async (historical) => { const temporary = await startEmbeddedPostgresTestDatabase("paperclip-exhaustion-events-"); const db = createDb(temporary.connectionString); @@ -165,4 +222,46 @@ describe("P6-11..13 / P6-17 canonical event allocator", () => { await temporary.cleanup(); } }, 60_000); + + it("reports the highest contiguous source seq per source instance across gaps and late arrivals", async () => { + const temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-cursor-"); + const db = createDb(temporary.connectionString); + const companyId = "20000000-0000-4000-8000-000000000001"; + const agentId = "20000000-0000-4000-8000-000000000002"; + const runId = "20000000-0000-4000-8000-000000000003"; + try { + await db.insert(companies).values({ id: companyId, name: "Cursor fixture", issuePrefix: "CUR" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Cursor agent" }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running" }); + const append = async (sourceInstanceId: string, sourceSeq: number) => (await appendHeartbeatRunEvent(db, { + companyId, + runId, + agentId, + eventType: "item.delta", + payload: { sourceInstanceId, sourceSeq }, + nativeSource: { + sourceInstanceId, + sourceEventId: `${sourceInstanceId}:${sourceSeq}`, + sourceSeq, + protocolSchemaVersion: 1, + canonicalPayload: { sourceInstanceId, sourceSeq }, + }, + })).highestContiguousSourceSeq; + + expect(await append("runner-a", 2)).toBe(0); + expect(await append("runner-a", 1)).toBe(2); + expect(await append("runner-a", 3)).toBe(3); + expect(await append("runner-a", 5)).toBe(3); + expect(await append("runner-a", 6)).toBe(3); + expect(await append("runner-b", 1)).toBe(1); + expect(await append("runner-a", 4)).toBe(6); + for (let sourceSeq = 7; sourceSeq <= 200; sourceSeq += 1) { + expect(await append("runner-a", sourceSeq)).toBe(sourceSeq); + } + expect(await append("runner-b", 3)).toBe(1); + expect(await append("runner-b", 2)).toBe(3); + } finally { + await temporary.cleanup(); + } + }, 60_000); }); diff --git a/server/src/services/heartbeat-run-events.ts b/server/src/services/heartbeat-run-events.ts index ecb7ff5864a..efefbc63b3c 100644 --- a/server/src/services/heartbeat-run-events.ts +++ b/server/src/services/heartbeat-run-events.ts @@ -202,18 +202,36 @@ export async function appendHeartbeatRunEvent( } async function contiguousCursor(db: Db, runId: string, sourceInstanceId: string): Promise { - const rows = await db + const scope = and( + eq(heartbeatRunEvents.runId, runId), + eq(heartbeatRunEvents.sourceInstanceId, sourceInstanceId), + sql`${heartbeatRunEvents.sourceSeq} >= 1`, + ); + const [range] = await db + .select({ + count: sql`count(*)`.mapWith(Number), + max: sql`max(${heartbeatRunEvents.sourceSeq})`.mapWith(Number), + }) + .from(heartbeatRunEvents) + .where(scope); + if (!range || !range.count) return 0; + if (range.max === range.count) return range.count; + const [first] = await db .select({ sourceSeq: heartbeatRunEvents.sourceSeq }) .from(heartbeatRunEvents) - .where(and( - eq(heartbeatRunEvents.runId, runId), - eq(heartbeatRunEvents.sourceInstanceId, sourceInstanceId), - )) - .orderBy(asc(heartbeatRunEvents.sourceSeq)); - let cursor = 0; - for (const row of rows) { - if (row.sourceSeq === cursor + 1) cursor += 1; - else if (row.sourceSeq !== null && row.sourceSeq > cursor + 1) break; - } - return cursor; + .where(and(scope, eq(heartbeatRunEvents.sourceSeq, 1))) + .limit(1); + if (!first) return 0; + const [last] = await db + .select({ sourceSeq: heartbeatRunEvents.sourceSeq }) + .from(heartbeatRunEvents) + .where(and(scope, sql`not exists ( + select 1 from heartbeat_run_events next_event + where next_event.run_id = heartbeat_run_events.run_id + and next_event.source_instance_id = heartbeat_run_events.source_instance_id + and next_event.source_seq = heartbeat_run_events.source_seq + 1 + )`)) + .orderBy(asc(heartbeatRunEvents.sourceSeq)) + .limit(1); + return last?.sourceSeq ?? 0; }