mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
perf(chat): index GitHub review lookups and share review-wait reads (#20400)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - GitHub chat endpoints let agents review pull requests. The server reads review state when it lists issues and when it updates GitHub checks. > - On a hosted instance, two GitHub review queries used the most total database time. One query ran about 660,000 times at about 33 ms. The other query ran about 10,000 times at about 1 s. > - The first query finds a source comment's chat link. No index covers `chat_message_links.comment_id`, so each call reads the full table. Issue lists also read the same review waits two times per request. > - The second query joins deliveries, links and runs on a JSON key. The planner walks the agent's run history in creation order and reads each run's context. > - This pull request adds four concurrent indexes, splits the run lookup into two small reads, and shares review-wait reads in one issue-list request. > - The benefit is less background database load, so request reads wait less. The results stay the same. ## Linked Issues or Issue Description **What existing behavior does this improve?** GitHub review state reads. These are the inherited review-wait lookup by source comment, and the review-check lookup of the run that a delivery's comment woke. **Subsystem affected** Server chat channels (GitHub review waits and checks), issue list attention maps, and the database schema. **Current behavior** - `readInheritedGitHubReviewWait` filters `chat_message_links` by `company_id`, `comment_id` and `direction`. No index has `comment_id`, so PostgreSQL reads the full table on each call. The run-origin check does the same lookup first. - An issue list calls the external review-wait read from the review attention map and again from the blocked-inbox attention map. The two maps read the same in-review issues. - `githubReviewCheckState` joins deliveries, links and runs with `context_snapshot->>'wakeCommentId' = comment_id::text`, sorts by `created_at desc` and takes one row. The planner reads the agent's runs in creation order and reads each run context. If no run matches yet, it reads all of them. The delivery filter on `normalized_event` and the link join on `delivery_id` also have no index. **Proposed behavior** The same rows come back from small index reads. Each issue's review wait is read one time per issue-list request. **Reason and benefit** These two statement families used about 33,000 s of database time in one cumulative statistics window. The database had about 50 concurrent active queries, and request reads waited behind this background work. **Breaking changes** None. The migration adds indexes only. ## What Changed - Migration `0456_chat_review_lookup_indexes` builds four indexes with `CREATE INDEX CONCURRENTLY IF NOT EXISTS`. It runs through the existing top-level statement path, which repairs an invalid index on retry: - `chat_message_links_company_comment_idx` on `(company_id, comment_id)` where `comment_id is not null`. - `chat_message_links_delivery_idx` on `(delivery_id)` where `delivery_id is not null`. - `chat_deliveries_github_automatic_delivery_idx` on `(endpoint_id, normalized_event->'githubAutomatic'->'context'->>'deliveryId')` where that value is not null. - `heartbeat_runs_company_wake_comment_idx` on `(company_id, context_snapshot->>'wakeCommentId')`. This index is not partial. The planner does not use expression statistics from a partial index. Without these statistics, it estimates many matches and keeps the creation-order walk. - `githubReviewCheckState` reads the delivery's comment IDs first. If there is a comment, it reads the newest same-agent run with one of these `wakeCommentId` values. If there is no comment, it skips the run read. The comparison stays text equality on `comment_id::text`. - `createExternalReviewWaitReader` shares one read per issue between the review and blocked-inbox attention maps of one issue-list request. The reader exists only for that request. - `doc/DATABASE.md` describes the indexes and the new read order. - `scripts/benchmark-review-lookup-indexes.ts` compares the pre-fix and new code with and without the indexes. `scripts/benchmark-review-lookup-index-migration.ts` times the migration on a large synthetic database. ## Verification Local SQL measurements on a synthetic fixture. They are not production timings. Environment: embedded PostgreSQL 18.1, baseline revision `02bfe71a6b`. Fixture: 50,000 links, 5,000 deliveries with about 2 KiB events, and 4,000 runs with about 12 KiB incompressible contexts. The wake run is in the middle of the run history. Each value is the median server execution time from `EXPLAIN ANALYZE` (2 warmups, 9 samples). The order was: no indexes, indexes, then no indexes again. | Caller and statement | Before (no indexes, two passes) | After | Shared buffers before → after | | --- | --- | --- | --- | | Inherited review wait, link by source comment (same SQL) | 2.72 / 2.36 ms | 0.076 ms | 929 → 12 | | Review check, wake run found | 31.9 / 21.9 ms | 0.29 ms (two statements) | 25,776 → 62 | | Review check, no wake run yet | 71.0 / 47.5 ms | 0.06 ms (two statements) | 34,861 → 12 | - Indexes alone do not fix the review check. The old SQL with the new indexes took 8.6 ms (run found) and 29.7 ms (no run), because it still walked the run history. The new SQL without the new indexes was neutral (23.5 and 35.1 ms). - Service wall times on this shared machine were noisy, so the table reports server execution time. - Correctness: the benchmark fails unless the inherited statement returns the same positional rows in each index state. The check service result and the final wake-run rows of the new code must also match the pre-fix module for both cases. - Migration timing, normal migration runner, synthetic data: 200,000 runs (1.8 GB with 8 KiB incompressible contexts), 500,000 links and 100,000 deliveries. First run 4.2 s; rerun 0.38 s. Separate builds: links by comment 1.5 s, run wake comment 1.1 s, deliveries 0.7 s, links by delivery 0.3 s. All indexes were valid. The run index was 8.3 MB. - New tests: - `server/src/__tests__/github-review-check-state.test.ts` checks that the newest same-agent run is selected. It also checks that other agents, other companies, uppercase IDs, non-string values, list-only `wakeCommentIds`, comment-less links and unknown deliveries do not match. It checks that the generated statements can use the new indexes. - `server/src/services/native-runtime/external-review-wait.test.ts` checks that the reader reads each issue one time, keeps concurrent callers correct, and rejects all waiting callers when a read fails. - Red/green: with the pre-fix `review-check-state.ts`, the behavior test passes and the index test fails. With this change, both pass. - Focused runs pass: `issue-blocker-attention`, `heartbeat-issue-liveness-escalation`, review-wait, review-check-state, external-review-wait, review-repair (106 tests), and `external-chat-wait.integration` (285 tests). `pnpm --filter @paperclipai/server typecheck` and `pnpm --filter @paperclipai/db check:migrations` pass. Commands: ```sh pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-review-lookup-indexes.ts --baseline-ref=02bfe71a6b pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-review-lookup-index-migration.ts ``` ## Risks - The migration builds four indexes concurrently. It does not block writes. On a large database, the run index reads every run context one time. The local run took 1.1 s for 1.8 GB from a warm cache. Cold storage on smaller compute can be slower. - The run index adds one entry for each run insert and for each run update that is not a heap-only update. Many runs have no `wakeCommentId`, so these entries hold a null key. - The review check now sends two statements instead of one when a delivery has a comment. Each statement reads only a few index entries. - When two matching runs have the same `created_at`, either can be selected. The previous query had the same tie behavior. - This change does not change the number of in-review issues that one issue list reads. The blocked-inbox map still reads all in-review issues of the company. A smaller scope needs a separate change to the liveness classifier. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`) by Anthropic, in Claude Code with tool use (shell, file edits, local PostgreSQL benchmarks). ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
DottaandClaude Opus 5.5 authored and GitHub committed
1 parent
326df52673
commit
347d85fef3
15 files changed
+741
-27
No files matched your search
@@ -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
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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'));
|
||||
+149
-3
@@ -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": [
|
||||
+7
@@ -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
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -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],
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
|
||||
let db: ReturnType<typeof createDb> | 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); }
|
||||
}
|
||||
@@ -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=<pre-fix commit>
|
||||
*
|
||||
* 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=<commit>");
|
||||
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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
|
||||
let db: ReturnType<typeof createDb> | 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<number>) {
|
||||
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<string, unknown>; "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<unknown>) => 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<string, unknown>) => {
|
||||
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<Record<string, unknown>> | 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<string, unknown> = {};
|
||||
const outputs: Record<string, unknown> = {};
|
||||
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); }
|
||||
}
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
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");
|
||||
});
|
||||
});
|
||||
@@ -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<ReturnType<typeof seedGitHubReviewWait>>) {
|
||||
await db.update(toolConnections).set({ enabled: false }).where(and(
|
||||
eq(toolConnections.id, fixture.connectionId), eq(toolConnections.companyId, fixture.companyId)));
|
||||
|
||||
@@ -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 })
|
||||
|
||||
@@ -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<Map<string, IssueReviewAttention>> {
|
||||
const result = new Map<string, IssueReviewAttention>();
|
||||
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<Map<string, IssueBlockedInboxAttention>> {
|
||||
const rowIssueIds = [...new Set(issueRows.map((row) => row.id))];
|
||||
const result = new Map<string, IssueBlockedInboxAttention>();
|
||||
@@ -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<string, number>()),
|
||||
@@ -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<string, IssueBlockerAttention>(), new Map<string, IssueReviewAttention>(), new Map<string, IssueBlockedInboxAttention>()]
|
||||
: await timePhase("attention", () => Promise.all([
|
||||
filters?.listProjection === "index" ? Promise.resolve(new Map<string, IssueBlockerAttention>()) : listIssueBlockerAttentionMap(db, companyId, withRuns),
|
||||
filters?.listProjection === "index" ? Promise.resolve(new Map<string, IssueReviewAttention>()) : listIssueReviewAttentionMap(db, companyId, withRuns),
|
||||
filters?.listProjection === "index" ? Promise.resolve(new Map<string, IssueReviewAttention>()) : listIssueReviewAttentionMap(db, companyId, withRuns, readExternalReviewWaits),
|
||||
includeBlockedInboxAttention
|
||||
? listIssueBlockedInboxAttentionMap(db, companyId, withRuns)
|
||||
? listIssueBlockedInboxAttentionMap(db, companyId, withRuns, readExternalReviewWaits)
|
||||
: Promise.resolve(new Map<string, IssueBlockedInboxAttention>()),
|
||||
]));
|
||||
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
@@ -78,3 +78,27 @@ export async function listExternalReviewWaits(db: Db, companyId: string, issueId
|
||||
}));
|
||||
return waits.filter((wait): wait is NonNullable<typeof wait> => wait !== null);
|
||||
}
|
||||
|
||||
export type ExternalReviewWait = Awaited<ReturnType<typeof listExternalReviewWaits>>[number];
|
||||
export type ExternalReviewWaitReader = (issueIds: string[]) => Promise<ExternalReviewWait[]>;
|
||||
|
||||
/** 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<string, Promise<ExternalReviewWait | null>>();
|
||||
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);
|
||||
};
|
||||
}
|
||||
Reference in new issue
Block a user