From 577bd86cdf28c16424ea1ef365248b806ed7fa25 Mon Sep 17 00:00:00 2001 From: Bestony Date: Fri, 31 Jul 2026 11:28:24 +0800 Subject: [PATCH 1/2] fix(server): bound inbox ACK cost to the partition's non-acked rows `ackThroughEntryIdForBoundAgents` selected - and, under `FOR UPDATE`, locked - every notify=true row in the `(inbox_id, chat_id)` partition below the ACK cursor, including rows acked long ago, then materialised them in JS. Cost per ACK was O(chat history) and O(N^2) over a chat's life; a duplicate ACK that commits nothing paid the same price. The cause was not a missing index. `idx_inbox_chat_silent` is (inbox_id, chat_id, notify, status) and the planner was already using it - the query simply never constrained `status`, so a four-column index was used as a three-column one and `id <= cursor` became a heap filter over the whole partition. Already-acked rows cannot change any of the three decisions this commit makes: they never form a prefix gap, are never committable, and are never reset-from-pending. Excluding them is an exact equivalence, not an approximation, and it holds because `acked` is terminal for status transitions - every UPDATE here is guarded on 'pending' or 'delivered', and the only statement touching acked rows is the GC DELETE, restricted to notify=false and therefore disjoint from this notify=true scan. Spelled as a positive IN because `status <> 'acked'` is not sargable: measured on the same dataset it stays a filter, reads 168,020 rows and saves only the row locks. The status set is derived from the shared domain minus the terminal value rather than hardcoded, so a future fourth status keeps being scanned and can still reject a commit as a gap; a literal pair would start skipping it and let the ACK through. The domain has changed once already (a legacy `failed` value cleared by migration 0066) and its CHECK constraint was added NOT VALID. No schema change, no new index, no migration. Measured end-to-end through the real service call, median of 40 samples, on both major versions this repo runs (docker-compose uses postgres:16-alpine, CI uses postgres:17): history before after (17.10) after (16.14) 50,000 119.6 ms 1.46 ms 2.53 ms 150,000 341.4 ms 2.25 ms 2.77 ms 300,000 545.0 ms 2.66 ms 2.61 ms Flat instead of linear; the residual is the transaction's fixed cost, which an empty-history ACK also pays. At the query level the duplicate ACK goes from 168,000 rows / 2,937 buffers to 0 rows / 4 buffers, and with FOR UPDATE from 338,980 buffers / 2,800 dirtied pages to 44 / 0. `LockRows` still sits above the `Sort`, so rows are still locked in ascending id order. Cost now tracks the number of non-acked rows in the partition rather than the history. `id <= cursor` remains a filter, so a large pending backlog still costs proportionally to that backlog. Refs #1671 --- .../__tests__/inbox-ack-prefix-scan.test.ts | 195 ++++++++++++++++++ packages/server/src/services/inbox.ts | 37 ++++ 2 files changed, 232 insertions(+) create mode 100644 packages/server/src/__tests__/inbox-ack-prefix-scan.test.ts diff --git a/packages/server/src/__tests__/inbox-ack-prefix-scan.test.ts b/packages/server/src/__tests__/inbox-ack-prefix-scan.test.ts new file mode 100644 index 000000000..4f101b780 --- /dev/null +++ b/packages/server/src/__tests__/inbox-ack-prefix-scan.test.ts @@ -0,0 +1,195 @@ +import { INBOX_ENTRY_STATUSES, inboxEntryStatusSchema } from "@first-tree/shared"; +import { and, eq, sql } from "drizzle-orm"; +import { drizzle } from "drizzle-orm/postgres-js"; +import type { FastifyInstance } from "fastify"; +import postgres from "postgres"; +import { describe, expect, it } from "vitest"; +import type { Database } from "../db/connection.js"; +import { inboxEntries } from "../db/schema/inbox-entries.js"; +import * as schema from "../db/schema/index.js"; +import { messages } from "../db/schema/messages.js"; +import * as inboxService from "../services/inbox.js"; +import { createTestAgent, useTestApp } from "./helpers.js"; + +/** + * Guards for the ACK-through prefix scan (PERF-008 / #1671). + * + * The scan used to read every notify=true row in the `(inbox_id, chat_id)` + * partition below the cursor, including rows acked long ago, making a single + * ACK cost O(chat history). These tests pin the two properties that keep it + * bounded, plus the `chat_id IS NULL` branch that the existing ACK coverage + * does not reach. + */ +describe("inbox ACK prefix scan", () => { + const getApp = useTestApp(); + + /** Create a chat between two agents and bulk-seed `history` acked rows for + * the recipient, followed by `live` delivered rows it can still ACK. */ + async function seedHistory( + app: FastifyInstance, + opts: { history: number; live: number; nullChat?: boolean }, + ): Promise<{ inboxId: string; chatId: string; liveIds: number[] }> { + const uid = crypto.randomUUID().slice(0, 8); + const sender = await createTestAgent(app, { name: `ack-s-${uid}` }); + const recipient = await createTestAgent(app, { name: `ack-r-${uid}` }); + const chatRes = await sender.request("POST", "/api/v1/agent/chats", { + type: "group", + participantIds: [recipient.agent.uuid], + }); + const chatId: string = chatRes.json().id; + const inboxId = recipient.agent.inboxId; + const total = opts.history + opts.live; + + // Bulk-insert rather than sending `total` real messages: this exercises the + // ACK read path, and the fan-out path has its own coverage elsewhere. + const messageRows = Array.from({ length: total }, (_, i) => ({ + id: `ack-msg-${uid}-${i}`, + chatId, + senderId: sender.agent.uuid, + format: "text", + content: `seed ${i}`, + source: "api" as const, + })); + await app.db.insert(messages).values(messageRows); + await app.db.insert(inboxEntries).values( + messageRows.map((m, i) => ({ + inboxId, + messageId: m.id, + chatId: opts.nullChat ? null : chatId, + notify: true, + status: i < opts.history ? INBOX_ENTRY_STATUSES.ACKED : INBOX_ENTRY_STATUSES.DELIVERED, + deliveredAt: new Date(), + ackedAt: i < opts.history ? new Date() : null, + })), + ); + await app.db.execute(sql`ANALYZE inbox_entries`); + + const live = await app.db + .select({ id: inboxEntries.id }) + .from(inboxEntries) + .where(and(eq(inboxEntries.inboxId, inboxId), eq(inboxEntries.status, INBOX_ENTRY_STATUSES.DELIVERED))); + return { inboxId, chatId, liveIds: live.map((r) => r.id).sort((a, b) => a - b) }; + } + + describe("scanned status set", () => { + it("is derived from the status domain rather than hardcoded", () => { + // The direction of this assertion is the point. If a fourth status is + // added, it must land in the scanned set so an ACK can still reject it + // as a prefix gap. A hardcoded ["pending", "delivered"] would silently + // skip it and let the commit through. + expect([...inboxService.ACK_PREFIX_SCAN_STATUSES].sort()).toEqual( + inboxEntryStatusSchema.options.filter((s) => s !== INBOX_ENTRY_STATUSES.ACKED).sort(), + ); + expect(inboxService.ACK_PREFIX_SCAN_STATUSES).not.toContain(INBOX_ENTRY_STATUSES.ACKED); + }); + + it("covers every status the database itself allows", async () => { + // Cross-check against the live CHECK constraint: it was added NOT VALID + // and the domain has changed once before (a legacy `failed` value). If + // the database domain and the shared enum ever diverge, the derived set + // stops being equivalent to "everything except acked" and this fails. + const [row] = await getApp().db.execute<{ def: string }>(sql` + SELECT pg_get_constraintdef(oid) AS def + FROM pg_constraint WHERE conname = 'ck_inbox_entries_status'`); + const dbDomain = [...String(row?.def ?? "").matchAll(/'([a-z_]+)'/g)].map((m) => m[1]).sort(); + expect(dbDomain).toEqual([...inboxEntryStatusSchema.options].sort()); + }); + }); + + it("keeps ACK cost independent of chat history", async () => { + const app = getApp(); + + /** Run one real ACK, capturing every statement the service emits, then + * replay each against EXPLAIN ANALYZE and report the widest scan. + * + * Capturing the service's own SQL — rather than a copy of the query + * written into this test — is what makes the guard survive refactors: + * inlining the query back into the service, or dropping the status + * restriction, both show up here. */ + async function widestScan(cursor: number, inboxId: string): Promise { + const captured: Array<{ query: string; params: unknown[] }> = []; + const client = postgres(process.env.DATABASE_URL ?? "", { max: 1 }); + const instrumented = Object.assign( + drizzle(client, { + schema, + logger: { logQuery: (query, params) => captured.push({ query, params }) }, + }), + { end: () => client.end() }, + ) as unknown as Database; + + try { + await inboxService.ackThroughEntryIdForBoundAgents(instrumented, cursor, [inboxId]); + const reads = captured.filter((c) => /^\s*select/i.test(c.query) && c.query.includes("inbox_entries")); + expect(reads.length).toBeGreaterThan(0); + + let widest = 0; + for (const c of reads) { + const plan = await client.unsafe(`EXPLAIN (ANALYZE, FORMAT JSON) ${c.query}`, c.params as never[]); + widest = Math.max(widest, deepestActualRows(plan[0]?.["QUERY PLAN"]?.[0]?.Plan)); + } + return widest; + } finally { + await client.end(); + } + } + + const small = await seedHistory(app, { history: 300, live: 4 }); + const smallScan = await widestScan(small.liveIds[0] ?? 0, small.inboxId); + + const large = await seedHistory(app, { history: 1500, live: 4 }); + const largeScan = await widestScan(large.liveIds[0] ?? 0, large.inboxId); + + // Five times the history must not cost more rows read. A few rows of slack + // absorbs the live window; it is nowhere near the 1200-row difference a + // history-proportional scan would produce. + expect(largeScan).toBeLessThanOrEqual(smallScan + 8); + expect(largeScan).toBeLessThan(100); + }); + + it("commits through a long acked history in the chat_id IS NULL partition", async () => { + const app = getApp(); + const { inboxId, liveIds } = await seedHistory(app, { history: 400, live: 3, nullChat: true }); + const target = liveIds[liveIds.length - 1]; + expect(target).toBeDefined(); + + const result = await inboxService.ackThroughEntryIdForBoundAgents(app.db, target ?? 0, [inboxId]); + expect(result.ok).toBe(true); + if (!result.ok) return; + expect(result.disposition).toBe("acked"); + expect(result.ackedCount).toBe(3); + + const remaining = await app.db + .select({ id: inboxEntries.id }) + .from(inboxEntries) + .where(and(eq(inboxEntries.inboxId, inboxId), eq(inboxEntries.status, INBOX_ENTRY_STATUSES.DELIVERED))); + expect(remaining).toHaveLength(0); + }); + + it("still rejects a gap that sits behind a long acked history", async () => { + const app = getApp(); + const { inboxId, liveIds } = await seedHistory(app, { history: 400, live: 3 }); + const [first, , third] = liveIds; + expect(first).toBeDefined(); + + // Turn the first live row into a never-delivered pending row: it is now a + // genuine prefix gap sitting far above the acked history. + await app.db + .update(inboxEntries) + .set({ status: INBOX_ENTRY_STATUSES.PENDING, deliveredAt: null }) + .where(eq(inboxEntries.id, first ?? 0)); + + const rejected = await inboxService.ackThroughEntryIdForBoundAgents(app.db, third ?? 0, [inboxId]); + expect(rejected).toEqual({ ok: false, reason: "prefix_gap" }); + }); +}); + +type PlanNode = { "Actual Rows"?: number; Plans?: PlanNode[] }; + +/** Rows touched by the deepest scan node — what "how much did this read" means + * once the planner has layered sorts and lock nodes on top. */ +function deepestActualRows(node: PlanNode | undefined): number { + if (!node) return 0; + const children = node.Plans ?? []; + if (children.length === 0) return node["Actual Rows"] ?? 0; + return Math.max(...children.map(deepestActualRows)); +} diff --git a/packages/server/src/services/inbox.ts b/packages/server/src/services/inbox.ts index 62ab8d04a..84ebd048e 100644 --- a/packages/server/src/services/inbox.ts +++ b/packages/server/src/services/inbox.ts @@ -1,4 +1,5 @@ import { + INBOX_ENTRY_STATUSES, type InboxEntryWithMessage, inboxEntryStatusSchema, messageSourceSchema, @@ -18,6 +19,23 @@ import { buildClientMessagePayloadsForInbox } from "./message-dispatcher.js"; * conversions (bigserial → number, timestamp → Date) flow through. */ type ClaimedEntry = typeof inboxEntries.$inferSelect; +/** + * The statuses whose rows can still change an ACK-through decision. + * + * Derived from the shared status domain minus the terminal one rather than + * written out as a literal pair. The direction matters: if a fourth status is + * ever added, a derived set keeps scanning it — so an unknown status can still + * reject the commit as a prefix gap, exactly as the original status-agnostic + * query did. A hand-written `["pending", "delivered"]` would instead start + * skipping those rows and let the ACK through, turning a safe rejection into + * silent data loss. The table's status domain has changed once already (a + * legacy `failed` value, normalised by migration 0066), and its CHECK + * constraint was added `NOT VALID`, so this is not a hypothetical. + */ +export const ACK_PREFIX_SCAN_STATUSES = inboxEntryStatusSchema.options.filter( + (status) => status !== INBOX_ENTRY_STATUSES.ACKED, +); + export type AckEntryResult = | { ok: true; @@ -606,6 +624,24 @@ export async function ackThroughEntryIdForBoundAgents( if (!entry.notify) return { ok: false, reason: "non_notify" }; const chatPredicate = entry.chatId === null ? isNull(inboxEntries.chatId) : eq(inboxEntries.chatId, entry.chatId); + // Scan only the rows that can still change the outcome. Already-acked + // rows flip none of the three decisions made below: they never form a + // prefix gap, are never committable, and are never reset-from-pending. + // Excluding them is therefore an exact equivalence rather than an + // approximation, and it holds because `acked` is terminal for status + // transitions — every UPDATE against this table is guarded on 'pending' + // or 'delivered', and the single statement that touches acked rows at + // all is the GC DELETE in `pruneStaleSilentEntries`, restricted to + // notify=false. + // + // This clause is also what bounds the scan. `idx_inbox_chat_silent` is + // (inbox_id, chat_id, notify, status); leaving `status` unconstrained + // used only three of its four equality columns, so the planner had to + // walk every row of the partition and re-check `id <= cursor` as a + // filter — O(chat history) per ACK and O(N^2) over a chat's life, even + // for a duplicate ACK that commits nothing. It has to be a positive IN: + // `status <> 'acked'` is not sargable, stays a filter, and saves only + // the row locks. const prefixRows = await tx .select() .from(inboxEntries) @@ -614,6 +650,7 @@ export async function ackThroughEntryIdForBoundAgents( eq(inboxEntries.inboxId, entry.inboxId), chatPredicate, eq(inboxEntries.notify, true), + inArray(inboxEntries.status, ACK_PREFIX_SCAN_STATUSES), sql`${inboxEntries.id} <= ${entryId}`, ), ) From 82ef6d599105555a394f990cac394aae238a0bc8 Mon Sep 17 00:00:00 2001 From: Bestony Date: Fri, 31 Jul 2026 11:46:01 +0800 Subject: [PATCH 2/2] docs(server): record the plan-cache caveat and the cost-guard's replay limit Two review follow-ups, comments only. The status restriction only reaches the index condition under a custom plan on PostgreSQL 16. It is stable under the default plan_cache_mode on 16.14 and 17.10 because the generic estimate is far more expensive, but a deployment forcing force_generic_plan globally would silently return the scan to O(history) there; 17 is unaffected. Worth stating next to the clause rather than only in review. The cost guard replays the captured SQL after the ACK has committed, so a correct implementation matches close to zero rows instead of the handful it saw live. It still separates the two cases -- a query without the restriction reads the whole partition on replay -- but the absolute number is not the live scan, and the next reader should not mistake it for one. Refs #1671 --- .../server/src/__tests__/inbox-ack-prefix-scan.test.ts | 8 +++++++- packages/server/src/services/inbox.ts | 8 ++++++++ 2 files changed, 15 insertions(+), 1 deletion(-) diff --git a/packages/server/src/__tests__/inbox-ack-prefix-scan.test.ts b/packages/server/src/__tests__/inbox-ack-prefix-scan.test.ts index 4f101b780..555b00d45 100644 --- a/packages/server/src/__tests__/inbox-ack-prefix-scan.test.ts +++ b/packages/server/src/__tests__/inbox-ack-prefix-scan.test.ts @@ -105,7 +105,13 @@ describe("inbox ACK prefix scan", () => { * Capturing the service's own SQL — rather than a copy of the query * written into this test — is what makes the guard survive refactors: * inlining the query back into the service, or dropping the status - * restriction, both show up here. */ + * restriction, both show up here. + * + * The replay runs after the ACK has committed, so a correct + * implementation matches close to zero rows rather than the handful it + * saw live. What is being asserted is the presence of the restriction, + * not the live row count: a query missing it still reads the whole + * partition on replay, which is what makes the two cases separable. */ async function widestScan(cursor: number, inboxId: string): Promise { const captured: Array<{ query: string; params: unknown[] }> = []; const client = postgres(process.env.DATABASE_URL ?? "", { max: 1 }); diff --git a/packages/server/src/services/inbox.ts b/packages/server/src/services/inbox.ts index 84ebd048e..e4ac353e3 100644 --- a/packages/server/src/services/inbox.ts +++ b/packages/server/src/services/inbox.ts @@ -642,6 +642,14 @@ export async function ackThroughEntryIdForBoundAgents( // for a duplicate ACK that commits nothing. It has to be a positive IN: // `status <> 'acked'` is not sargable, stays a filter, and saves only // the row locks. + // + // Operational caveat: on PostgreSQL 16 the index condition is only + // reached under a custom plan. Measured stable under the default + // `plan_cache_mode = auto` on 16.14 and 17.10 — the generic estimate is + // far more expensive, so the planner keeps rejecting it — but a + // deployment that forces `force_generic_plan` globally would silently + // return this scan to O(history) on 16. PostgreSQL 17 uses the bound + // parameters as an index condition either way. const prefixRows = await tx .select() .from(inboxEntries)