From 94c60683b8b437e4eaeda574812b36c017de543c Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 25 Oct 2026 06:17:50 -0500 Subject: [PATCH] fix(server): index the finite voice report window Co-Authored-By: Paperclip --- doc/idle-voice-call-reports.md | 16 ++- ...026-10-25-voice-call-report-sleep-audit.md | 42 ++++++ packages/db/src/client.ts | 1 + .../0389_workable_spencer_smythe.sql | 1 + ...{0384_snapshot.json => 0389_snapshot.json} | 126 +++++++++++++++++- packages/db/src/migrations/meta/_journal.json | 7 + packages/db/src/schema/chat_voice_sessions.ts | 2 + .../src/__tests__/voice-call-reports.test.ts | 38 ++++++ .../src/services/voice/voice-call-reports.ts | 13 +- 9 files changed, 234 insertions(+), 12 deletions(-) create mode 100644 packages/db/src/migrations/0389_workable_spencer_smythe.sql rename packages/db/src/migrations/meta/{0384_snapshot.json => 0389_snapshot.json} (99%) diff --git a/doc/idle-voice-call-reports.md b/doc/idle-voice-call-reports.md index a7ef6fb826a..a05c041058c 100644 --- a/doc/idle-voice-call-reports.md +++ b/doc/idle-voice-call-reports.md @@ -3,7 +3,8 @@ Voice report reconciliation uses the shared delivery coordinator and scheduler. It no longer runs from the chat maintenance interval. Existing `chat_voice_sessions` and `chat_voice_reports` rows own its durable state; this -change adds no queue table, schema, provider endpoint, or independent timer. +change adds no queue table, provider endpoint, or independent timer. A concurrent +partial index supports bounded reconstruction of the existing report window. ## Trigger and deadline @@ -20,7 +21,9 @@ result does the same. Healthy active-session observations do not activate report work. Nested transactions wait for the outer commit; rollback does not publish new work. The existing work-signals protocol owns uncertain outcomes. -Startup reconstructs eligible sessions in pages. Committed hints reload their +Startup reconstructs eligible sessions in pages ordered by `(ended_at, id)`. +A partial index restricts the search to terminal sessions with provider IDs and +the current 24-hour window. The cursor retains PostgreSQL timestamp precision. Committed hints reload their session, endpoint, connection, or company scope. Each cached entry carries the saved report deadline or the first millisecond after the refresh window ends, whichever comes first. Submillisecond PostgreSQL timestamps round in the safe @@ -57,9 +60,12 @@ work. Restart reconstructs the persisted retry instead of immediately repeating the provider GET. The explicit one-shot voice reconciliation method remains for diagnostics; production maintenance does not call it periodically. -Signals cover supported writes through the owning database client in this -process. Independent SQL writers and another process need their own notification -contract. This is worker quiet, not a claim of physical host sleep. Cloud alarm +The supported deployment has one active server process writing voice session +and report state through its owning database client. Active-active servers and +out-of-band SQL writers are not supported by this worker. Process-local signals +do not reach another process; such a deployment needs a durable cross-process +wake contract before removing its reconciliation backstop. This PR neither adds +that transport nor asserts that startup reconstruction covers concurrent writers. This is worker quiet, not a claim of physical host sleep. Cloud alarm coverage and conservative sleep eligibility are unchanged; a local deadline cannot wake stopped compute. diff --git a/doc/plans/2026-10-25-voice-call-report-sleep-audit.md b/doc/plans/2026-10-25-voice-call-report-sleep-audit.md index 9fedd893a59..8de80683264 100644 --- a/doc/plans/2026-10-25-voice-call-report-sleep-audit.md +++ b/doc/plans/2026-10-25-voice-call-report-sleep-audit.md @@ -150,3 +150,45 @@ After both merges, all 138 full voice/app cases pass on a fresh migrated databas those worker implementations are unchanged by the parent repairs. Remote checks and reviews are required for the final pushed head. Parent #17580's company deletion finding remains unresolved; denial latency finding is resolved. + + +## Paperclip review: report window index and writer scope + +Review findings 4370594150 and 4370594357 identified deployment-contract clarity +and expensive report reconstruction. The supported writer model is now explicit: +one active server process owns voice writes through its DB client. Active-active +servers and independent SQL mutation are unsupported without a separate durable +wake contract. This is a documented scope restriction, not enforcement added by +this PR, and not a claim that startup recovers concurrent foreign writers. + +The query finding reproduces on fresh PostgreSQL with 20,001 old sessions and +205 eligible sessions: an actual reconstruction query filters 20,001 historical +rows. The repair adds partial `(ended_at, id)` index 0389 and uses that order and +keyset cursor. The cursor retains PostgreSQL text timestamp precision, so sessions +100 microseconds apart do not repeat or disappear across a page. All 14 report +lifecycle cases pass after the migration; the query-plan regression checks all +three pages and each stays below 1,000 rows removed by filtering. Report scope, +24-hour policy, and provider behavior are unchanged. + +The generated migration is concurrent and idempotent and registered with the +application's top-level migration runner. It applies and a second application is +a no-op. Migration numbering/safety checks pass. The generator retained exactly +five snapshots, 0385 through 0389, pruning 0384 per repository policy. + +Retained failures: the first probe reused a database with two unrelated eligible +fixture sessions and failed its exact fixture count. It is not performance proof. +A fresh database reproduced the 20,001-row filtering problem. The first migration +invocation used unsupported top-level await in tsx eval and did not apply; its +subsequent test still failed without the index. The corrected chained invocation +applied/replayed successfully and all 14 cases passed. No assertion was relaxed. +Logs: /tmp/sleep25-report-index-before.log (fixture contamination), +/tmp/sleep25-report-index-clean-before.log (cost reproduction), +/tmp/sleep25-report-index-apply.log (invalid invocation), +/tmp/sleep25-report-index-after.log (unapplied index), +/tmp/sleep25-report-index-apply-fixed.log (application/replay), +/tmp/sleep25-report-index-fixed-tests.log (14 pass). +Workspace typecheck and build pass. Fresh full voice integration and app wiring +pass all 138 cases after the index repair. Logs: +/tmp/sleep25-report-review-{types,build,acceptance}.log. +Publish and propagate through Teams #17590/#17592 after acceptance. Fresh remote +review/checks and all prior full-stack requirements remain open. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 4b598b3dc7d..72b9c1d9f2b 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -514,6 +514,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0386_omniscient_luke_cage.sql", "0387_huge_skrulls.sql", "0388_lying_trish_tilby.sql", + "0389_workable_spencer_smythe.sql", "0380_funny_adam_warlock.sql", ]); diff --git a/packages/db/src/migrations/0389_workable_spencer_smythe.sql b/packages/db/src/migrations/0389_workable_spencer_smythe.sql new file mode 100644 index 00000000000..df744394402 --- /dev/null +++ b/packages/db/src/migrations/0389_workable_spencer_smythe.sql @@ -0,0 +1 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "chat_voice_sessions_report_window_idx" ON "chat_voice_sessions" USING btree ("ended_at","id") WHERE "chat_voice_sessions"."state" in ('ended', 'failed', 'expired') and "chat_voice_sessions"."provider_session_id" is not null; \ No newline at end of file diff --git a/packages/db/src/migrations/meta/0384_snapshot.json b/packages/db/src/migrations/meta/0389_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0384_snapshot.json rename to packages/db/src/migrations/meta/0389_snapshot.json index e5bef8fce65..dadecf22e7f 100644 --- a/packages/db/src/migrations/meta/0384_snapshot.json +++ b/packages/db/src/migrations/meta/0389_snapshot.json @@ -1,6 +1,6 @@ { - "id": "45d3a35e-2d6d-4fc1-8445-c35eef1ea173", - "prevId": "75e7a97b-132d-4326-b683-003c1a2b7048", + "id": "8bd7f279-1063-4ba6-83bb-b85cba334eb3", + "prevId": "f83c3343-8665-48bc-919b-bc9fe7513b87", "version": "7", "dialect": "postgresql", "tables": { @@ -5721,6 +5721,28 @@ } }, "indexes": { + "approvals_pending_retention_idx": { + "name": "approvals_pending_retention_idx", + "columns": [ + { + "expression": "updated_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"approvals\".\"status\" = 'pending'", + "concurrently": false, + "method": "btree", + "with": {} + }, "approvals_company_status_type_idx": { "name": "approvals_company_status_type_idx", "columns": [ @@ -14138,6 +14160,24 @@ "primaryKey": false, "notNull": false }, + "provider_account_id": { + "name": "provider_account_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "cleanup_confirmed_at": { + "name": "cleanup_confirmed_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "cleanup_next_attempt_at": { + "name": "cleanup_next_attempt_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, "expires_at": { "name": "expires_at", "type": "timestamp with time zone", @@ -14160,6 +14200,22 @@ } }, "indexes": { + "chat_voice_inbound_cleanup_idx": { + "name": "chat_voice_inbound_cleanup_idx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_voice_inbound_calls\".\"cleanup_confirmed_at\" is null", + "concurrently": false, + "method": "btree", + "with": {} + }, "chat_voice_inbound_expiry_idx": { "name": "chat_voice_inbound_expiry_idx", "columns": [ @@ -14912,6 +14968,28 @@ "concurrently": false, "method": "btree", "with": {} + }, + "chat_voice_sessions_report_window_idx": { + "name": "chat_voice_sessions_report_window_idx", + "columns": [ + { + "expression": "ended_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_voice_sessions\".\"state\" in ('ended', 'failed', 'expired') and \"chat_voice_sessions\".\"provider_session_id\" is not null", + "concurrently": false, + "method": "btree", + "with": {} } }, "foreignKeys": { @@ -32079,6 +32157,28 @@ "method": "btree", "with": {} }, + "grok_dispatch_prepared_idx": { + "name": "grok_dispatch_prepared_idx", + "columns": [ + { + "expression": "created_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"grok_bot_dispatches\".\"status\" = 'prepared'", + "concurrently": false, + "method": "btree", + "with": {} + }, "grok_dispatch_company_status_idx": { "name": "grok_dispatch_company_status_idx", "columns": [ @@ -41873,6 +41973,28 @@ } }, "indexes": { + "join_requests_pending_retention_idx": { + "name": "join_requests_pending_retention_idx", + "columns": [ + { + "expression": "updated_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"join_requests\".\"status\" = 'pending_approval'", + "concurrently": false, + "method": "btree", + "with": {} + }, "join_requests_invite_unique_idx": { "name": "join_requests_invite_unique_idx", "columns": [ diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 779a28f7114..20ef0905b3b 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2710,6 +2710,13 @@ "when": 1792921047529, "tag": "0388_lying_trish_tilby", "breakpoints": true + }, + { + "idx": 389, + "version": "7", + "when": 1792926867183, + "tag": "0389_workable_spencer_smythe", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/chat_voice_sessions.ts b/packages/db/src/schema/chat_voice_sessions.ts index 6a652f97696..ac3dfcfe54e 100644 --- a/packages/db/src/schema/chat_voice_sessions.ts +++ b/packages/db/src/schema/chat_voice_sessions.ts @@ -36,6 +36,8 @@ export const chatVoiceSessions = pgTable("chat_voice_sessions", { uniqueIndex("chat_voice_sessions_request_uq").on(t.companyId, t.callerId, t.idempotencyKey), uniqueIndex("chat_voice_sessions_live_caller_task_uq").on(t.companyId, t.callerId, t.issueId).where(sql`${t.state} not in ('ended', 'failed', 'expired')`), index("chat_voice_sessions_expiry_idx").on(t.state, t.expiresAt), + index("chat_voice_sessions_report_window_idx").on(t.endedAt, t.id) + .where(sql`${t.state} in ('ended', 'failed', 'expired') and ${t.providerSessionId} is not null`), foreignKey({ columns: [t.companyId, t.endpointId], foreignColumns: [chatEndpoints.companyId, chatEndpoints.id], name: "chat_voice_sessions_endpoint_fk" }), foreignKey({ columns: [t.companyId, t.conversationId], foreignColumns: [chatConversations.companyId, chatConversations.id], name: "chat_voice_sessions_conversation_fk" }), foreignKey({ columns: [t.companyId, t.issueId], foreignColumns: [issues.companyId, issues.id], name: "chat_voice_sessions_issue_fk" }), diff --git a/server/src/__tests__/voice-call-reports.test.ts b/server/src/__tests__/voice-call-reports.test.ts index 8738049ccb8..1094e0fa815 100644 --- a/server/src/__tests__/voice-call-reports.test.ts +++ b/server/src/__tests__/voice-call-reports.test.ts @@ -195,6 +195,44 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(fetchReport).toHaveBeenCalledTimes(105); }, { timeout: 5000 }); expect((await report(broken.id)).attempts).toBe(1); await flush(); await quiet(); }); + it("reconstructs recent reports without scanning old history and preserves submillisecond page boundaries", async () => { + const seed = await session({ endedAt: new Date(Date.now() - 2 * day) }); + await db.execute(sql`insert into chat_voice_sessions + (company_id, endpoint_id, conversation_id, issue_id, assigned_agent_id, caller_id, + caller_authority, mode, state, generation, credential_fingerprint, provider_session_id, + tool_token_hash, idempotency_key, request_fingerprint, expires_at, ended_at, created_at) + select company_id, endpoint_id, conversation_id, issue_id, assigned_agent_id, gen_random_uuid()::text, + caller_authority, mode, state, generation, credential_fingerprint, gen_random_uuid()::text, + tool_token_hash, gen_random_uuid()::text, request_fingerprint, expires_at, ended_at, created_at + from chat_voice_sessions cross join generate_series(1, 20000) where id = ${seed.id}`); + const stamp = new Date(Date.now() - 60_000).toISOString(); + const expected: string[] = []; + for (let i = 0; i < 205; i++) { + const row = await session(); expected.push(row.id); + await db.execute(sql`update chat_voice_sessions set ended_at = ${stamp}::timestamptz + (${i % 3} * interval '0.0001 second') where id = ${row.id}`); + } + await db.execute(sql`analyze chat_voice_sessions`); + const queries: Array<{ text: string; parameters: unknown[] }> = []; + const original = db.$client.unsafe.bind(db.$client); + const capture = vi.spyOn(db.$client, "unsafe").mockImplementation(((text: string, parameters: unknown[] = [], ...rest: unknown[]) => { + if (text.includes('from "chat_voice_sessions"') && text.includes('left join "chat_voice_reports"')) queries.push({ text, parameters }); + return original(text, parameters as never, ...rest as []); + }) as typeof db.$client.unsafe); + let loaded: Awaited["load"]>>; + try { loaded = await voiceCallReports(db, fetchReport).load(); } + finally { capture.mockRestore(); } + expect(loaded!.map(row => row.id).sort()).toEqual(expected.sort()); + expect(queries).toHaveLength(3); + function removed(node: Record): number { + return Math.max(Number(node["Rows Removed by Filter"] ?? 0), + ...(Array.isArray(node.Plans) ? node.Plans.map(child => removed(child)) : [])); + } + for (const query of queries) { + const [row] = await db.$client.unsafe(`explain (analyze, buffers, format json) ${query.text}`, query.parameters as never); + expect(removed(row["QUERY PLAN"][0].Plan)).toBeLessThan(1000); + } + expect(fetchReport).not.toHaveBeenCalled(); + }, 30_000); it("joins an in-flight report at shutdown and resumes its saved retry", async () => { const call = await session(); let release!: (value: SpekoCallReport) => void; fetchReport.mockImplementationOnce(() => new Promise(resolve => { release = resolve; })); diff --git a/server/src/services/voice/voice-call-reports.ts b/server/src/services/voice/voice-call-reports.ts index 79a3bb8343f..fbb286e4bab 100644 --- a/server/src/services/voice/voice-call-reports.ts +++ b/server/src/services/voice/voice-call-reports.ts @@ -1,4 +1,4 @@ -import { and, eq, gt, gte, inArray, isNotNull, lte, ne, or, sql } from "drizzle-orm"; +import { and, eq, gte, inArray, isNotNull, lte, ne, or, sql } from "drizzle-orm"; import { activityLog, chatEndpoints, chatVoiceReports, chatVoiceSessions, issues, type Db } from "@paperclipai/db"; import { logger } from "../../middleware/logger.js"; import { documentService } from "../documents.js"; @@ -21,11 +21,12 @@ const eligible = (now: number) => and(inArray(chatVoiceSessions.state, ["ended", * state, not a promise of immutable provider data. Keep that finite obligation. */ export function voiceCallReports(db: Db, fetchReport: (session: VoiceSessionRow) => Promise) { async function load(scope?: VoiceRecoveryScope, signal?: AbortSignal): Promise { - const entries: Entry[] = []; let cursor: string | undefined; + const entries: Entry[] = []; let cursor: { id: string; endedAt: string } | undefined; for (;;) { signal?.throwIfAborted(); const rows = await db.select({ id: chatVoiceSessions.id, companyId: chatVoiceSessions.companyId, endpointId: chatVoiceSessions.endpointId, connectionId: chatEndpoints.connectionId, + endedAt: sql`${chatVoiceSessions.endedAt}::text`, dueAt: sql`least(coalesce(ceil(extract(epoch from ${chatVoiceReports.nextCheckAt}) * 1000), 0), floor(extract(epoch from ${chatVoiceSessions.endedAt}) * 1000) + ${WINDOW_MS} + 1)`, }).from(chatVoiceSessions) .innerJoin(chatEndpoints, and(eq(chatEndpoints.id, chatVoiceSessions.endpointId), eq(chatEndpoints.companyId, chatVoiceSessions.companyId))) @@ -35,10 +36,12 @@ export function voiceCallReports(db: Db, fetchReport: (session: VoiceSessionRow) scope?.kind === "endpoint" ? eq(chatVoiceSessions.endpointId, scope.id) : undefined, scope?.kind === "connection" ? eq(chatEndpoints.connectionId, scope.id) : undefined, scope?.kind === "company" ? eq(chatVoiceSessions.companyId, scope.id) : undefined, - cursor ? gt(chatVoiceSessions.id, cursor) : undefined)).orderBy(chatVoiceSessions.id).limit(100); - entries.push(...rows.map(row => ({ ...row, dueAt: Number(row.dueAt) }))); + cursor ? sql`(${chatVoiceSessions.endedAt}, ${chatVoiceSessions.id}) > (${cursor.endedAt}::timestamptz, ${cursor.id}::uuid)` : undefined)) + .orderBy(chatVoiceSessions.endedAt, chatVoiceSessions.id).limit(100); + entries.push(...rows.map(({ endedAt: _cursor, ...row }) => ({ ...row, dueAt: Number(row.dueAt) }))); if (rows.length < 100) return entries; - cursor = rows[rows.length - 1].id; + // Keep PostgreSQL precision; JS Date truncation can repeat a page. + cursor = rows[rows.length - 1]; } } async function projectTranscript(session: VoiceSessionRow) {