diff --git a/doc/DATABASE.md b/doc/DATABASE.md index ee653377291..461c9ca63af 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -463,6 +463,18 @@ benchmark loads that revision's chat query, uses synthetic data only, compares complete returned rows, and prints timings and `EXPLAIN (ANALYZE, BUFFERS)` plans. Local results measure query execution, not staging request latency. +The inbound conversation head reads one endpoint's open delivery IDs first. These +are received, retry and processing deliveries (from `chat_deliveries_work_idx`) +and processed deliveries that the endpoint's issued or processing `inbound_wakeup` +actions still own. Only those rows are fetched, filtered by thread and sorted. The +query does not scan the endpoint's processed history or read its +`normalized_event` values. To compare it with a pre-change revision on a busy +endpoint fixture: + +```sh +pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-inbound-queue-discovery.ts --baseline-ref= --endpoint-history=20000 --only=endpoint-history --require-work-budget +``` + Run history defaults to 200 rows and permits at most 1,000 rows, including direct service calls. The UI client requests 200 rows when its caller omits a limit. Context summaries expand the stored JSON once per returned row, retaining diff --git a/scripts/benchmark-inbound-queue-discovery.ts b/scripts/benchmark-inbound-queue-discovery.ts index 9486a6736a5..13a12daa577 100644 --- a/scripts/benchmark-inbound-queue-discovery.ts +++ b/scripts/benchmark-inbound-queue-discovery.ts @@ -9,6 +9,16 @@ import { captureInboundQuery, runQuery, seedInboundDiscovery, type QueueFactory, const ref = process.argv.find(arg => arg.startsWith("--baseline-ref="))?.slice(15); assert.ok(ref, "Pass an explicit pre-change --baseline-ref="); +// `--endpoint-history=` adds one endpoint with a long processed history +// and 4 KiB payloads, like a busy hosted bot. `--only=` selects one fixture. +const endpointHistoryArg = process.argv.find(arg => arg.startsWith("--endpoint-history="))?.slice(19); +const endpointHistory = endpointHistoryArg === undefined ? 0 : Number(endpointHistoryArg); +assert.ok(endpointHistoryArg === undefined || (Number.isSafeInteger(endpointHistory) && endpointHistory > 0), "--endpoint-history must be a positive integer"); +const only = process.argv.find(arg => arg.startsWith("--only="))?.slice(7) ?? (process.argv.includes("--retained-only") ? "retained-history" : undefined); +const fixtures = [{ name: "small", history: 20, payloadBytes: 0 }, { name: "retained-history", history: 1500, payloadBytes: 32 * 1024 }, + ...(endpointHistory > 0 ? [{ name: "endpoint-history", history: endpointHistory, payloadBytes: 4096 }] : [])]; +const selectedFixtures = fixtures.filter(value => !only || value.name === only); +assert.ok(selectedFixtures.length > 0, `No fixture matches --only=${only}; endpoint-history also needs --endpoint-history=`); const baselineRevision = execFileSync("git", ["rev-parse", "--verify", `${ref}^{commit}`], { encoding: "utf8" }).trim(); const root = fileURLToPath(new URL("../", import.meta.url)); const sources = ["server/src/services/chat-channels/inbound/delivery-ledger.ts", "server/src/services/chat-channels/inbound/conversation-queue.ts", "server/src/services/chat-channels.ts"]; @@ -43,7 +53,7 @@ try { const malformedReadyControls = []; let baselineExceedsWorkBudget = false; const budgetFailures: string[] = []; - for (const fixture of [{ name: "small", history: 20, payloadBytes: 0 }, { name: "retained-history", history: 1500, payloadBytes: 32 * 1024 }].filter(value => !process.argv.includes("--retained-only") || value.name === "retained-history")) { + for (const fixture of selectedFixtures) { const f = await seedInboundDiscovery(db, fixture.history, fixture.payloadBytes); await db.$client.unsafe("begin"); if (process.argv.includes("--jit-off")) await db.$client.unsafe("set local jit = off"); @@ -89,9 +99,21 @@ try { if (kind !== "ready" && newPlan.JIT) budgetFailures.push(`${kind}: unnecessary JIT on the retained-history fixture`); if (kind === "deadline" && newBuffers > 1000) budgetFailures.push(`${kind}: ${newBuffers} shared buffer accesses exceeds fixture budget 1000`); } + // The conversation head must start from open work. It must not visit + // the endpoint's processed delivery history to find the few open rows. + const deliveryRowsVisited = {} as Record<"before" | "after", number>; + for (const arm of ["before", "after"] as const) { + deliveryRowsVisited[arm] = nodes((plans[arm] as { Plan: Record }[])[0]!.Plan) + .filter(node => node["Relation Name"] === "chat_deliveries") + .reduce((sum, node) => sum + (Number(node["Actual Rows"] ?? 0) + Number(node["Rows Removed by Filter"] ?? 0)) * Number(node["Actual Loops"] ?? 1), 0); + } + if (kind === "head" && fixture.name !== "small") { + baselineExceedsWorkBudget ||= deliveryRowsVisited.before > 100; + if (deliveryRowsVisited.after > 100) budgetFailures.push(`head: visits ${deliveryRowsVisited.after} delivery rows on the ${fixture.name} fixture; budget 100`); + } const median = (values: number[]) => [...values].sort((a, b) => a - b)[Math.floor(values.length / 2)]!; - results.push({ fixture, kind, equalPositionalRows: expected.length, before, after, samples, beforeMedianMs: median(samples.before), afterMedianMs: median(samples.after), actionLoops, plans }); - process.stderr.write(`${fixture.name}/${kind}: ${median(samples.before).toFixed(2)} -> ${median(samples.after).toFixed(2)} ms; action scans ${actionLoops.before} -> ${actionLoops.after}\n`); + results.push({ fixture, kind, equalPositionalRows: expected.length, before, after, samples, beforeMedianMs: median(samples.before), afterMedianMs: median(samples.after), actionLoops, deliveryRowsVisited, plans }); + process.stderr.write(`${fixture.name}/${kind}: ${median(samples.before).toFixed(2)} -> ${median(samples.after).toFixed(2)} ms; action scans ${actionLoops.before} -> ${actionLoops.after}; delivery rows visited ${deliveryRowsVisited.before} -> ${deliveryRowsVisited.after}\n`); } } finally { connection.release(); } if (fixture.name === "small") { diff --git a/server/src/__tests__/inbound-queue-discovery.test.ts b/server/src/__tests__/inbound-queue-discovery.test.ts index 5bd8f84589b..b66d65a7ef9 100644 --- a/server/src/__tests__/inbound-queue-discovery.test.ts +++ b/server/src/__tests__/inbound-queue-discovery.test.ts @@ -1,9 +1,9 @@ import { randomUUID } from "node:crypto"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; -import { and, eq, getTableColumns, sql } from "drizzle-orm"; +import { and, eq, getTableColumns, inArray, notInArray, or, sql } from "drizzle-orm"; import { chatActions, chatDeliveries, chatEndpoints, createDb, startEmbeddedPostgresTestDatabase, type Db } from "@paperclipai/db"; import { createChatConversationDeliveryQueue } from "../services/chat-channels/inbound/conversation-queue.js"; -import { pendingInboundWakeupCondition } from "../services/chat-channels/inbound/delivery-ledger.js"; +import { conversationDeliveryOrder, openConversationDeliveryIds, pendingInboundWakeupCondition } from "../services/chat-channels/inbound/delivery-ledger.js"; import { captureInboundQuery, pendingIds, queueOptions, runQuery, seedInboundDiscovery } from "./helpers/inbound-discovery-fixture.js"; let database: Awaited>; @@ -29,6 +29,56 @@ describe("inbound queue discovery", () => { expect((await pendingIds(db, pendingInboundWakeupCondition(), f.companyId)).map(row => row.id).sort()).toEqual([f.pending.id, duplicate.id].sort()); }); + it("selects the same open conversation work as the endpoint state predicate", async () => { + const f = await seedInboundDiscovery(db); + const other = await seedInboundDiscovery(db, 0); + const otherEndpoint = await f.anotherEndpoint(); + const open = await Promise.all((["received", "retry", "processing"] as const).map(state => f.delivery({ state }))); + const closed = await Promise.all((["filtered", "failed", "processed"] as const).map(state => f.delivery({ state }))); + const processingWake = await f.delivery({ state: "processed" }); + await f.action(processingWake.id, { status: "processing" }); + const duplicateWake = await f.delivery({ state: "processed" }); + await f.action(duplicateWake.id); await f.action(duplicateWake.id); + const wakeForOpenDelivery = await f.delivery({ state: "retry" }); + await f.action(wakeForOpenDelivery.id); + const wakeForFailedDelivery = await f.delivery({ state: "failed" }); + await f.action(wakeForFailedDelivery.id); + const notPending = await f.delivery({ state: "processed" }); + await f.action(notPending.id, { status: "processed" }); + await f.action(notPending.id, { kind: "other" }); + await f.action(notPending.id, { endpointId: otherEndpoint }); + await f.action(null); + const crossEndpointWake = await f.delivery({ state: "processed", endpointId: otherEndpoint }); + await f.action(crossEndpointWake.id); + const otherEndpointWake = await f.delivery({ state: "processed", endpointId: otherEndpoint }); + await f.action(otherEndpointWake.id, { endpointId: otherEndpoint }); + const reaction = await f.delivery({ state: "received", eventKind: "reaction_added" }); + const oracle = or(inArray(chatDeliveries.state, ["received", "retry", "processing"]), + and(eq(chatDeliveries.state, "processed"), pendingInboundWakeupCondition())); + for (const endpointId of [f.endpointId, other.endpointId, otherEndpoint]) { + const expected = await db.select({ id: chatDeliveries.id }).from(chatDeliveries) + .where(and(eq(chatDeliveries.endpointId, endpointId), oracle)).orderBy(chatDeliveries.id); + const actual = await db.select({ id: chatDeliveries.id }).from(chatDeliveries) + .where(and(eq(chatDeliveries.endpointId, endpointId), inArray(chatDeliveries.id, openConversationDeliveryIds(endpointId)))).orderBy(chatDeliveries.id); + expect(actual).toEqual(expected); + } + const selected = (await db.select({ id: chatDeliveries.id }).from(chatDeliveries) + .where(and(eq(chatDeliveries.endpointId, f.endpointId), inArray(chatDeliveries.id, openConversationDeliveryIds(f.endpointId))))).map(row => row.id).sort(); + for (const row of [f.pending, ...open, processingWake, duplicateWake, wakeForOpenDelivery, reaction]) expect(selected).toContain(row.id); + expect(new Set(selected).size).toBe(selected.length); + for (const row of [...closed, wakeForFailedDelivery, notPending, other.pending, crossEndpointWake, otherEndpointWake]) expect(selected).not.toContain(row.id); + expect((await db.select({ id: chatDeliveries.id }).from(chatDeliveries) + .where(and(eq(chatDeliveries.endpointId, otherEndpoint), inArray(chatDeliveries.id, openConversationDeliveryIds(otherEndpoint))))).map(row => row.id)).toEqual([otherEndpointWake.id]); + // The real head query returns the oracle's first row in conversation order. + const earliest = await f.delivery({ state: "processed", normalizedEvent: { conversation: { externalThreadId: f.threadId }, message: { providerSentAt: "2025-12-31T00:00:00Z" } } }); + await f.action(earliest.id, { status: "processing" }); + const [expectedHead] = await db.select({ id: chatDeliveries.id }).from(chatDeliveries).where(and(eq(chatDeliveries.endpointId, f.endpointId), oracle, + notInArray(chatDeliveries.eventKind, ["reaction_added", "reaction_removed"]), sql`${chatDeliveries.normalizedEvent}->'conversation'->>'externalThreadId' = ${f.threadId}`)) + .orderBy(...conversationDeliveryOrder()).limit(1); + expect(expectedHead?.id).toBe(earliest.id); + expect((await runQuery(db, await captureInboundQuery(db, f, "head")))[0]?.[0]).toBe(earliest.id); + }); + it("preserves scoped ready-head probes with unrelated malformed retry data", async () => { const f = await seedInboundDiscovery(db); const other = await seedInboundDiscovery(db, 0); diff --git a/server/src/services/chat-channels/inbound/conversation-queue.ts b/server/src/services/chat-channels/inbound/conversation-queue.ts index 6b0c4af9cfa..c8fa5407595 100644 --- a/server/src/services/chat-channels/inbound/conversation-queue.ts +++ b/server/src/services/chat-channels/inbound/conversation-queue.ts @@ -60,7 +60,7 @@ import type { Message, Thread, } from "chat"; -import { DELIVERY_PROCESSING_STALE_MS, deliveryContentWasRedacted, setupDestinationActivationSuppressed, normalizedDeliveryThreadId, normalizedLifecycleTargetEventId, normalizedLifecycleEffect, deliveryReady, conversationDeliveryOrder, pendingInboundWakeupCondition, lifecycleRuntimeFence } from "./delivery-ledger.js"; +import { DELIVERY_PROCESSING_STALE_MS, deliveryContentWasRedacted, setupDestinationActivationSuppressed, normalizedDeliveryThreadId, normalizedLifecycleTargetEventId, normalizedLifecycleEffect, deliveryReady, conversationDeliveryOrder, pendingInboundWakeupCondition, openConversationDeliveryIds, lifecycleRuntimeFence } from "./delivery-ledger.js"; import type { createChatMessageLifecycle } from "./lifecycle.js"; import type { LiveInboundMessage } from "../runtime-types.js"; type DeliveryRow = typeof chatDeliveries.$inferSelect; @@ -102,13 +102,7 @@ export function createChatConversationDeliveryQueue(db: Db, options: Conversatio .where( and( eq(chatDeliveries.endpointId, endpointId), - or( - inArray(chatDeliveries.state, ["received", "retry", "processing"]), - and( - eq(chatDeliveries.state, "processed"), - pendingInboundWakeupCondition(), - ), - ), + inArray(chatDeliveries.id, openConversationDeliveryIds(endpointId)), notInArray(chatDeliveries.eventKind, [ "reaction_added", "reaction_removed", diff --git a/server/src/services/chat-channels/inbound/delivery-ledger.ts b/server/src/services/chat-channels/inbound/delivery-ledger.ts index cb6938024f4..e8da4fb846f 100644 --- a/server/src/services/chat-channels/inbound/delivery-ledger.ts +++ b/server/src/services/chat-channels/inbound/delivery-ledger.ts @@ -210,6 +210,36 @@ const DELIVERY_PROCESSING_STALE_MS = 60_000; )`; } + /** + * IDs of one endpoint's open conversation work: a received, retry or + * processing delivery, or a processed delivery whose inbound wake is still + * issued or processing. This is the same set as the endpoint filter with + * `state in (...) or (state = 'processed' and pendingInboundWakeupCondition())`. + * + * Each branch starts from a small set (open states, or the endpoint's + * pending wake actions). The caller does not scan the endpoint's processed + * history or read its `normalized_event` values to find the few open rows. + */ + function openConversationDeliveryIds(endpointId: string): SQL { + return sql`( + select open_delivery.id + from ${chatDeliveries} open_delivery + where open_delivery.endpoint_id = ${endpointId} + and open_delivery.state in ('received', 'retry', 'processing') + union all + select woken_delivery.id + from ${chatActions} pending_wake + join ${chatDeliveries} woken_delivery + on woken_delivery.id = pending_wake.delivery_id + and woken_delivery.company_id = pending_wake.company_id + and woken_delivery.endpoint_id = pending_wake.endpoint_id + where pending_wake.endpoint_id = ${endpointId} + and pending_wake.kind = 'inbound_wakeup' + and pending_wake.status in ('issued', 'processing') + and woken_delivery.state = 'processed' + )`; + } + function lifecycleRuntimeFence( delivery: DeliveryRow, ): LifecycleRuntimeFence | null { @@ -247,4 +277,4 @@ const DELIVERY_PROCESSING_STALE_MS = 60_000; return { ...normalized, runtimeContext }; } -export { DELIVERY_PROCESSING_STALE_MS, deliveryContentWasRedacted, setupDestinationActivationSuppressed, normalizedDeliveryThreadId, normalizedLifecycleTargetEventId, normalizedLifecycleEffect, deliveryReady, conversationDeliveryOrder, pendingInboundWakeupCondition, lifecycleRuntimeFence, normalizedLifecycleWithRuntimeFence }; +export { DELIVERY_PROCESSING_STALE_MS, deliveryContentWasRedacted, setupDestinationActivationSuppressed, normalizedDeliveryThreadId, normalizedLifecycleTargetEventId, normalizedLifecycleEffect, deliveryReady, conversationDeliveryOrder, pendingInboundWakeupCondition, openConversationDeliveryIds, lifecycleRuntimeFence, normalizedLifecycleWithRuntimeFence };