diff --git a/doc/DATABASE.md b/doc/DATABASE.md index fb0e9198acc..8a0d3a50750 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -379,6 +379,22 @@ local fixture: pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-budget-cancellation-scan.ts --runs=20000 --payload-kib=16 ``` +GitHub review lookups use four indexes from migration 0456, which builds them +concurrently. `chat_message_links_company_comment_idx` finds a source comment's +link for inherited review waits and run-origin checks. +`chat_message_links_delivery_idx` finds a delivery's links. +`chat_deliveries_github_automatic_delivery_idx` finds a delivery by its GitHub +webhook delivery ID on one endpoint. These three are partial. +`heartbeat_runs_company_wake_comment_idx` keys runs by company and +`context_snapshot.wakeCommentId`. It is not partial, because the planner does not +use expression statistics from a partial index; without them it estimates many +matches and walks the run history by creation time instead. The review check +reads the delivery's comment IDs first and then the newest same-agent run woken +by one of them, so it no longer walks the agent's run history. The two link +indexes also serve the comment and delivery `on delete set null` actions. Issue +lists share one external review-wait read per issue between the review and +blocked-inbox attention maps. + The chat wake-notice sweep compares UUID columns directly against guarded, canonical lowercase UUID payload values. Malformed and noncanonical payloads still match nothing. The query can use existing primary keys without adding a diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 10743acd93c..b6524aa1b51 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -618,6 +618,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0453_curious_ben_parker.sql", "0454_lumpy_ronan.sql", "0455_pending_budget_cancellation_index.sql", + "0456_chat_review_lookup_indexes.sql", "0384_funny_adam_warlock.sql", diff --git a/packages/db/src/migrations/0456_chat_review_lookup_indexes.sql b/packages/db/src/migrations/0456_chat_review_lookup_indexes.sql new file mode 100644 index 00000000000..4c37efeb656 --- /dev/null +++ b/packages/db/src/migrations/0456_chat_review_lookup_indexes.sql @@ -0,0 +1,4 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "chat_deliveries_github_automatic_delivery_idx" ON "chat_deliveries" USING btree ("endpoint_id",("normalized_event"->'githubAutomatic'->'context'->>'deliveryId')) WHERE ("chat_deliveries"."normalized_event"->'githubAutomatic'->'context'->>'deliveryId') is not null;--> statement-breakpoint +CREATE INDEX CONCURRENTLY IF NOT EXISTS "chat_message_links_company_comment_idx" ON "chat_message_links" USING btree ("company_id","comment_id") WHERE "chat_message_links"."comment_id" is not null;--> statement-breakpoint +CREATE INDEX CONCURRENTLY IF NOT EXISTS "chat_message_links_delivery_idx" ON "chat_message_links" USING btree ("delivery_id") WHERE "chat_message_links"."delivery_id" is not null;--> statement-breakpoint +CREATE INDEX CONCURRENTLY IF NOT EXISTS "heartbeat_runs_company_wake_comment_idx" ON "heartbeat_runs" USING btree ("company_id",("context_snapshot"->>'wakeCommentId')); \ No newline at end of file diff --git a/packages/db/src/migrations/meta/0451_snapshot.json b/packages/db/src/migrations/meta/0456_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0451_snapshot.json rename to packages/db/src/migrations/meta/0456_snapshot.json index 371419bc95d..ea96ea87ab7 100644 --- a/packages/db/src/migrations/meta/0451_snapshot.json +++ b/packages/db/src/migrations/meta/0456_snapshot.json @@ -1,6 +1,6 @@ { - "id": "b5699366-77c2-48c8-b7dc-658a75de7658", - "prevId": "b8f43f1a-3230-4cda-991d-0e7b6323114b", + "id": "a2606b41-36c2-4eb5-a59d-28cc730bc455", + "prevId": "2be53987-902a-41b6-8cb4-6e89a768f8cf", "version": "7", "dialect": "postgresql", "tables": { @@ -10955,6 +10955,28 @@ "concurrently": false, "method": "btree", "with": {} + }, + "chat_deliveries_github_automatic_delivery_idx": { + "name": "chat_deliveries_github_automatic_delivery_idx", + "columns": [ + { + "expression": "endpoint_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "(\"normalized_event\"->'githubAutomatic'->'context'->>'deliveryId')", + "asc": true, + "isExpression": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "(\"chat_deliveries\".\"normalized_event\"->'githubAutomatic'->'context'->>'deliveryId') is not null", + "concurrently": true, + "method": "btree", + "with": {} } }, "foreignKeys": { @@ -12393,6 +12415,44 @@ "concurrently": false, "method": "btree", "with": {} + }, + "chat_message_links_company_comment_idx": { + "name": "chat_message_links_company_comment_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "comment_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_message_links\".\"comment_id\" is not null", + "concurrently": true, + "method": "btree", + "with": {} + }, + "chat_message_links_delivery_idx": { + "name": "chat_message_links_delivery_idx", + "columns": [ + { + "expression": "delivery_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_message_links\".\"delivery_id\" is not null", + "concurrently": true, + "method": "btree", + "with": {} } }, "foreignKeys": { @@ -17399,7 +17459,7 @@ "columnsTo": [ "id" ], - "onDelete": "set null", + "onDelete": "cascade", "onUpdate": "no action" }, "company_secret_proposals_secret_proposal_id_company_secret_proposals_id_fk": { @@ -34552,6 +34612,34 @@ "concurrently": false, "method": "btree", "with": {} + }, + "heartbeat_run_events_retry_exhausted_idx": { + "name": "heartbeat_run_events_retry_exhausted_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": false, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"heartbeat_run_events\".\"event_type\" = 'lifecycle' and \"heartbeat_run_events\".\"message\" like 'Bounded retry exhausted%'", + "concurrently": true, + "method": "btree", + "with": {} } }, "foreignKeys": { @@ -35393,6 +35481,22 @@ "method": "btree", "with": {} }, + "heartbeat_runs_pending_budget_cancellation_idx": { + "name": "heartbeat_runs_pending_budget_cancellation_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "jsonb_typeof(\"heartbeat_runs\".\"result_json\"->'startupCancellation'->'budgetEnforcement') = 'object'\n and \"heartbeat_runs\".\"result_json\"->'startupCancellation'->'budgetEnforcement'->>'deliveredAt' is null", + "concurrently": true, + "method": "btree", + "with": {} + }, "heartbeat_runs_tree_hold_suppression_idx": { "name": "heartbeat_runs_tree_hold_suppression_idx", "columns": [ @@ -35431,6 +35535,27 @@ "method": "btree", "with": {} }, + "heartbeat_runs_company_wake_comment_idx": { + "name": "heartbeat_runs_company_wake_comment_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "(\"context_snapshot\"->>'wakeCommentId')", + "asc": true, + "isExpression": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": true, + "method": "btree", + "with": {} + }, "heartbeat_runs_cost_accounting_pending_idx": { "name": "heartbeat_runs_cost_accounting_pending_idx", "columns": [ @@ -35562,6 +35687,27 @@ "method": "btree", "with": {} }, + "heartbeat_runs_company_native_workspace_idx": { + "name": "heartbeat_runs_company_native_workspace_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "(\"runner_profile_json\"->'nativeWorkspaceSync'->>'workspaceId')", + "asc": true, + "isExpression": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": true, + "method": "btree", + "with": {} + }, "heartbeat_runs_company_issue_created_idx": { "name": "heartbeat_runs_company_issue_created_idx", "columns": [ diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index f6f36f4fd59..6da1ec71b79 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -3179,6 +3179,13 @@ "when": 1794268118742, "tag": "0455_pending_budget_cancellation_index", "breakpoints": true + }, + { + "idx": 456, + "version": "7", + "when": 1794270045155, + "tag": "0456_chat_review_lookup_indexes", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/chat_channels.ts b/packages/db/src/schema/chat_channels.ts index 06bf31d12df..3a0a147d7b8 100644 --- a/packages/db/src/schema/chat_channels.ts +++ b/packages/db/src/schema/chat_channels.ts @@ -487,6 +487,11 @@ export const chatDeliveries = pgTable( table.endpointId, table.deduplicationKey, ), + // Review checks find their delivery by the GitHub webhook delivery id. + index("chat_deliveries_github_automatic_delivery_idx") + .on(table.endpointId, sql`(${table.normalizedEvent}->'githubAutomatic'->'context'->>'deliveryId')`) + .where(sql`(${table.normalizedEvent}->'githubAutomatic'->'context'->>'deliveryId') is not null`) + .concurrently(), foreignKey({ columns: [table.companyId, table.endpointId], foreignColumns: [chatEndpoints.companyId, chatEndpoints.id], @@ -638,6 +643,16 @@ export const chatMessageLinks = pgTable( table.conversationId, table.providerMessageId, ), + // Source-comment and delivery lookups (GitHub review waits and checks). + // These also serve the comment and delivery `on delete set null` actions. + index("chat_message_links_company_comment_idx") + .on(table.companyId, table.commentId) + .where(sql`${table.commentId} is not null`) + .concurrently(), + index("chat_message_links_delivery_idx") + .on(table.deliveryId) + .where(sql`${table.deliveryId} is not null`) + .concurrently(), foreignKey({ columns: [table.companyId, table.conversationId], foreignColumns: [chatConversations.companyId, chatConversations.id], diff --git a/packages/db/src/schema/heartbeat_runs.ts b/packages/db/src/schema/heartbeat_runs.ts index aaaf4fc2e74..819aaa20a0c 100644 --- a/packages/db/src/schema/heartbeat_runs.ts +++ b/packages/db/src/schema/heartbeat_runs.ts @@ -170,6 +170,12 @@ export const heartbeatRuns = pgTable( legacyControllerLeaseIdx: index("heartbeat_runs_legacy_controller_lease_idx") .on(table.controllerLeaseExpiresAt) .where(sql`${table.runtimeMode} = 'legacy' AND ${table.status} = 'running' AND ${table.controllerBootId} is not null`), + // Runs woken by one chat comment. Not partial: the planner ignores the + // expression statistics of a partial index and would then walk the run + // history by creation time to satisfy `order by created_at limit 1`. + companyWakeCommentIdx: index("heartbeat_runs_company_wake_comment_idx") + .on(table.companyId, sql`(${table.contextSnapshot}->>'wakeCommentId')`) + .concurrently(), costAccountingPendingIdx: index("heartbeat_runs_cost_accounting_pending_idx").on(table.updatedAt, table.id).where(sql`${table.costAccountingPending} = true`), costDataQualityIdx: index("heartbeat_runs_cost_data_quality_idx") .on(table.companyId, sql`coalesce(${table.finishedAt}, ${table.createdAt}) desc`, table.id) diff --git a/scripts/benchmark-review-lookup-index-migration.ts b/scripts/benchmark-review-lookup-index-migration.ts new file mode 100644 index 00000000000..5b689dce32f --- /dev/null +++ b/scripts/benchmark-review-lookup-index-migration.ts @@ -0,0 +1,96 @@ +/** + * Times migration 0456 through the normal migration runner on a large + * synthetic database, including a rerun after the history row is removed. + * + * pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-review-lookup-index-migration.ts + * + * Optional sizes: RUNS (default 200000), RUN_CONTEXT_KIB (8), LINKS (500000), + * DELIVERIES (100000). + */ +import assert from "node:assert/strict"; +import { readFileSync } from "node:fs"; +import { applyPendingMigrations, createDb, startEmbeddedPostgresTestDatabase } from "../packages/db/src/index.js"; +import { seedGitHubReviewWait } from "../server/src/__tests__/helpers/github-review-wait.js"; + +const size = (name: string, fallback: number) => Number(process.env[name] ?? fallback); +const RUNS = size("RUNS", 200_000), RUN_CONTEXT_KIB = size("RUN_CONTEXT_KIB", 8); +const LINKS = size("LINKS", 500_000), DELIVERIES = size("DELIVERIES", 100_000); +const migrationName = "0456_chat_review_lookup_indexes"; +const migration = readFileSync(new URL(`../packages/db/src/migrations/${migrationName}.sql`, import.meta.url), "utf8"); +const indexNames = [...migration.matchAll(/IF NOT EXISTS "([a-z_]+)"/g)].map(match => match[1]!); + +const keepAlive = setInterval(() => {}, 1_000); +let temporary: Awaited> | undefined; +let db: ReturnType | undefined; +try { + temporary = await startEmbeddedPostgresTestDatabase("review-lookup-index-migration-"); + db = createDb(temporary.connectionString); + const sql = db.$client; + for (const name of indexNames) await sql.unsafe(`DROP INDEX IF EXISTS "${name}"`); + const { companyId, agentId, issueId, endpointId, conversationId } = await seedGitHubReviewWait(db); + await sql.unsafe("create extension if not exists pgcrypto"); + + const loadStart = performance.now(); + // Incompressible run contexts; half were woken by a comment. + await sql.unsafe(`insert into heartbeat_runs (company_id, agent_id, status, context_snapshot) + select $1::uuid, $2::uuid, 'succeeded', jsonb_build_object( + 'wakeCommentId', case when g % 2 = 0 then gen_random_uuid()::text end, 'issueId', $3::text, + 'prompt', (select string_agg(encode(gen_random_bytes(768), 'base64'), '') from generate_series(1, $4::int) s(i) where g > 0)) + from generate_series(1, $5::int) g`, [companyId, agentId, issueId, RUN_CONTEXT_KIB, RUNS]); + await sql.unsafe(`insert into issue_comments (company_id, issue_id, author_user_id, body) + select $1::uuid, $2::uuid, 'operator', 'load ' || g from generate_series(1, $3::int) g`, [companyId, issueId, LINKS]); + await sql.unsafe(`insert into chat_deliveries (company_id, endpoint_id, conversation_id, provider_event_id, + deduplication_key, event_kind, state, normalized_event) + select $1::uuid, $2::uuid, $3::uuid, 'load-' || g, 'load-' || g, 'message', 'processed', + jsonb_build_object('githubAutomatic', jsonb_build_object('context', jsonb_build_object( + 'deliveryId', gen_random_uuid()::text, 'body', (select string_agg(encode(gen_random_bytes(768), 'base64'), '') from generate_series(1, 2) s(i) where g > 0)))) + from generate_series(1, $4::int) g`, [companyId, endpointId, conversationId, DELIVERIES]); + await sql.unsafe(`with c as (select id, row_number() over (order by id) rn from issue_comments where company_id = $1::uuid), + d as (select id, row_number() over (order by id) rn from chat_deliveries where endpoint_id = $2::uuid) + insert into chat_message_links (company_id, endpoint_id, conversation_id, delivery_id, comment_id, + provider_message_id, direction) + select $1::uuid, $2::uuid, $3::uuid, d.id, c.id, 'load-' || c.rn, + case when c.rn % 2 = 0 then 'inbound' else 'outbound' end + from c left join d on d.rn = c.rn`, [companyId, endpointId, conversationId]); + await sql.unsafe("vacuum analyze"); + const loadMs = performance.now() - loadStart; + const tableSizes = await sql.unsafe(`select relname as table, n_live_tup::bigint::text as rows, + pg_size_pretty(pg_total_relation_size(relid)) as total_size from pg_stat_user_tables + where relname in ('chat_message_links', 'chat_deliveries', 'heartbeat_runs') order by relname`); + + async function migrate() { + await sql.unsafe(`delete from drizzle.__drizzle_migrations where hash in ( + select hash from drizzle.__drizzle_migrations order by created_at desc limit 1)`); + const start = performance.now(); + await applyPendingMigrations(temporary!.connectionString); + return performance.now() - start; + } + // First run builds every index; the rerun finds them valid and skips them. + const firstRunMs = await migrate(); + const valid = await sql.unsafe(`select c.relname, i.indisvalid, pg_size_pretty(pg_relation_size(c.oid)) as size + from pg_index i join pg_class c on c.oid = i.indexrelid where c.relname in (${indexNames.map((_, n) => `$${n + 1}`).join(", ")}) + order by c.relname`, indexNames); + assert.equal(valid.length, indexNames.length); + assert.ok(valid.every(row => row.indisvalid)); + const rerunMs = await migrate(); + // Per-index build time, measured separately with the same statements. + const builds: Array<{ index: string; ms: number }> = []; + for (const statement of migration.split("--> statement-breakpoint").map(part => part.trim()).filter(Boolean)) { + const name = statement.match(/IF NOT EXISTS "([a-z_]+)"/)![1]!; + await sql.unsafe(`DROP INDEX CONCURRENTLY "${name}"`); + const start = performance.now(); + await sql.unsafe(statement); + builds.push({ index: name, ms: Math.round(performance.now() - start) }); + } + console.log(JSON.stringify({ + postgresVersion: (await sql.unsafe("select version()"))[0]!.version, + fixture: { synthetic: true, runs: RUNS, runContextKiB: RUN_CONTEXT_KIB, links: LINKS, deliveries: DELIVERIES, + loadMs: Math.round(loadMs), tableSizes }, + migration: { firstRunMs: Math.round(firstRunMs), rerunMs: Math.round(rerunMs), indexes: valid, builds }, + }, null, 2)); +} finally { + try { + await db?.$client.end({ timeout: 0 }); + await temporary?.cleanup(); + } finally { clearInterval(keepAlive); } +} diff --git a/scripts/benchmark-review-lookup-indexes.ts b/scripts/benchmark-review-lookup-indexes.ts new file mode 100644 index 00000000000..1e3a3bd9a06 --- /dev/null +++ b/scripts/benchmark-review-lookup-indexes.ts @@ -0,0 +1,234 @@ +/** + * Local proof for the GitHub review lookups (migration 0456). + * + * Measures the SQL that the services generate: the inherited review-wait lookup + * by source comment (SQL unchanged; four new indexes) and the review-check + * fallback that finds the wake run from the delivery's inbound comments (SQL + * split into two statements). The pre-fix check module is loaded from + * --baseline-ref. Each version runs with and without the new indexes, and the + * script fails unless complete rows and service results match. + * + * pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-review-lookup-indexes.ts --baseline-ref= + * + * Optional sizes: LINKS (default 50000), DELIVERIES (5000), RUNS (4000), + * RUN_CONTEXT_KIB (12). + */ +import assert from "node:assert/strict"; +import { execFileSync } from "node:child_process"; +import { promises as fs, readFileSync } from "node:fs"; +import path from "node:path"; +import { fileURLToPath, pathToFileURL } from "node:url"; +import { createDb, startEmbeddedPostgresTestDatabase } from "../packages/db/src/index.js"; +import { readInheritedGitHubReviewWait } from "../server/src/services/chat-channels/github/review-wait.js"; +import { githubReviewCheckState } from "../server/src/services/chat-channels/github/review-check-state.js"; +import { captureReviewWaitQueries, seedGitHubReviewCheck, seedGitHubReviewWait, + type CapturedReviewWaitQuery } from "../server/src/__tests__/helpers/github-review-wait.js"; + +const size = (name: string, fallback: number) => Number(process.env[name] ?? fallback); +const LINKS = size("LINKS", 50_000), DELIVERIES = size("DELIVERIES", 5_000); +const RUNS = size("RUNS", 4_000), RUN_CONTEXT_KIB = size("RUN_CONTEXT_KIB", 12); +const baselineRef = process.argv.find(arg => arg.startsWith("--baseline-ref="))?.slice(15); +assert.ok(baselineRef, "Pass an explicit pre-fix --baseline-ref="); +const git = (...args: string[]) => execFileSync("git", args, { encoding: "utf8" }); +const baselineRevision = git("rev-parse", "--verify", `${baselineRef}^{commit}`).trim(); +const root = fileURLToPath(new URL("../", import.meta.url)); +const checkModule = "server/src/services/chat-channels/github/review-check-state.ts"; +const waitModule = "server/src/services/chat-channels/github/review-wait.ts"; +assert.equal(git("show", `${baselineRevision}:${waitModule}`), readFileSync(path.join(root, waitModule), "utf8"), + "The review-wait module must be unchanged; only its indexes differ"); +const migration = readFileSync(path.join(root, "packages/db/src/migrations/0456_chat_review_lookup_indexes.sql"), "utf8"); +const indexStatements = migration.split("--> statement-breakpoint").map(statement => statement.trim()).filter(Boolean); +const indexNames = indexStatements.map(statement => statement.match(/IF NOT EXISTS "([a-z_]+)"/)![1]!); + +type CheckState = typeof githubReviewCheckState; +const keepAlive = setInterval(() => {}, 1_000); +let temporary: Awaited> | undefined; +let db: ReturnType | undefined; +let baselineDir: string | undefined; +try { + const moduleUrl = pathToFileURL(path.join(root, checkModule)); + baselineDir = await fs.mkdtemp(path.join(root, "server/scripts/.review-check-baseline-")); + const baselineFile = path.join(baselineDir, "review-check-state.ts"); + await fs.writeFile(baselineFile, git("show", `${baselineRevision}:${checkModule}`).replace(/from "(\.{1,2}\/[^"]+)"/g, + (_match, specifier: string) => `from ${JSON.stringify(fileURLToPath(new URL(specifier, moduleUrl)))}`)); + const baselineCheck = (await import(pathToFileURL(baselineFile).href) as { githubReviewCheckState: CheckState }).githubReviewCheckState; + + temporary = await startEmbeddedPostgresTestDatabase("review-lookups-benchmark-"); + db = createDb(temporary.connectionString); + const sql = db.$client; + const wait = await seedGitHubReviewWait(db); + const check = await seedGitHubReviewCheck(db); + + // Unrelated chat history: comments, deliveries and links on the same endpoint. + const loadStart = performance.now(); + await sql.unsafe(`insert into issue_comments (company_id, issue_id, author_user_id, body) + select $1::uuid, $2::uuid, 'operator', 'noise ' || g from generate_series(1, $3::int) g`, + [check.companyId, check.issueId, LINKS]); + await sql.unsafe(`insert into chat_deliveries (company_id, endpoint_id, conversation_id, provider_event_id, + deduplication_key, event_kind, state, normalized_event) + select $1::uuid, $2::uuid, $3::uuid, 'noise-' || g, 'noise-' || g, 'message', 'processed', + jsonb_build_object('githubAutomatic', jsonb_build_object('context', jsonb_build_object( + 'deliveryId', gen_random_uuid()::text, + 'body', (select string_agg(md5(random()::text || s.i || g), '') from generate_series(1, 64) s(i))))) + from generate_series(1, $4::int) g`, [check.companyId, check.endpointId, check.conversationId, DELIVERIES]); + await sql.unsafe(`with c as (select id, row_number() over (order by id) rn from issue_comments + where company_id = $1::uuid and body like 'noise %'), + d as (select id, row_number() over (order by id) rn from chat_deliveries + where endpoint_id = $2::uuid and provider_event_id like 'noise-%') + insert into chat_message_links (company_id, endpoint_id, conversation_id, delivery_id, comment_id, + provider_message_id, direction) + select $1::uuid, $2::uuid, $3::uuid, d.id, c.id, 'noise-' || c.rn, + case when c.rn % 2 = 0 then 'inbound' else 'outbound' end + from c left join d on d.rn = c.rn`, [check.companyId, check.endpointId, check.conversationId]); + // Runs of the same agent before and after the wake run; most were woken by + // some other comment. + await sql.unsafe(`insert into heartbeat_runs (company_id, agent_id, status, context_snapshot, created_at) + select $1::uuid, $2::uuid, 'succeeded', jsonb_build_object( + 'wakeCommentId', case when g % 5 <> 0 then gen_random_uuid()::text end, + 'issueId', $3::text, + 'prompt', (select string_agg(md5(random()::text || s.i || g), '') from generate_series(1, $4::int) s(i))), + now() + make_interval(secs => g - $5::int / 2) + from generate_series(1, $5::int) g`, [check.companyId, check.agentId, check.issueId, RUN_CONTEXT_KIB * 32, RUNS]); + const loadMs = performance.now() - loadStart; + const tableSizes = await sql.unsafe(`select relname as table, n_live_tup::int as rows, + pg_size_pretty(pg_total_relation_size(relid)) as total_size from pg_stat_user_tables + where relname in ('chat_message_links', 'chat_deliveries', 'heartbeat_runs', 'issue_comments') order by relname`); + // A delivery whose inbound comment has no wake run yet. + const [unmatched] = await sql.unsafe(`select d.normalized_event->'githubAutomatic'->'context'->>'deliveryId' as id + from chat_deliveries d join chat_message_links l on l.delivery_id = d.id and l.direction = 'inbound' + where d.endpoint_id = $1::uuid and d.provider_event_id like 'noise-%' order by d.provider_event_id limit 1`, [check.endpointId]); + assert.ok(unmatched?.id); + const unmatchedAction = { ...check.action, payload: { ...check.action.payload, + event: { ...check.event, deliveryId: unmatched.id as string } } }; + + const isWakeRead = (query: CapturedReviewWaitQuery) => query.sql.startsWith('select "chat_message_links"."comment_id"') || + query.sql.includes('from "chat_deliveries" inner join "chat_message_links"') || + (query.sql.startsWith('select "id", "status" from "heartbeat_runs"') && query.sql.includes("wakeCommentId")); + const rows = async (query: CapturedReviewWaitQuery) => + (await sql.unsafe(query.sql, query.params as never[]).values()).map(row => row.map(value => value?.toString() ?? null)); + async function captureCheck(reader: CheckState, action: typeof check.action) { + const queries: CapturedReviewWaitQuery[] = []; + const result = await reader(captureReviewWaitQueries(db!, queries), action); + return { result, wakeReads: queries.filter(isWakeRead) }; + } + async function captureWait() { + const queries: CapturedReviewWaitQuery[] = []; + const result = await readInheritedGitHubReviewWait(captureReviewWaitQueries(db!, queries), wait.run); + const inherited = queries.find(query => query.sql.includes('join "chat_github_reviews"')); + assert.ok(inherited); + return { result, inherited }; + } + const median = (values: number[]) => [...values].sort((a, b) => a - b)[Math.floor(values.length / 2)]!; + const round = (value: number) => Number(value.toFixed(3)); + async function sample(run: () => Promise) { + for (let n = 0; n < 2; n++) await run(); + const samples: number[] = []; + for (let n = 0; n < 9; n++) samples.push(await run()); + return { samplesMs: samples.map(round), medianMs: round(median(samples)), + minMs: round(Math.min(...samples)), maxMs: round(Math.max(...samples)) }; + } + // Server execution time from EXPLAIN ANALYZE, summed over a caller's statements. + async function explain(query: CapturedReviewWaitQuery) { + const [result] = await sql.unsafe(`EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${query.sql}`, query.params as never[]); + return (result!["QUERY PLAN"] as Array<{ Plan: Record; "Execution Time": number }>)[0]!; + } + const executionMs = async (queries: CapturedReviewWaitQuery[]) => { + let total = 0; + for (const query of queries) total += (await explain(query))["Execution Time"]; + return total; + }; + const wall = (run: () => Promise) => async () => { + const start = performance.now(); + await run(); + return performance.now() - start; + }; + async function plans(queries: CapturedReviewWaitQuery[]) { + return Promise.all(queries.map(async query => { + const root = await explain(query), nodes: string[] = []; + const walk = (node: Record) => { + if (/Scan|Join|Loop/.test(String(node["Node Type"]))) nodes.push(`${node["Node Type"]}${node["Index Name"] ? ` using ${node["Index Name"]}` : ""}${node["Relation Name"] ? ` on ${node["Relation Name"]}` : ""}`); + for (const child of (node.Plans as Array> | undefined) ?? []) walk(child); + }; + walk(root.Plan); + return { executionMs: root["Execution Time"], sharedHitBlocks: root.Plan["Shared Hit Blocks"], + sharedReadBlocks: root.Plan["Shared Read Blocks"], nodes }; + })); + } + + async function measure(indexes: boolean) { + const builds: Array<{ index: string; ms: number }> = []; + if (!indexes) { + for (const name of indexNames) await sql.unsafe(`DROP INDEX CONCURRENTLY IF EXISTS "${name}"`); + } else { + for (const [n, statement] of indexStatements.entries()) { + const start = performance.now(); + await sql.unsafe(statement); + builds.push({ index: indexNames[n]!, ms: round(performance.now() - start) }); + } + } + await sql.unsafe("ANALYZE"); + const waitCapture = await captureWait(); + const versions = { baseline: baselineCheck, treatment: githubReviewCheckState }; + const checks: Record = {}; + const outputs: Record = {}; + for (const [version, reader] of Object.entries(versions)) { + for (const [name, action] of [["matched", check.action], ["noWakeRunYet", unmatchedAction]] as const) { + const captured = await captureCheck(reader, action); + const finalRows = captured.wakeReads.length ? await rows(captured.wakeReads.at(-1)!) : []; + outputs[`${version}.${name}`] = { result: captured.result, finalRows }; + checks[`${version}.${name}`] = { + statements: captured.wakeReads.length, + serverExecution: await sample(() => executionMs(captured.wakeReads)), + serviceWall: await sample(wall(() => reader(db!, action))), + plans: await plans(captured.wakeReads), + }; + } + } + return { + indexes, builds, waitCapture, outputs, + inheritedRows: await rows(waitCapture.inherited), + inherited: { + serverExecution: await sample(() => executionMs([waitCapture.inherited])), + serviceWall: await sample(wall(() => readInheritedGitHubReviewWait(db!, wait.run))), + plans: await plans([waitCapture.inherited]), + }, + checks, + }; + } + + // No indexes, indexes, then no indexes again to expose cache-order bias. + const runs = [await measure(false), await measure(true), await measure(false)]; + const reference = runs[0]!; + for (const run of runs) { + assert.equal(run.waitCapture.inherited.sql, reference.waitCapture.inherited.sql); + assert.deepEqual(run.inheritedRows, reference.inheritedRows, "Inherited review rows must match"); + assert.deepEqual(run.waitCapture.result, reference.waitCapture.result); + for (const name of ["matched", "noWakeRunYet"]) { + for (const version of ["baseline", "treatment"]) { + assert.deepEqual(run.outputs[`${version}.${name}`], reference.outputs[`baseline.${name}`], + `${version} ${name} result and final wake-run rows must match the pre-fix module`); + } + } + } + assert.equal(reference.waitCapture.result?.kind, "published"); + assert.deepEqual((reference.outputs["baseline.matched"] as { finalRows: unknown }).finalRows, [[check.wakeRunId, "running"]]); + assert.deepEqual((reference.outputs["baseline.noWakeRunYet"] as { finalRows: unknown }).finalRows, []); + + console.log(JSON.stringify({ + baselineRevision, treatmentRevision: git("rev-parse", "HEAD").trim(), + treatmentDiff: execFileSync("git", ["diff", "HEAD", "--", checkModule], { cwd: root, encoding: "utf8" }), + postgresVersion: (await sql.unsafe("select version()"))[0]!.version, + settings: await sql.unsafe("select name, setting from pg_settings where name in ('shared_buffers','work_mem','jit') order by name"), + fixture: { synthetic: true, links: LINKS, deliveries: DELIVERIES, runs: RUNS, runContextKiB: RUN_CONTEXT_KIB, + loadMs: Math.round(loadMs), tableSizes }, + correctness: { inheritedSqlUnchanged: true, inheritedRowsEqual: true, checkResultsEqual: true, finalWakeRunRowsEqual: true, + cases: ["matched", "noWakeRunYet"] }, + results: runs.map(({ waitCapture: _capture, outputs: _outputs, inheritedRows: _rows, ...run }) => run), + }, null, 2)); +} finally { + try { + await db?.$client.end({ timeout: 0 }); + await temporary?.cleanup(); + if (baselineDir) await fs.rm(baselineDir, { recursive: true, force: true }); + } finally { clearInterval(keepAlive); } +} diff --git a/server/src/__tests__/github-review-check-state.test.ts b/server/src/__tests__/github-review-check-state.test.ts new file mode 100644 index 00000000000..4950ecad9ce --- /dev/null +++ b/server/src/__tests__/github-review-check-state.test.ts @@ -0,0 +1,93 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { agents, chatMessageLinks, createDb, heartbeatRuns, type Db } from "@paperclipai/db"; +import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { captureReviewWaitQueries, seedGitHubReviewCheck, seedGitHubReviewWait, + type CapturedReviewWaitQuery } from "./helpers/github-review-wait.js"; +import { githubReviewCheckState } from "../services/chat-channels/github/review-check-state.js"; +import { readInheritedGitHubReviewWait } from "../services/chat-channels/github/review-wait.js"; + +describe("GitHub review lookups by source comment and delivery", () => { + let temporary: Awaited>; + let db: Db; + beforeAll(async () => { + if (process.env.PAPERCLIP_TEST_DATABASE_URL) { + db = createDb(process.env.PAPERCLIP_TEST_DATABASE_URL); + } else { + temporary = await startEmbeddedPostgresTestDatabase("github-review-check-state-"); + db = createDb(temporary.connectionString); + } + }, 30_000); + afterAll(async () => { await db?.$client.end({ timeout: 0 }); await temporary?.cleanup(); }); + + it("finds only the newest same-agent run woken by the delivery's comment", async () => { + const f = await seedGitHubReviewCheck(db); + const other = await seedGitHubReviewWait(db); + const otherAgentId = randomUUID(), at = (seconds: number) => new Date(Date.now() + seconds * 1_000); + await db.insert(agents).values({ id: otherAgentId, companyId: f.companyId, name: "Other", adapterType: "process" }); + // A delivery link without a comment cannot select a run. + await db.insert(chatMessageLinks).values({ companyId: f.companyId, endpointId: f.endpointId, + conversationId: f.conversationId, deliveryId: f.chatDeliveryId, providerMessageId: "outbound-reply", direction: "outbound" }); + await db.insert(heartbeatRuns).values([ + // Same comment, but another agent or company: never the review's run. + { companyId: f.companyId, agentId: otherAgentId, status: "succeeded", createdAt: at(60), + contextSnapshot: { wakeCommentId: f.commentId } }, + { companyId: other.companyId, agentId: other.agentId, status: "succeeded", createdAt: at(60), + contextSnapshot: { wakeCommentId: f.commentId } }, + // Text comparison: noncanonical, non-string and list-only values do not match. + { companyId: f.companyId, agentId: f.agentId, status: "failed", createdAt: at(60), + contextSnapshot: { wakeCommentId: f.commentId.toUpperCase() } }, + { companyId: f.companyId, agentId: f.agentId, status: "failed", createdAt: at(60), + contextSnapshot: { wakeCommentId: [f.commentId] } }, + { companyId: f.companyId, agentId: f.agentId, status: "failed", createdAt: at(60), + contextSnapshot: { wakeCommentIds: [f.commentId] } }, + { companyId: f.companyId, agentId: f.agentId, status: "failed", createdAt: at(60), contextSnapshot: null }, + { companyId: f.companyId, agentId: f.agentId, status: "succeeded", createdAt: at(-60), + contextSnapshot: { wakeCommentId: f.commentId } }, + ]); + expect(await githubReviewCheckState(db, f.action)).toMatchObject({ + disposition: "publish", run: { id: f.wakeRunId, status: "running" }, state: "running", status: "in_progress", + }); + + const [newest] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, + status: "succeeded", createdAt: at(120), contextSnapshot: { wakeCommentId: f.commentId } }).returning(); + expect(await githubReviewCheckState(db, f.action)).toMatchObject({ + run: { id: newest!.id, status: "succeeded" }, state: "incomplete", status: "completed", + }); + + // Another delivery id on the same endpoint selects no run. + const unmatched = { ...f.action, payload: { ...f.action.payload, + event: { ...f.event, deliveryId: randomUUID() } } }; + expect(await githubReviewCheckState(db, unmatched)).toMatchObject({ run: undefined, state: "queued" }); + }); + + it("reads source comments, deliveries and wake runs through their indexes", async () => { + const f = await seedGitHubReviewCheck(db); + const wait = await seedGitHubReviewWait(db); + // A run history in which each run was woken by another comment. With its + // statistics, the planner must prefer the wake-comment index to walking + // the history by creation time. + await db.$client.unsafe(`insert into heartbeat_runs (company_id, agent_id, status, context_snapshot, created_at) + select $1::uuid, $2::uuid, 'succeeded', jsonb_build_object('wakeCommentId', gen_random_uuid()::text), + now() + make_interval(secs => g) from generate_series(1, 2000) g`, [f.companyId, f.agentId]); + await db.$client.unsafe("analyze heartbeat_runs"); + const queries: CapturedReviewWaitQuery[] = []; + await githubReviewCheckState(captureReviewWaitQueries(db, queries), f.action); + await readInheritedGitHubReviewWait(captureReviewWaitQueries(db, queries), wait.run); + const comments = queries.find(query => query.sql.includes('from "chat_deliveries" inner join "chat_message_links"')); + const wakeRun = queries.find(query => query.sql.includes('from "heartbeat_runs"') && query.sql.includes("wakeCommentId")); + const inherited = queries.find(query => query.sql.includes('join "chat_github_reviews"')); + expect(comments && wakeRun && inherited).toBeTruthy(); + const plan = (query: CapturedReviewWaitQuery) => db.$client.begin(async sql => { + // Small test tables favour sequential scans; disable them to prove the + // generated predicates can use the partial indexes at all. + await sql.unsafe("set local enable_seqscan = off"); + return JSON.stringify(await sql.unsafe(`explain (format json) ${query.sql}`, query.params as never[])); + }); + const commentsPlan = await plan(comments!), wakeRunPlan = await plan(wakeRun!), inheritedPlan = await plan(inherited!); + expect(commentsPlan).toContain("chat_deliveries_github_automatic_delivery_idx"); + expect(commentsPlan).toContain("chat_message_links_delivery_idx"); + expect(wakeRunPlan).toContain("heartbeat_runs_company_wake_comment_idx"); + expect(inheritedPlan).toContain("chat_message_links_company_comment_idx"); + }); +}); diff --git a/server/src/__tests__/helpers/github-review-wait.ts b/server/src/__tests__/helpers/github-review-wait.ts index eb3bea3fcef..ecdcbe0075f 100644 --- a/server/src/__tests__/helpers/github-review-wait.ts +++ b/server/src/__tests__/helpers/github-review-wait.ts @@ -1,7 +1,7 @@ import { createHash, randomUUID } from "node:crypto"; import { and, eq, fillPlaceholders } from "drizzle-orm"; -import { agents, chatActions, chatConversations, chatEndpointResources, chatEndpoints, - chatGitHubReviews, chatMessageLinks, companies, heartbeatRuns, issueComments, issues, +import { agents, chatActions, chatConversations, chatDeliveries, chatEndpointResources, chatEndpoints, + chatGitHubConfigurations, chatGitHubReviews, chatMessageLinks, companies, heartbeatRuns, issueComments, issues, runIdentityContexts, toolApplications, toolConnections, type Db } from "@paperclipai/db"; import { defaultGitHubReviewPolicy } from "@paperclipai/shared"; @@ -79,6 +79,33 @@ export function captureReviewWaitQueries(db: Db, queries: CapturedReviewWaitQuer }}); } +/** Extend the review-wait fixture to a check action whose review has no run + * yet, so the check state must find the wake run from the inbound comment. */ +export async function seedGitHubReviewCheck(db: Db) { + const f = await seedGitHubReviewWait(db); + const githubDeliveryId = randomUUID(), chatDeliveryId = randomUUID(), wakeRunId = randomUUID(); + const event = { deliveryId: githubDeliveryId, repository: "synthetic/repository", repositoryId: "42", + pullNumber: 7, headSha: f.headSha }; + await db.update(chatEndpoints).set({ status: "active" }).where(eq(chatEndpoints.id, f.endpointId)); + await db.update(chatEndpointResources).set({ enabled: true, availability: "available" }) + .where(eq(chatEndpointResources.id, f.resourceId)); + await db.insert(chatGitHubConfigurations).values({ endpointId: f.endpointId, companyId: f.companyId, + configuration: { toolsEnabled: true } as never, updatedByUserId: "operator" }); + await db.insert(chatDeliveries).values({ id: chatDeliveryId, companyId: f.companyId, endpointId: f.endpointId, + conversationId: f.conversationId, providerEventId: githubDeliveryId, deduplicationKey: githubDeliveryId, + eventKind: "message", state: "processed", normalizedEvent: { githubAutomatic: { context: event } } }); + await db.update(chatMessageLinks).set({ deliveryId: chatDeliveryId }) + .where(and(eq(chatMessageLinks.companyId, f.companyId), eq(chatMessageLinks.commentId, f.commentId))); + await db.insert(heartbeatRuns).values({ id: wakeRunId, companyId: f.companyId, agentId: f.agentId, + nativeIssueId: f.issueId, status: "running", contextSnapshot: { wakeCommentId: f.commentId } }); + await db.update(chatGitHubReviews).set({ runId: null, assessment: null, state: "queued" }) + .where(eq(chatGitHubReviews.id, f.reviewId)); + const [action] = await db.insert(chatActions).values({ companyId: f.companyId, endpointId: f.endpointId, + kind: "github_review_check", providerActionId: `github_review_check:${githubDeliveryId}`, + payload: { event, admitted: true, runtimeGeneration: 0 } }).returning(); + return { ...f, action: action!, event, githubDeliveryId, chatDeliveryId, wakeRunId }; +} + export async function revokeReviewWaitConnection(db: Db, fixture: Awaited>) { await db.update(toolConnections).set({ enabled: false }).where(and( eq(toolConnections.id, fixture.connectionId), eq(toolConnections.companyId, fixture.companyId))); diff --git a/server/src/services/chat-channels/github/review-check-state.ts b/server/src/services/chat-channels/github/review-check-state.ts index 2254a82feb0..8c92c9ef421 100644 --- a/server/src/services/chat-channels/github/review-check-state.ts +++ b/server/src/services/chat-channels/github/review-check-state.ts @@ -1,4 +1,4 @@ -import { and, desc, eq, sql } from "drizzle-orm"; +import { and, desc, eq, inArray, isNotNull, sql } from "drizzle-orm"; import { chatActions, chatDeliveries, chatEndpoints, chatEndpointResources, chatGitHubConfigurations, chatGitHubReviews, chatMessageLinks, companies, heartbeatRuns, type Db } from "@paperclipai/db"; @@ -66,30 +66,37 @@ export async function githubReviewCheckState(db: Db, action: typeof chatActions. )[0] : undefined; if (review && !run) { - const [sourceRun] = await db - .select({ run: { id: heartbeatRuns.id, status: heartbeatRuns.status } }) + // Read the delivery's few comments first. A join lets the planner walk + // the agent's whole run history by creation time; a known comment list + // uses the wake-comment index, and no comment means no run to find. + const commentIds = [...new Set((await db + .select({ commentId: chatMessageLinks.commentId }) .from(chatDeliveries) .innerJoin( chatMessageLinks, eq(chatMessageLinks.deliveryId, chatDeliveries.id), ) - .innerJoin( - heartbeatRuns, - and( - eq(heartbeatRuns.companyId, action.companyId), - eq(heartbeatRuns.agentId, source.endpoint.assignedAgentId), - sql`${heartbeatRuns.contextSnapshot}->>'wakeCommentId' = ${chatMessageLinks.commentId}::text`, - ), - ) .where( and( eq(chatDeliveries.endpointId, action.endpointId), sql`${chatDeliveries.normalizedEvent}->'githubAutomatic'->'context'->>'deliveryId' = ${event.deliveryId}`, + isNotNull(chatMessageLinks.commentId), ), - ) - .orderBy(desc(heartbeatRuns.createdAt)) - .limit(1); - run = sourceRun?.run; + )).map(link => link.commentId!))]; + if (commentIds.length) { + [run] = await db + .select({ id: heartbeatRuns.id, status: heartbeatRuns.status }) + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.companyId, action.companyId), + eq(heartbeatRuns.agentId, source.endpoint.assignedAgentId), + inArray(sql`${heartbeatRuns.contextSnapshot}->>'wakeCommentId'`, commentIds), + ), + ) + .orderBy(desc(heartbeatRuns.createdAt)) + .limit(1); + } } const [delivery] = await db .select({ state: chatDeliveries.state }) diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 3ad5ebad329..bf009635f5b 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -69,7 +69,7 @@ import { createHash, randomUUID } from "node:crypto"; import { isExplicitContinuationRetryClaim } from "./explicit-continuation-retry-claim.js"; import { nativeManualAdmissionHoldsExecution } from "./native-runtime/native-runner-ownership.js"; import { markdownToPlainText, parseMarkdown } from "chat"; -import { listExternalReviewWaits } from "./native-runtime/external-review-wait.js"; +import { createExternalReviewWaitReader, type ExternalReviewWaitReader } from "./native-runtime/external-review-wait.js"; import { and, asc, @@ -4686,6 +4686,7 @@ async function listIssueReviewAttentionMap( dbOrTx: any, companyId: string, issueRows: IssueReviewAttentionInput[], + readExternalReviewWaits: ExternalReviewWaitReader = createExternalReviewWaitReader(dbOrTx, companyId), ): Promise> { const result = new Map(); for (const row of issueRows) result.set(row.id, reviewAttentionNone()); @@ -4929,7 +4930,7 @@ async function listIssueReviewAttentionMap( queuedWakeRequests: wakeRows, pendingInteractions: interactionRows, pendingApprovals: approvalRows, - externalResponseWaits: await listExternalReviewWaits(dbOrTx, companyId, reviewIds), + externalResponseWaits: await readExternalReviewWaits(reviewIds), openRecoveryIssues: recoveryPaths, now: new Date(), }; @@ -5777,6 +5778,7 @@ async function listIssueBlockedInboxAttentionMap( dbOrTx: any, companyId: string, issueRows: BlockedInboxIssueRow[], + readExternalReviewWaits: ExternalReviewWaitReader = createExternalReviewWaitReader(dbOrTx, companyId), ): Promise> { const rowIssueIds = [...new Set(issueRows.map((row) => row.id))]; const result = new Map(); @@ -6036,7 +6038,7 @@ async function listIssueBlockedInboxAttentionMap( }); const findings = classifyIssueGraphLiveness({ - externalResponseWaits: await listExternalReviewWaits(dbOrTx, companyId, graphIssues.filter(issue => issue.status === "in_review").map(issue => issue.id)), + externalResponseWaits: await readExternalReviewWaits(graphIssues.filter(issue => issue.status === "in_review").map(issue => issue.id)), issues: graphIssues.map((issue) => ({ id: issue.id, companyId: issue.companyId, @@ -6637,6 +6639,8 @@ async function listBlockedInboxIssues( const issueIds = withRuns.map((row) => row.id); const includeLiveDescendantSummary = filters?.includeLiveDescendantSummary === true; + // Both attention maps read external review waits for overlapping issues. + const readExternalReviewWaits = createExternalReviewWaitReader(dbOrTx, companyId); const [ statsRows, readRows, @@ -6656,8 +6660,8 @@ async function listBlockedInboxIssues( lastActivityStatsForIssues(dbOrTx, companyId, issueIds), blockedByMapForIssues(dbOrTx, companyId, issueIds), listIssueBlockerAttentionMap(dbOrTx, companyId, withRuns), - listIssueReviewAttentionMap(dbOrTx, companyId, withRuns), - listIssueBlockedInboxAttentionMap(dbOrTx, companyId, withRuns), + listIssueReviewAttentionMap(dbOrTx, companyId, withRuns, readExternalReviewWaits), + listIssueBlockedInboxAttentionMap(dbOrTx, companyId, withRuns, readExternalReviewWaits), includeLiveDescendantSummary ? liveDescendantCountMapForIssues(dbOrTx, companyId, issueIds) : Promise.resolve(new Map()), @@ -8601,6 +8605,8 @@ export function issueService(db: Db) { const archiveByIssueId = new Map( archiveRows.map((row) => [row.issueId, row]), ); + // Both attention maps read external review waits for overlapping issues. + const readExternalReviewWaits = createExternalReviewWaitReader(db, companyId); const [ blockerAttentionByIssueId, reviewAttentionByIssueId, @@ -8609,9 +8615,9 @@ export function issueService(db: Db) { ? [new Map(), new Map(), new Map()] : await timePhase("attention", () => Promise.all([ filters?.listProjection === "index" ? Promise.resolve(new Map()) : listIssueBlockerAttentionMap(db, companyId, withRuns), - filters?.listProjection === "index" ? Promise.resolve(new Map()) : listIssueReviewAttentionMap(db, companyId, withRuns), + filters?.listProjection === "index" ? Promise.resolve(new Map()) : listIssueReviewAttentionMap(db, companyId, withRuns, readExternalReviewWaits), includeBlockedInboxAttention - ? listIssueBlockedInboxAttentionMap(db, companyId, withRuns) + ? listIssueBlockedInboxAttentionMap(db, companyId, withRuns, readExternalReviewWaits) : Promise.resolve(new Map()), ])); diff --git a/server/src/services/native-runtime/external-review-wait.test.ts b/server/src/services/native-runtime/external-review-wait.test.ts new file mode 100644 index 00000000000..ed1c65229db --- /dev/null +++ b/server/src/services/native-runtime/external-review-wait.test.ts @@ -0,0 +1,32 @@ +import { describe, expect, it, vi } from "vitest"; +import type { Db } from "@paperclipai/db"; +import { createExternalReviewWaitReader, type ExternalReviewWait } from "./external-review-wait.js"; + +const db = {} as Db; +const wait = (issueId: string) => ({ issueId, companyId: "company" }) as ExternalReviewWait; + +describe("createExternalReviewWaitReader", () => { + it("reads each issue once for callers that share one request", async () => { + const list = vi.fn(async (_db: Db, _companyId: string, issueIds: string[]) => + issueIds.filter(issueId => issueId !== "b").map(wait)); + const read = createExternalReviewWaitReader(db, "company", list); + // The review and blocked-inbox maps start their reads concurrently. + const [page, graph] = await Promise.all([read(["a", "b"]), read(["c", "a", "b", "c"])]); + expect(page).toEqual([wait("a")]); + expect(graph).toEqual([wait("c"), wait("a")]); + expect(await read(["a"])).toEqual([wait("a")]); + expect(await read([])).toEqual([]); + expect(list.mock.calls.map(([, companyId, issueIds]) => [companyId, issueIds])).toEqual([ + ["company", ["a", "b"]], ["company", ["c"]], + ]); + expect(list.mock.calls.every(([database]) => database === db)).toBe(true); + }); + + it("rejects every caller that waits on a failed read", async () => { + const failure = new Error("database unavailable"); + const list = vi.fn(async () => { throw failure; }); + const read = createExternalReviewWaitReader(db, "company", list); + await expect(Promise.all([read(["a"]), read(["a", "b"])])).rejects.toBe(failure); + expect(list).toHaveBeenCalledTimes(2); + }); +}); diff --git a/server/src/services/native-runtime/external-review-wait.ts b/server/src/services/native-runtime/external-review-wait.ts index b59aed390c3..054113ce9f1 100644 --- a/server/src/services/native-runtime/external-review-wait.ts +++ b/server/src/services/native-runtime/external-review-wait.ts @@ -78,3 +78,27 @@ export async function listExternalReviewWaits(db: Db, companyId: string, issueId })); return waits.filter((wait): wait is NonNullable => wait !== null); } + +export type ExternalReviewWait = Awaited>[number]; +export type ExternalReviewWaitReader = (issueIds: string[]) => Promise; + +/** Share one company's wait reads between callers that serve one request. + * Each issue's wait is independent of the other requested issues, so an issue + * is read at most once. Do not keep a reader beyond its request. */ +export function createExternalReviewWaitReader(db: Db, companyId: string, + list: typeof listExternalReviewWaits = listExternalReviewWaits): ExternalReviewWaitReader { + const byIssueId = new Map>(); + return async (issueIds) => { + const requested = [...new Set(issueIds)]; + const missing = requested.filter(issueId => !byIssueId.has(issueId)); + if (missing.length) { + const read = list(db, companyId, missing) + .then(waits => new Map(waits.map(wait => [wait.issueId, wait]))); + for (const issueId of missing) { + byIssueId.set(issueId, read.then(waits => waits.get(issueId) ?? null)); + } + } + const waits = await Promise.all(requested.map(issueId => byIssueId.get(issueId)!)); + return waits.filter((wait): wait is ExternalReviewWait => wait !== null); + }; +}