From 2a6d4d169425677d3b6850730d853f159cd62d7e Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Thu, 29 Oct 2026 09:06:41 -0500 Subject: [PATCH] perf(db): index canonical task activity lookups (#18043) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit https://entire.com/gh/paperclipai/-/trails/177/changes/paperclip/177 ## Thinking Path > - Paperclip helps people manage AI agents and their tasks. > - Task lists rank tasks by their latest canonical activity. > - The existing lookup reads every event for each task before pagination. > - Query-only alternatives failed a narrow visibility control or the performance goal. > - This PR adds an index that lets the unchanged lookup stop at the newest qualifying event. > - The result is less history work with the same permissions, ordering, and response. ## Linked Issues or Issue Description Refs: #13803. That broader proposal stores a new activity field with triggers and a backfill. This index was authored independently and uses the existing query. **Subsystem affected** Task-list activity ranking and the database schema. **Current behavior** The list computes `MAX(activity_log.created_at)` for every eligible task. The existing entity index reads 50 events per task across 1,199 candidates in the representative fixture. **Proposed behavior** Add `(company_id, entity_type, entity_id, created_at)`. Keep the production task service and permission helpers unchanged. PostgreSQL reads backward until an event passes the existing exclusions. **Reason and benefit** Fresh paired measurements on the current base improve the full-description scoped SQL from **44.60 to 20.33 ms (2.19×)**. Activity buffer accesses fall from **63,547 to 4,796**, and history rows per probe fall from 50 to one. The complete service median falls from 50.63 to 31.31 ms. Complete positional SQL results and service responses match. ## What Changed - Add generated migration `0414_closed_edwin_jarvis.sql` and the matching Drizzle index. - Use `CREATE INDEX CONCURRENTLY IF NOT EXISTS` through the existing top-level runner and invalid-index retry cleanup. The client change is one whitelist entry. - Preserve all upstream SQL and journal entries; retain snapshots 0410–0414. - Cover full and index projections, canonical ordering, invalid-index recovery, concurrent migrators, and incomplete fixture rejection. - Repair benchmark seeding, apply requested work memory to both connections, and record both settings. Hosted defaults now include six Slack conversations, 642 GitHub conversations, and 53 removed conversations. ## Verification - **All final-head CI gates passed**, including full workspace typechecks, tests, build, Runner checks, and all eight E2E shards. Greptile is **5/5** on `67ca7ae1cf15fc054ef515ffd8205b1fe30a0d55`; no unresolved review threads remain. - Local database/server typechecks and migration numbering/safety checks pass. Three server regression tests and three migration regression suites pass on real PostgreSQL. - The original pre-index work guard fails at eight rows per probe; the indexed guard now checks both the full-description query and index projection. The service oracle covers cleanup/inbox exclusions, raw text IDs, company scope, comments, ties, sort directions, visibility before pagination, empty lists, and one child per parent. - The documented pre-index seed command reproduced the review failure before the fix and now succeeds. A paired run with non-default memory confirms both independent connections use `16MB`. - Fresh proof pins pre-index base `929e840595c8933ca485ac7a2e99413e1e11ca64`, with identical cloned data, unchanged generated SQL and parameters, two warmups, and five alternating samples per arm. All **18 comparison cells** preserve every positional row and complete service response. The 15-case default-memory matrix passes its physical-work, speed, and control bounds. - The one-visible-task control is 6.64 → 7.25 ms SQL; the 20-task control is 3.71 → 3.72 ms. Both pass the declared 25% or 2 ms budget. Full-query memory controls use matching `4MB` or `16MB` settings. - Committed 1,000-row write batches measured 40.09 → 43.17 ms median (+7.7%). Ranges overlap (34.67–71.85 and 34.27–48.05 ms); this is an observed local cost, not a precise throughput guarantee. - The final-numbered migration builds a valid index through the normal runner on the fresh 300,000-row fixture in 0.55 s. Earlier large-fixture proof of the same index SQL used 5,000 tasks and 1.25M activity rows: 2.58 s first apply, concurrent write/read probes 5.16/2.43 ms, interrupted invalid-index cleanup, successful retry, and one receipt. Current migration recovery tests verify the renumbered path. - [The proof document](https://github.com/paperclipai/paperclip/blob/67ca7ae1cf15fc054ef515ffd8205b1fe30a0d55/docs/performance/task-list-activity-index.md) retains the historical baseline, rejected rewrites, and the earlier matrix invalidated by severe host contention. Those historical timings are not substituted for the fresh measurements above. ## Risks The migration is an exception after measured query-only alternatives failed. The index adds storage and work to every activity write: about 30 MiB for 300,000 rows and 123 MiB for 1.25M rows. Concurrent creation avoids the writer blocking observed with an ordinary index build, but long transactions can delay it. Local timing does not establish Neon build duration. The query still ranks all eligible tasks, and excluded newer events still need to be skipped. These are local SQL/service measurements; deployed HTTP latency remains unverified. No hosted DDL or deployment was performed. ## Model Used OpenAI Codex, GPT-6 family, with reasoning, code execution, and tool use. The environment does not expose the exact model identifier or context-window size. ## 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 linked existing issues or described the issue using the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links - [x] My branch name describes the change and contains no internal issue id - [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 - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge Co-authored-by: Paperclip --- docs/performance/task-list-activity-index.md | 162 ++++++++++++++++ packages/db/src/client.ts | 1 + .../migrations/0414_closed_edwin_jarvis.sql | 1 + ...{0409_snapshot.json => 0414_snapshot.json} | 183 +++++++++++++++++- packages/db/src/migrations/meta/_journal.json | 7 + packages/db/src/schema/activity_log.ts | 6 + ...task-list-activity-index-migration.test.ts | 45 +++++ ...benchmark-task-activity-index-migration.ts | 94 +++++++++ .../benchmark-task-activity-index-writes.ts | 54 ++++++ .../scripts/benchmark-task-list-projection.ts | 173 +++++++++++++++++ .../helpers/task-list-projection-fixture.ts | 104 ++++++++++ .../task-list-projection-query.test.ts | 69 +++++++ 12 files changed, 897 insertions(+), 2 deletions(-) create mode 100644 docs/performance/task-list-activity-index.md create mode 100644 packages/db/src/migrations/0414_closed_edwin_jarvis.sql rename packages/db/src/migrations/meta/{0409_snapshot.json => 0414_snapshot.json} (99%) create mode 100644 packages/db/src/task-list-activity-index-migration.test.ts create mode 100644 scripts/benchmark-task-activity-index-migration.ts create mode 100644 scripts/benchmark-task-activity-index-writes.ts create mode 100644 server/scripts/benchmark-task-list-projection.ts create mode 100644 server/src/__tests__/helpers/task-list-projection-fixture.ts create mode 100644 server/src/__tests__/task-list-projection-query.test.ts diff --git a/docs/performance/task-list-activity-index.md b/docs/performance/task-list-activity-index.md new file mode 100644 index 00000000000..8dec2655ea0 --- /dev/null +++ b/docs/performance/task-list-activity-index.md @@ -0,0 +1,162 @@ +# Task-list canonical activity index + +The main task list ranks tasks by the greatest task update, comment, or canonical +activity timestamp. The activity subquery used the existing entity index, then +read every activity row for each candidate task. The new B-tree index orders +`(company_id, entity_type, entity_id, created_at)`. PostgreSQL can read backward +until the first activity that passes the existing exclusions. + +The task service, conversation-state helper, and authorization helper are +unchanged. Permissions and waiting-conversation filters still run before +pagination. The index preserves raw text entity IDs, company scope, automatic +read-cleanup and local inbox exclusions, null behavior, all sort directions, +priority order, offsets, and one-child-per-parent selection. + +## Why this change needs a migration + +Query-only alternatives failed a required control. Reusing live conversation +state was neutral on the measured six-Slack-conversation workload: SQL +56.09 to 52.94 ms, service 67.66 to 72.93 ms. Grouping all company activity +improved a broad list but regressed a one-visible-task list from 23.70 to +52.09 ms. A direct join kept the narrow case fast but retained 1,199 history +probes and improved the broad case only 1.14 times. An eligible-ID CTE and +array lookup repaired the narrow case but still touched 67,401 activity buffers +and improved the broad case only 1.06 times. Those rewrites were rejected. + +All seven existing activity indexes were inspected. None orders timestamps after +both company and entity equality. The new index permits the original query and +all its filters to stay intact. It does not encode JSON or bound action values +in a partial-index predicate. The ordinary index build was rejected by the +migration safety gate. The final migration uses the existing concurrent-index +runner path and its existing invalid-index retry recovery. + +## Local paired query proof + +Initial baseline: `71b448b9cb841e6fa18fe883585430c5081245da`. Those query +sources also matched the earlier follow-up base `820c458465fa40137c81fe510f3e6dc573b4f895`. +The reproduction commands now pin `929e840595c8933ca485ac7a2e99413e1e11ca64` +after upstream migrations advanced. Upstream query helpers have changed since +the initial measurements; the historical timings below remain labeled with their +original baseline. This PR leaves the current task service and its helpers +unchanged. Our index is migration `0414_closed_edwin_jarvis.sql`. Both database arms contain +identical synthetic rows. The only physical difference is the new index. +The harness asserts identical generated SQL and parameters. It compares every +positional SQL column in order and the complete real service response outside +the timing interval. It uses two warmups and five alternating measured samples. +Plans are captured outside ordinary timing. + +The main fixture has 1,200 tasks, 300,000 activity rows, 7,200 comments, +4,802 runs, 33,810 wakes, and 701 conversations. Six conversations are eligible +Slack conversations, with one waiting; 642 GitHub conversations and 53 removed +Slack conversations are controls. Run contexts are about 27 KiB and contain +incompressible synthetic content. Descriptions include null, short Unicode, +and long Unicode content. The primary query includes the full description +preview, the real unchanged actor-privacy predicate, and canonical activity, +update timestamp, and ID descending order with limit 100. It does not use the +index-only list projection. + +Initial paired samples on Docker PostgreSQL 17.11, ordinary Docker storage, +4 MiB work memory, and JIT off: + +| Case | SQL median before → after | SQL range before → after | Service median before → after | +| --- | --- | --- | --- | +| Full scoped list, 100 rows | 96.78 → 45.07 ms | 84.44–109.24 → 39.35–49.09 ms | 126.74 → 70.19 ms | +| Same company, one visible task | 10.55 → 12.67 ms | 9.84–12.07 → 10.53–13.58 ms | 19.82 → 20.20 ms | + +The primary SQL gain is 2.15 times. Activity work falls from 50 rows per probe +and 63,547 buffer hits to one qualifying row per probe and 4,796 hits. The +one-visible-task SQL median increases 20.1%; this remains within the declared +25% or 2 ms control budget and is retained explicitly. + +A later 15-case matrix preserved all positional rows and complete service +responses. Host load reached 364–370 with about 32 GiB swap in use. Those timings +are **inconclusive** and are not the performance claim. The matrix also exposed +a benchmark selector bug that applied the broad 1,000-probe requirement to a +two-ID control; the selector now excludes explicit ID filters. + +A local dense-Slack experiment demonstrated JIT overhead, but the hosted target +had zero recorded JIT functions. This index makes no claim that hosted latency +was caused by JIT. SQL, full service timing, and HTTP request timing are separate +claims. No HTTP speedup or deployed improvement has been established. + +## Migration cost and recovery + +The normal application migrator was exercised on a larger synthetic database: +5,000 tasks, 1.25 million activity rows, 30,000 comments, 20,004 runs, and +140,420 wakes. Activity occupied about 2.19 GB including prior indexes and TOAST. +The new index is 129,163,264 bytes, about 123 MiB. The 300,000-row query fixture +index is about 30 MiB. This is an extra index on every activity write. + +The exact concurrent migration completed its first full normal-runner apply, +inspection, journal write, and commit in 2.577 seconds. A simultaneous write +took 5.16 ms and a read took 2.43 ms. A second trial cancelled an actual table +scan after 3,070 of 61,112 blocks. PostgreSQL left an invalid, unjournaled index. +The existing runner removed it and completed retry in 1.892 seconds; concurrent +write/read probes took 7.13/2.76 ms. The valid index was journaled once, and an +applied rerun took 26.8 ms. Deterministic tests also cover concurrent replay, +invalid-index recovery, retained data, and one journal receipt. + +These are local timings, including observed host contention. They do not predict +Neon storage or compute performance. The rejected ordinary-build trial blocked +a writer for 1.2–2.0 seconds; the final migration uses `CONCURRENTLY`. No hosted +DDL or deployment was performed. + +## Regression and reproduction + +The final work assertion fails on the pre-index schema: eight activity rows per +probe exceed the required one qualifying row. The indexed test passes. A real +service oracle checks excluded cleanup and all four inbox actions, foreign +company and malformed/uppercase entity IDs, comments, timestamp ties, ascending, +default and descending order, pagination, permissions, empty results, and +one-child-per-parent selection. + +Run the focused tests with the repository's normal PostgreSQL test support: + +```sh +pnpm --filter @paperclipai/server exec vitest run src/__tests__/task-list-projection-query.test.ts +pnpm --filter @paperclipai/db exec vitest run src/task-list-activity-index-migration.test.ts src/voice-inbound-cleanup-migration.test.ts src/question-response-history-migration.test.ts +``` + +The query harness is `server/scripts/benchmark-task-list-projection.ts`. It +requires a loopback admin URL in `TASK_PROJECTION_DATABASE_URL` and a disposable +`task_projection_*` database. For a fresh baseline fixture, use `--seed-only +--pre-index`: this applies the normal schema, then removes only this index and +its exact migration receipt before seeding. Existing fixtures use `--reuse` +and are not silently migrated. Set `--fixture-file=/tmp/task-projection.json` +to retain fixture IDs at a known portable path. + +```sh +node --import ./server/node_modules/tsx/dist/loader.mjs server/scripts/benchmark-task-list-projection.ts \ + --baseline-ref=929e840595c8933ca485ac7a2e99413e1e11ca64 \ + --db-name=task_projection_base --seed-only --pre-index \ + --fixture-file=/tmp/task-projection.json --count=1200 --chat-count=6 \ + --history=4 --payload-bytes=20480 --activity-per-issue=50 \ + --comments-per-issue=6 --activity-decoys=240000 --description-bytes=4096 --hosted-shape +``` + +Clone that stopped database to `task_projection_index` with PostgreSQL's normal +`CREATE DATABASE ... TEMPLATE ...` command. Apply its pending migration through +the normal application migrator. Then run an alternating pair: + +```sh +node --import ./server/node_modules/tsx/dist/loader.mjs server/scripts/benchmark-task-list-projection.ts \ + --baseline-ref=929e840595c8933ca485ac7a2e99413e1e11ca64 \ + --db-name=task_projection_base --treatment-db-name=task_projection_index \ + --fixture-file=/tmp/task-projection.json --reuse --all-read-modes --jit=off \ + --assert-performance --output=/tmp/task-projection-report.json +``` + +Use `--quick --full-projection --scoped --read-one` for the one-visible-task +control and `--jit=on` for the compiler control. Reports retain settings, +source hashes, index catalogs, complete samples, SQL, parameters, plans, and +host load. Per-case checkpoints preserve completed work if a later case stops. +The separate root scripts `benchmark-task-activity-index-migration.ts` and +`benchmark-task-activity-index-writes.ts` measure the normal migration and +committed write transactions. Their URLs must also name disposable loopback +`task_projection_*` databases. + +The first useful baseline parse, seed, execution, and plan evidence took about +24 seconds. Focused final work red and green loops took 12.16 and 26.29 seconds. +Schema generation and the owning database typecheck/migration gates passed. +Full workspace verification belongs to the final PR head; older discarded +runtime checks are not final-index evidence. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index af421ce9779..a872d9ff65a 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -563,6 +563,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0411_real_scrambler.sql", "0412_tearful_squadron_supreme.sql", "0413_tired_kronos.sql", + "0414_closed_edwin_jarvis.sql", "0384_funny_adam_warlock.sql", diff --git a/packages/db/src/migrations/0414_closed_edwin_jarvis.sql b/packages/db/src/migrations/0414_closed_edwin_jarvis.sql new file mode 100644 index 00000000000..64ccc843ba7 --- /dev/null +++ b/packages/db/src/migrations/0414_closed_edwin_jarvis.sql @@ -0,0 +1 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "activity_log_company_entity_created_idx" ON "activity_log" USING btree ("company_id","entity_type","entity_id","created_at"); \ No newline at end of file diff --git a/packages/db/src/migrations/meta/0409_snapshot.json b/packages/db/src/migrations/meta/0414_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0409_snapshot.json rename to packages/db/src/migrations/meta/0414_snapshot.json index 54b5f4d70ba..2679e7d98bf 100644 --- a/packages/db/src/migrations/meta/0409_snapshot.json +++ b/packages/db/src/migrations/meta/0414_snapshot.json @@ -1,6 +1,6 @@ { - "id": "94acf3fe-50f3-4a97-a7fe-865abbc9d0a3", - "prevId": "6bd67ad8-0d80-4163-873b-8b8eacf32bb2", + "id": "27c9008c-60e3-449b-9630-9e4977fad444", + "prevId": "5690a14a-f9cc-41a6-af51-c27f688f4ef6", "version": "7", "dialect": "postgresql", "tables": { @@ -1351,6 +1351,39 @@ "method": "btree", "with": {} }, + "activity_log_company_entity_created_idx": { + "name": "activity_log_company_entity_created_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "entity_type", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "entity_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "created_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, "activity_log_company_user_issue_actor_idx": { "name": "activity_log_company_user_issue_actor_idx", "columns": [ @@ -33771,6 +33804,34 @@ } }, "indexes": { + "heartbeat_runs_native_unpresented_response_idx": { + "name": "heartbeat_runs_native_unpresented_response_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "native_issue_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"heartbeat_runs\".\"runtime_mode\" = 'native' AND \"heartbeat_runs\".\"status\" in ('succeeded', 'failed', 'timed_out') AND \"heartbeat_runs\".\"finished_at\" is not null AND jsonb_typeof(\"heartbeat_runs\".\"result_json\" #> '{presentationDecision,commentId}') is distinct from 'string'", + "concurrently": false, + "method": "btree", + "with": {} + }, "heartbeat_runs_cancelled_workspace_preservation_idx": { "name": "heartbeat_runs_cancelled_workspace_preservation_idx", "columns": [ @@ -36475,6 +36536,40 @@ "method": "btree", "with": {} }, + "issue_comments_review_response_receipt_idx": { + "name": "issue_comments_review_response_receipt_idx", + "columns": [ + { + "expression": "created_by_run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "issue_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "author_agent_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"issue_comments\".\"author_type\" = 'agent' AND \"issue_comments\".\"metadata\"->>'authorizationReason' = 'allow_chat_run_presentation'", + "concurrently": false, + "method": "btree", + "with": {} + }, "issue_comments_company_idx": { "name": "issue_comments_company_idx", "columns": [ @@ -43108,6 +43203,62 @@ } }, "indexes": { + "native_run_finalizations_pending_effect_owner_idx": { + "name": "native_run_finalizations_pending_effect_owner_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "issue_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "decision_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"native_run_finalizations\".\"decision_id\" is not null AND \"native_run_finalizations\".\"result_id\" is not null AND \"native_run_finalizations\".\"phase\" <> 'terminal_failure'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "native_run_finalizations_unsafe_export_recovery_idx": { + "name": "native_run_finalizations_unsafe_export_recovery_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "issue_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"native_run_finalizations\".\"phase\" = 'terminal_failure' AND \"native_run_finalizations\".\"failure_code\" = 'native_workspace_sync_out_unsafe_archive'", + "concurrently": false, + "method": "btree", + "with": {} + }, "native_run_finalizations_control_deadline_idx": { "name": "native_run_finalizations_control_deadline_idx", "columns": [ @@ -52412,6 +52563,34 @@ } }, "indexes": { + "status_decision_effects_pending_delivery_idx": { + "name": "status_decision_effects_pending_delivery_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "issue_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "decision_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"status_decision_effects\".\"delivery_state\" in ('pending', 'failed')", + "concurrently": false, + "method": "btree", + "with": {} + }, "status_decision_effects_decision_ordinal_uq": { "name": "status_decision_effects_decision_ordinal_uq", "columns": [ diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 16ac6fbe13b..6fdfc546aee 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2885,6 +2885,13 @@ "when": 1793251083562, "tag": "0413_tired_kronos", "breakpoints": true + }, + { + "idx": 414, + "version": "7", + "when": 1793280975229, + "tag": "0414_closed_edwin_jarvis", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/activity_log.ts b/packages/db/src/schema/activity_log.ts index 03896f7c702..1dc44a561ff 100644 --- a/packages/db/src/schema/activity_log.ts +++ b/packages/db/src/schema/activity_log.ts @@ -34,6 +34,12 @@ export const activityLog = pgTable( ), runIdIdx: index("activity_log_run_id_idx").on(table.runId), entityIdx: index("activity_log_entity_type_id_idx").on(table.entityType, table.entityId), + companyEntityCreatedIdx: index("activity_log_company_entity_created_idx").on( + table.companyId, + table.entityType, + table.entityId, + table.createdAt, + ), companyUserIssueActorIdx: index("activity_log_company_user_issue_actor_idx") .on(table.companyId, table.actorId, table.entityId) .where(sql`${table.actorType} = 'user' AND ${table.entityType} = 'issue'`), diff --git a/packages/db/src/task-list-activity-index-migration.test.ts b/packages/db/src/task-list-activity-index-migration.test.ts new file mode 100644 index 00000000000..004a690fb0a --- /dev/null +++ b/packages/db/src/task-list-activity-index-migration.test.ts @@ -0,0 +1,45 @@ +import { createHash } from "node:crypto"; +import { readFile } from "node:fs/promises"; +import postgres from "postgres"; +import { describe, expect, it } from "vitest"; +import { applyPendingMigrations } from "./client.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./test-embedded-postgres.js"; + +const support = await getEmbeddedPostgresTestSupport(); +const indexName = "activity_log_company_entity_created_idx"; +const migrationFile = "0414_closed_edwin_jarvis.sql"; + +(support.supported ? describe : describe.skip)("task-list activity index migration", () => { + it("repairs an unjournaled invalid index and records one receipt after concurrent replay", async () => { + const database = await startEmbeddedPostgresTestDatabase("task-activity-index-"); + const sql = postgres(database.connectionString, { max: 1, onnotice: () => {} }); + try { + const companies = await sql`INSERT INTO companies (name,issue_prefix) VALUES ('First','TAI1'),('Second','TAI2') RETURNING id`; + await sql`INSERT INTO activity_log(company_id,actor_id,action,entity_type,entity_id) + VALUES (${companies[0]!.id},'reader','issue.updated','issue','canonical-task')`; + const content = await readFile(new URL(`./migrations/${migrationFile}`, import.meta.url), "utf8"); + const hash = createHash("sha256").update(content).digest("hex"); + await sql`DELETE FROM drizzle.__drizzle_migrations WHERE hash=${hash}`; + await sql.unsafe(`DROP INDEX ${indexName}`); + await expect(sql.unsafe(`CREATE UNIQUE INDEX CONCURRENTLY ${indexName} ON companies ((true))`)).rejects.toThrow(); + const [invalid] = await sql`SELECT indisvalid FROM pg_index WHERE indexrelid=${indexName}::regclass`; + expect(invalid!.indisvalid).toBe(false); + await Promise.all([applyPendingMigrations(database.connectionString), applyPendingMigrations(database.connectionString)]); + const [rebuilt] = await sql`SELECT indisvalid,indisunique,pg_get_indexdef(indexrelid) definition + FROM pg_index WHERE indexrelid=${indexName}::regclass`; + expect(rebuilt!.indisvalid).toBe(true); + expect(rebuilt!.indisunique).toBe(false); + expect(rebuilt!.definition).toContain("activity_log"); + expect(rebuilt!.definition).toContain("company_id, entity_type, entity_id, created_at"); + expect(await sql`SELECT entity_id FROM activity_log WHERE company_id=${companies[0]!.id}`).toEqual([{ entity_id: "canonical-task" }]); + const [history] = await sql`SELECT count(*)::int count FROM drizzle.__drizzle_migrations WHERE hash=${hash}`; + expect(history!.count).toBe(1); + await applyPendingMigrations(database.connectionString); + const [unchanged] = await sql`SELECT count(*)::int count FROM drizzle.__drizzle_migrations WHERE hash=${hash}`; + expect(unchanged!.count).toBe(1); + } finally { + await sql.end(); + await database.cleanup(); + } + }, 120_000); +}); diff --git a/scripts/benchmark-task-activity-index-migration.ts b/scripts/benchmark-task-activity-index-migration.ts new file mode 100644 index 00000000000..f835c5c4f30 --- /dev/null +++ b/scripts/benchmark-task-activity-index-migration.ts @@ -0,0 +1,94 @@ +import assert from "node:assert/strict"; +import { availableParallelism, loadavg } from "node:os"; +import { promises as fs } from "node:fs"; +import { applyPendingMigrations, createDb, inspectMigrations } from "../packages/db/src/index.js"; + +const host = { availableCpus: availableParallelism(), loadBefore: loadavg() }; +const arg = (name: string) => process.argv.find(value => value.startsWith(`--${name}=`))?.slice(name.length + 3); +const connectionString = process.env.TASK_PROJECTION_MIGRATION_URL; +assert.ok(connectionString, "Set TASK_PROJECTION_MIGRATION_URL to an isolated pre-migration database"); +const url = new URL(connectionString); +assert.ok(["127.0.0.1", "localhost"].includes(url.hostname)); +assert.match(url.pathname, /^\/task_projection_[a-z0-9_]+$/); +const db = createDb(connectionString, { maxConnections: 1 }); +const writer = createDb(connectionString, { maxConnections: 1 }); +const reader = createDb(connectionString, { maxConnections: 1 }); +const indexName = "activity_log_company_entity_created_idx"; +const pause = () => new Promise(resolve => setTimeout(resolve, 10)); +const indexState = () => db.$client.unsafe("select i.indisvalid,i.indisready,pg_relation_size(c.oid)::text bytes from pg_index i join pg_class c on c.oid=i.indexrelid where c.relname=$1", [indexName]); +const keepAlive = setInterval(() => {}, 1000); +try { + await writer.$client.unsafe("select 1"); + await reader.$client.unsafe("select 1"); + const before = await inspectMigrations(connectionString); + assert.equal(before.pendingMigrations.length, 1, "Only the activity index migration may be pending"); + assert.equal((await indexState()).length, 0, "Start without the index"); + const sizes = await db.$client.unsafe("select c.relname,c.reltuples::bigint estimated_rows,pg_relation_size(c.oid)::text heap_bytes,pg_total_relation_size(c.oid)::text total_bytes from pg_class c join pg_namespace n on n.oid=c.relnamespace where n.nspname='public' and c.relname in ('issues','activity_log','issue_comments','heartbeat_runs','agent_wakeup_requests') order by c.relname"); + const settings = await db.$client.unsafe("select version(),current_setting('work_mem') work_mem,current_setting('maintenance_work_mem') maintenance_work_mem,current_setting('max_parallel_maintenance_workers') max_parallel_maintenance_workers"); + const progress = async () => db.$client.unsafe("select pid,phase,blocks_total,blocks_done from pg_stat_progress_create_index where datname=current_database() and relid='activity_log'::regclass"); + let cancellation: unknown; + if (!process.argv.includes("--skip-cancel")) { + const cancelStart = performance.now(); + let cancelledFinished = false; + const cancelledRun = applyPendingMigrations(connectionString).then(() => ({ completed: true }), error => ({ code: error.code, message: error.message })).finally(() => { cancelledFinished = true; }); + let cancellationProgress: unknown; + while (!cancelledFinished) { + const rows = await progress(); + if (rows[0]?.phase.startsWith("building index:") && Number(rows[0]?.blocks_done) > 100) { + cancellationProgress = rows[0]; + assert.equal((await db.$client.unsafe("select pg_cancel_backend($1) cancelled", [rows[0]!.pid]))[0]!.cancelled, true); + break; + } + await pause(); + } + const result = await cancelledRun; + const invalidIndex = await indexState(); + cancellation = { result, progress: cancellationProgress, totalMs: performance.now() - cancelStart, invalidIndex }; + assert.equal((result as { code?: string }).code, "57014", "Cancel the actual index build"); + assert.equal(invalidIndex[0]?.indisvalid, false, "Cancelled concurrent build must leave an invalid index for the runner to repair"); + assert.deepEqual((await inspectMigrations(connectionString)).pendingMigrations, before.pendingMigrations, "Cancelled build must not be journaled"); + } + const [sample] = await db.$client.unsafe("select company_id,entity_id from activity_log where entity_type='issue' limit 1"); + assert.ok(sample); + const insert = () => writer.$client.unsafe("insert into activity_log(company_id,entity_type,entity_id,actor_type,actor_id,action,details) values($1,'issue',$2,'system','migration-probe','benchmark.migration_probe','{}') returning id", [sample.company_id, sample.entity_id]); + const buildStart = performance.now(); + let finished = false; + const build = applyPendingMigrations(connectionString).finally(() => { finished = true; }); + let buildProgress: unknown; + let writeStart = 0; + let writeMs: number | undefined; + let readMs: number | undefined; + let insertedId: string | undefined; + let concurrent: Promise | undefined; + while (!finished) { + const rows = await progress(); + if (rows.length) { + buildProgress = rows[0]; + writeStart = performance.now(); + concurrent = insert().then(rows => { writeMs = performance.now() - writeStart; insertedId = rows[0]!.id; }); + const readStart = performance.now(); + await reader.$client.unsafe("select id from activity_log where company_id=$1 order by created_at desc limit 20", [sample.company_id]); + readMs = performance.now() - readStart; + break; + } + await pause(); + } + await build; + const buildMs = performance.now() - buildStart; + await concurrent; + assert.ok(buildProgress, "Observe the build while ordinary readers and writers run"); + assert.equal((await inspectMigrations(connectionString)).status, "upToDate"); + assert.equal((await indexState())[0]!.indisvalid, true); + if (insertedId) await writer.$client.unsafe("delete from activity_log where id=$1", [insertedId]); + const retryStart = performance.now(); + await applyPendingMigrations(connectionString); + const retryMs = performance.now() - retryStart; + const report = { host: { ...host, loadAfter: loadavg() }, settings, beforePending: before.pendingMigrations, sizes, cancellation, successfulBuild: { totalMs: buildMs, progress: buildProgress, concurrentWriteMs: writeMs, concurrentReadMs: readMs }, appliedRerunMs: retryMs, index: await indexState() }; + if (arg("output")) await fs.writeFile(arg("output")!, JSON.stringify(report, null, 2)); + console.log(JSON.stringify(report, null, 2)); +} finally { + clearInterval(keepAlive); + await reader.$client.end({ timeout: 0 }); + await writer.$client.end({ timeout: 0 }); + await db.$client.end({ timeout: 0 }); +} diff --git a/scripts/benchmark-task-activity-index-writes.ts b/scripts/benchmark-task-activity-index-writes.ts new file mode 100644 index 00000000000..cd3077b1888 --- /dev/null +++ b/scripts/benchmark-task-activity-index-writes.ts @@ -0,0 +1,54 @@ +import assert from "node:assert/strict"; +import { availableParallelism, loadavg } from "node:os"; +import { randomBytes } from "node:crypto"; +import { promises as fs } from "node:fs"; +import { createDb } from "../packages/db/src/index.js"; + +const host = { availableCpus: availableParallelism(), loadBefore: loadavg() }; +const arg = (name: string) => process.argv.find(value => value.startsWith(`--${name}=`))?.slice(name.length + 3); +const urls = [process.env.TASK_PROJECTION_BASELINE_URL, process.env.TASK_PROJECTION_TREATMENT_URL]; +assert.ok(urls.every(Boolean), "Set isolated baseline and treatment database URLs"); +for (const value of urls) { + const url = new URL(value!); + assert.ok(["127.0.0.1", "localhost"].includes(url.hostname)); + assert.match(url.pathname, /^\/task_projection_[a-z0-9_]+$/); +} +const clients = urls.map(value => createDb(value!, { maxConnections: 1 })); +const samples: number[][] = [[], []]; +const noise = randomBytes(3072).toString("base64"); +const count = Number(arg("count") ?? 1000); +assert.ok(Number.isSafeInteger(count) && count > 0 && count <= 10000); +const keepAlive = setInterval(() => {}, 1000); +try { + const records = await clients[0]!.$client.unsafe("select company_id,id from issues order by id limit 1200"); + assert.ok(records.length); + const companyId = records[0]!.company_id; + const issueIds = records.filter(row => row.company_id === companyId).map(row => row.id); + for (const client of clients) await client.$client.unsafe("select 1"); + for (let round = 0; round < 7; round++) for (const arm of round % 2 ? [1, 0] : [0, 1]) { + const client = clients[arm]!; + const started = performance.now(); + await client.$client.unsafe("begin"); + let inserted: { id: string }[] = []; + try { + inserted = await client.$client.unsafe<{ id: string }[]>(`insert into activity_log(company_id,entity_type,entity_id,actor_type,actor_id,action,details) + select $1::uuid,case when n%5=0 then 'issue' else 'heartbeat_run' end, + case when n%5=0 then ($2::text[])[1+n%array_length($2::text[],1)] else gen_random_uuid()::text end, + 'agent','index-write-benchmark','issue.updated',jsonb_build_object('noise',left($3,case when n%5=0 then 4096 else 256 end)) + from generate_series(1,$4::integer)n returning id`, [companyId, issueIds, noise, count]); + await client.$client.unsafe("commit"); + } catch (error) { + await client.$client.unsafe("rollback"); + throw error; + } + if (round >= 2) samples[arm]!.push(performance.now() - started); + await client.$client.unsafe("delete from activity_log where id=ANY($1::uuid[])", [inserted.map(row => row.id)]); + } + const stats = (values: number[]) => ({ medianMs: [...values].sort((a, b) => a - b)[2], meanMs: values.reduce((sum, n) => sum + n, 0) / values.length, minMs: Math.min(...values), maxMs: Math.max(...values), samplesMs: values }); + const report = { host: { ...host, loadAfter: loadavg() }, rowsPerTransaction: count, warmupsPerArm: 2, samplesPerArm: 5, commitIncluded: true, cleanupOutsideTiming: true, baseline: stats(samples[0]!), treatment: stats(samples[1]!) }; + if (arg("output")) await fs.writeFile(arg("output")!, JSON.stringify(report, null, 2)); + console.log(JSON.stringify(report, null, 2)); +} finally { + clearInterval(keepAlive); + for (const client of clients) await client.$client.end({ timeout: 0 }); +} diff --git a/server/scripts/benchmark-task-list-projection.ts b/server/scripts/benchmark-task-list-projection.ts new file mode 100644 index 00000000000..c3eb77296cc --- /dev/null +++ b/server/scripts/benchmark-task-list-projection.ts @@ -0,0 +1,173 @@ +import assert from "node:assert/strict"; +import { availableParallelism, loadavg, tmpdir } from "node:os"; +import { execFileSync } from "node:child_process"; +import { createHash, randomUUID } from "node:crypto"; +import { promises as fs } from "node:fs"; +import path from "node:path"; +import { fileURLToPath, pathToFileURL } from "node:url"; +import { sql } from "drizzle-orm"; +import { applyPendingMigrations, createDb, issues } from "../../packages/db/src/index.js"; +import { issueReadSqlCondition } from "../src/services/authorization.js"; +import { issueService } from "../src/services/issues.js"; +import { captureTaskListQuery, seedTaskListProjectionFixture, type ListFilters, type ProjectionQuery } from "../src/__tests__/helpers/task-list-projection-fixture.js"; + +const started = performance.now(); +const host = { availableCpus: availableParallelism(), loadBefore: loadavg() }; +const root = fileURLToPath(new URL("../../", import.meta.url)); +const arg = (name: string) => process.argv.find(value => value.startsWith(`--${name}=`))?.slice(name.length + 3); +const git = (args: string[]) => execFileSync("git", args, { cwd: root, encoding: "utf8" }); +const baselineRef = arg("baseline-ref"); +assert.ok(baselineRef, "Pass a pinned pre-fix --baseline-ref"); +const baselineRevision = git(["rev-parse", `${baselineRef}^{commit}`]).trim(); +const modulePaths = ["server/src/services/issues.ts", "server/src/services/slack-conversation-state.ts"]; +const proofPaths = [...modulePaths, "server/src/services/authorization.ts", "server/scripts/benchmark-task-list-projection.ts", "server/src/__tests__/helpers/task-list-projection-fixture.ts", "packages/db/src/schema/activity_log.ts", "packages/db/src/client.ts", "packages/db/src/migrations/0414_closed_edwin_jarvis.sql"]; +const changedDependencies = git(["diff", "--name-only", baselineRevision, "--", "server/src/services", "packages/db/src", "packages/shared/src"]).trim().split("\n").filter(Boolean); +const indexOnly = Boolean(arg("treatment-db-name") || arg("validate-report")) + || (process.argv.includes("--seed-only") && process.argv.includes("--pre-index")); +assert.deepEqual(changedDependencies.filter(file => !file.endsWith(".test.ts") && !modulePaths.includes(file) && !(indexOnly && ((file === "packages/db/src/schema/activity_log.ts" || file === "packages/db/src/client.ts") || file.startsWith("packages/db/src/migrations/")))), [], "All imported query dependencies, including authorization, must stay identical"); +const baselineDir = await fs.mkdtemp(path.join(root, "server/scripts/.task-projection-baseline-")); +for (const modulePath of modulePaths) { + const source = git(["show", `${baselineRevision}:${modulePath}`]).replace(/from "(\.{1,2}\/[^\"]+)"/g, (_match, specifier: string) => { + const resolved = fileURLToPath(new URL(specifier, pathToFileURL(path.join(root, modulePath)))); + return `from ${JSON.stringify(resolved.endsWith("/services/slack-conversation-state.js") ? path.join(baselineDir, "slack-conversation-state.ts") : resolved)}`; + }); + await fs.writeFile(path.join(baselineDir, path.basename(modulePath)), source); +} +function assertPerformance(report: any, baseline = false) { + const activityNodes = (node: any): any[] => [ + ...(node["Relation Name"] === "activity_log" ? [node] : []), + ...(node.Plans ?? []).flatMap(activityNodes), + ]; + for (const result of report.results) { + const target = report.fixture.options.hostedShape && result.readMode === "scoped" && result.jit === "off" + && result.filters.limit === 100 && !result.filters.listProjection && result.filters.sortField === "updated" + && result.filters.sortDir === "desc" && !result.filters.q && !result.filters.offset && !result.filters.issueIds; + if (target) { + const oldNodes = activityNodes(result.baseline.plan[0].Plan); + const newNodes = activityNodes((baseline ? result.baseline : result.treatment).plan[0].Plan); + assert.ok(oldNodes.some(node => node["Actual Loops"] >= 1000 && node["Actual Rows"] >= 40), "Representative fixture must reproduce full per-task history probes"); + assert.ok(newNodes.every(node => node["Actual Rows"] <= 1), "Latest-activity probes must stop at one qualifying row"); + assert.ok(oldNodes.reduce((sum, node) => sum + node["Shared Hit Blocks"], 0) / newNodes.reduce((sum, node) => sum + node["Shared Hit Blocks"], 0) >= 5, "Activity buffer work must fall at least fivefold"); + if (!baseline) assert.ok(result.baseline.medianMs / result.treatment.medianMs >= 1.5, "Representative SQL must improve at least 1.5x"); + } + if (!target && !result.baseline.plan[0].JIT) { + assert.ok(result.treatment.medianMs <= Math.max(result.baseline.medianMs * 1.25, result.baseline.medianMs + 2), "Control SQL must stay within 25% or 2ms of baseline"); + } + } +} +if (arg("validate-report")) { + const report = JSON.parse(await fs.readFile(arg("validate-report")!, "utf8")); + assert.equal(report.baselineRevision, baselineRevision); + for (const [file, hash] of Object.entries(report.sourceHashes)) assert.equal(createHash("sha256").update(await fs.readFile(path.join(root, file))).digest("hex"), hash, `Measured source changed: ${file}`); + try { assertPerformance(report, process.argv.includes("--validate-baseline")); } + finally { await fs.rm(baselineDir, { recursive: true, force: true }); } + console.log(JSON.stringify({ status: "performance criteria passed", baselineRevision })); + process.exit(0); +} +const baseline = await import(pathToFileURL(path.join(baselineDir, "issues.ts")).href) as typeof import("../src/services/issues.js"); +const adminUrl = process.env.TASK_PROJECTION_DATABASE_URL; +assert.ok(adminUrl, "Set TASK_PROJECTION_DATABASE_URL to a disposable local PostgreSQL admin database"); +const url = new URL(adminUrl); +assert.ok(["127.0.0.1", "localhost"].includes(url.hostname), "Use isolated local PostgreSQL only"); +const databaseName = arg("db-name") ?? `task_projection_${randomUUID().replaceAll("-", "")}`; +assert.match(databaseName, /^task_projection_[a-z0-9_]+$/); +const fixtureFile = arg("fixture-file") ?? path.join(tmpdir(), `${databaseName}.fixture.json`); +const admin = createDb(adminUrl, { maxConnections: 1 }); +const [existing] = await admin.$client.unsafe("select 1 from pg_database where datname=$1", [databaseName]); +if (!existing) await admin.$client.unsafe(`CREATE DATABASE "${databaseName}"`); +url.pathname = `/${databaseName}`; +if (!process.argv.includes("--reuse")) await applyPendingMigrations(url.href); +const db = createDb(url.href, { maxConnections: 1 }); +const treatmentDatabaseName = arg("treatment-db-name") ?? databaseName; +assert.match(treatmentDatabaseName, /^task_projection_[a-z0-9_]+$/); +const treatmentUrl = new URL(url.href); +treatmentUrl.pathname = `/${treatmentDatabaseName}`; +const treatmentDb = treatmentDatabaseName === databaseName ? db : createDb(treatmentUrl.href, { maxConnections: 1 }); +if (process.argv.includes("--pre-index")) { + assert.ok(process.argv.includes("--seed-only") && !process.argv.includes("--reuse") && !arg("treatment-db-name"), "Pre-index restoration is only for a new seed-only fixture"); + const migration = await fs.readFile(path.join(root, "packages/db/src/migrations/0414_closed_edwin_jarvis.sql")); + const hash = createHash("sha256").update(migration).digest("hex"); + await db.$client.unsafe("drop index activity_log_company_entity_created_idx"); + const removed = await db.$client.unsafe("delete from drizzle.__drizzle_migrations where hash=$1 returning id", [hash]); + assert.equal(removed.length, 1, "Restore only the new index and its exact migration receipt"); +} +const keepAlive = setInterval(() => {}, 1000); +try { + const workMem = arg("work-mem"); + if (workMem) assert.match(workMem, /^\d+(kB|MB)$/); + for (const target of new Set([db, treatmentDb])) { + await target.$client.unsafe("set client_min_messages = warning"); + if (workMem) await target.$client.unsafe("select set_config('work_mem', $1, false)", [workMem]); + } + const settingsSql = "select version(), current_setting('shared_buffers') shared_buffers, current_setting('work_mem') work_mem, current_setting('jit_above_cost') jit_above_cost"; + const [settings] = await db.$client.unsafe(settingsSql); + const [treatmentSettings] = await treatmentDb.$client.unsafe(settingsSql); + assert.equal(treatmentSettings!.work_mem, settings!.work_mem, "Paired databases must use the same work_mem"); + await issueService(db).list(randomUUID(), { limit: 1 }); + const smokeMs = performance.now() - started; + const fixture = process.argv.includes("--reuse") + ? JSON.parse(await fs.readFile(fixtureFile, "utf8")) as Awaited> + : await seedTaskListProjectionFixture(db, { count: Number(arg("count") ?? 1200), chatCount: Number(arg("chat-count") ?? (process.argv.includes("--hosted-shape") ? 6 : 600)), history: Number(arg("history") ?? 12), payloadBytes: Number(arg("payload-bytes") ?? 3072), liveCount: Number(arg("live-count") ?? 12), activityPerIssue: Number(arg("activity-per-issue") ?? 0), commentsPerIssue: Number(arg("comments-per-issue") ?? 0), activityDecoys: Number(arg("activity-decoys") ?? 0), descriptionBytes: Number(arg("description-bytes") ?? 0), hostedShape: process.argv.includes("--hosted-shape") }); + await fs.writeFile(fixtureFile, JSON.stringify(fixture)); + const seedMs = performance.now() - started - smokeMs; + const execute = (query: ProjectionQuery, target = db) => target.$client.unsafe(query.sql, query.params as never[]).values(); + const stats = (values: number[]) => { const sorted = [...values].sort((a,b) => a-b); return { medianMs: sorted[Math.floor(values.length / 2)], minMs: sorted[0], maxMs: sorted.at(-1), meanMs: values.reduce((a,b) => a+b,0) / values.length, samplesMs: values }; }; + const results = []; + const cases: ListFilters[] = process.argv.includes("--seed-only") ? [] : process.argv.includes("--quick") ? [process.argv.includes("--full-projection") ? { limit: 100, sortField: "updated", sortDir: "desc" } : { limit: 100, listProjection: "index" }] : [ + { limit: 1, listProjection: "index" }, { limit: 100, listProjection: "index" }, { limit: 1000, listProjection: "index" }, + { limit: 100, sortField: "updated", sortDir: "desc" }, + { limit: 1000, sortField: "updated", sortDir: "desc" }, + { limit: 1, sortField: "updated", sortDir: "desc" }, + { limit: 100, sortField: "updated", sortDir: "asc" }, + { limit: 100, sortField: "updated", sortDir: "desc", offset: 100 }, + { limit: 100, issueIds: fixture.records.slice(0,2).map(row => row.id), sortField: "updated", sortDir: "desc" }, + { limit: 100, listProjection: "index", q: "Projection" }, { limit: 100, listProjection: "index", offset: 100 }, + ]; + const readModes = process.argv.includes("--scoped") ? ["scoped"] : process.argv.includes("--all-read-modes") ? ["unrestricted", "scoped"] : ["unrestricted"]; + for (const readMode of readModes) for (const jit of arg("jit") ? [arg("jit")!] : process.argv.includes("--quick") ? ["on"] : ["on", "off"]) for (const rawFilters of cases) { + assert.ok(jit === "on" || jit === "off"); + if (readMode === "scoped" && (rawFilters?.q || rawFilters?.offset || rawFilters?.limit !== 100 || rawFilters?.listProjection === "rows")) continue; + const filters = readMode === "scoped" ? { ...rawFilters, readCondition: await issueReadSqlCondition(db, { + type: "board", source: "session", userId: "projection-human", authorizingUserId: "projection-human", + }) } : { ...rawFilters }; + if (process.argv.includes("--read-one")) filters!.readCondition = sql`(${filters!.readCondition ?? sql`true`}) and exists ( + select 1 from jsonb_array_elements_text(${JSON.stringify([fixture.records.at(-1)!.id])}::jsonb) permitted(id) + where permitted.id = ${issues.id}::text)`; + await db.$client.unsafe(`set jit=${jit}`); + if (treatmentDb !== db) await treatmentDb.$client.unsafe(`set jit=${jit}`); + const before = await captureTaskListQuery(db, baseline.issueService, fixture.companyId, filters); + const after = await captureTaskListQuery(treatmentDb, issueService, fixture.companyId, filters); + if (treatmentDb !== db) assert.deepEqual(after, before, "Index-only comparison must generate identical SQL and parameters"); + const beforeRows = await execute(before), afterRows = await execute(after, treatmentDb); + assert.deepEqual(afterRows, beforeRows, "Every positional SQL column and ordered row must match"); + const samples = { baseline: [] as number[], treatment: [] as number[] }; + const serviceSamples = { baseline: [] as number[], treatment: [] as number[] }; + for (let round = 0; round < 7; round++) for (const arm of round % 2 ? ["treatment", "baseline"] as const : ["baseline", "treatment"] as const) { + let start = performance.now(); + await execute(arm === "baseline" ? before : after, arm === "baseline" ? db : treatmentDb); + if (round >= 2) samples[arm].push(performance.now() - start); + start = performance.now(); + await (arm === "baseline" ? baseline.issueService : issueService)(arm === "baseline" ? db : treatmentDb).list(fixture.companyId, filters); + if (round >= 2) serviceSamples[arm].push(performance.now() - start); + } + assert.deepEqual(await issueService(treatmentDb).list(fixture.companyId, filters), await baseline.issueService(db).list(fixture.companyId, filters), "Complete list response must match"); + const explain = async (query: ProjectionQuery, target = db) => (await target.$client.unsafe(`EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${query.sql}`, query.params as never[]))[0]!["QUERY PLAN"]; + results.push({ filters: rawFilters, readMode: readMode + (process.argv.includes("--read-one") ? "-one" : ""), jit, rows: afterRows.length, completePositionalEquality: true, serviceEquality: true, + baseline: { ...stats(samples.baseline), service: stats(serviceSamples.baseline), query: before, plan: await explain(before) }, + treatment: { ...stats(samples.treatment), service: stats(serviceSamples.treatment), query: after, plan: await explain(after, treatmentDb) } }); + if (arg("output")) await fs.writeFile(`${arg("output")}.partial.json`, JSON.stringify({ incomplete: true, baselineRevision, databaseName, treatmentDatabaseName, settings, treatmentSettings, results }, null, 2)); + } + const report = { baselineRevision, treatmentRevision: git(["rev-parse", "HEAD"]).trim(), sourceHashes: Object.fromEntries(await Promise.all(proofPaths.map(async file => [file, createHash("sha256").update(await fs.readFile(path.join(root, file))).digest("hex")]))), + changedDependencies, settings, treatmentSettings, indexes: { + baseline: await db.$client.unsafe("select indexname,indexdef,pg_relation_size(indexname::regclass)::text bytes from pg_indexes where tablename='activity_log' order by indexname"), + treatment: await treatmentDb.$client.unsafe("select indexname,indexdef,pg_relation_size(indexname::regclass)::text bytes from pg_indexes where tablename='activity_log' order by indexname"), + }, node: process.version, host: { ...host, loadAfter: loadavg() }, databaseName, treatmentDatabaseName, fixture: { ...fixture, records: undefined }, smokeMs, seedMs, totalMs: performance.now()-started, warmupsPerArm: 2, measuredSamplesPerArm: 5, results }; + if (arg("output")) await fs.writeFile(arg("output")!, JSON.stringify(report, null, 2)); + if (process.argv.includes("--assert-performance")) assertPerformance(report); + console.log(JSON.stringify({ ...report, results: results.map(result => ({ ...result, baseline: { ...result.baseline, query: undefined, plan: undefined }, treatment: { ...result.treatment, query: undefined, plan: undefined } })) }, null, 2)); +} finally { + clearInterval(keepAlive); + if (treatmentDb !== db) await treatmentDb.$client.end({ timeout: 0 }); + await db.$client.end({ timeout: 0 }); await admin.$client.end({ timeout: 0 }); + await fs.rm(baselineDir, { recursive: true, force: true }); +} diff --git a/server/src/__tests__/helpers/task-list-projection-fixture.ts b/server/src/__tests__/helpers/task-list-projection-fixture.ts new file mode 100644 index 00000000000..fce28ac92bd --- /dev/null +++ b/server/src/__tests__/helpers/task-list-projection-fixture.ts @@ -0,0 +1,104 @@ +import assert from "node:assert/strict"; +import { randomBytes, randomUUID } from "node:crypto"; +import { agents, chatConversations, chatEndpoints, companies, issues, toolApplications, toolConnections, type Db } from "@paperclipai/db"; +import type { issueService } from "../../services/issues.js"; + +export type ProjectionQuery = { sql: string; params: unknown[] }; +export type ListFactory = typeof issueService; +export type ListFilters = Parameters["list"]>[1]; + +export async function captureTaskListQuery(db: Db, factory: ListFactory, companyId: string, filters?: ListFilters) { + let captured: ProjectionQuery | undefined; + const stop = new Error("Task list query captured"); + const wrap = (target: object): typeof db => new Proxy(target, { get(target, key) { + if (key === "with") return (...ctes: unknown[]) => wrap(Reflect.apply(Reflect.get(target, key), target, ctes)); + if (key !== "select" && key !== "selectDistinctOn") return Reflect.get(target, key); + return (...fields: unknown[]) => { + const builder = Reflect.apply(Reflect.get(target, key), target, fields); + const from = builder.from.bind(builder); + builder.from = (...tables: unknown[]) => { + const query = from(...tables); + query.execute = async () => { captured = query.toSQL(); throw stop; }; + return query; + }; + return builder; + }; + } }) as typeof db; + const captureDb = wrap(db); + try { await factory(captureDb).list(companyId, filters); } + catch (error) { if (error !== stop) throw error; } + if (!captured) throw new Error("Real list caller did not generate a projection query"); + return captured; +} + +export async function seedTaskListProjectionFixture(db: Db, options: { count: number; chatCount: number; history: number; payloadBytes: number; liveCount?: number; activityPerIssue?: number; commentsPerIssue?: number; activityDecoys?: number; descriptionBytes?: number; hostedShape?: boolean }) { + if (options.hostedShape) assert.ok(options.count >= options.chatCount + 695, "Hosted shape needs chatCount + 695 tasks for all GitHub and removed-conversation controls"); + const companyId = randomUUID(), otherCompanyId = randomUUID(), agentId = randomUUID(); + await db.insert(companies).values([companyId, otherCompanyId].map(id => ({ id, name: "Projection fixture", issuePrefix: `P${id.slice(0, 7)}` }))); + await db.insert(agents).values({ id: agentId, companyId, name: "Projection agent", adapterType: "process" }); + const noise = randomBytes(options.payloadBytes).toString("base64"); + const records = await db.insert(issues).values(Array.from({ length: options.count }, (_, n) => ({ + companyId, title: `Projection task ${n}`, description: !options.descriptionBytes || n%3 === 0 ? null : n%3 === 1 ? "Projection résumé 世界" : `Projection résumé 世界 ${noise.slice(0, options.descriptionBytes)}`, status: n < options.chatCount ? "in_review" : "todo", assigneeAgentId: agentId, + createdByUserId: "projection-human", priority: n % 3 === 0 ? "high" : "medium", + originKind: n < options.chatCount ? "chat_channel" : "human", + createdAt: new Date("2026-01-01T00:00:00Z"), updatedAt: new Date(Date.parse("2026-01-01T00:00:00Z") + n * 1000), + }))).returning(); + const [application] = await db.insert(toolApplications).values({ companyId, name: "Projection chat", type: "native" }).returning(); + const [connection] = await db.insert(toolConnections).values({ companyId, applicationId: application!.id, name: "Projection chat", uid: randomUUID(), transport: "chat_sdk", connectionPurpose: "channel" }).returning(); + const [endpoint] = await db.insert(chatEndpoints).values({ companyId, connectionId: connection!.id, provider: "slack", publicId: randomUUID(), assignedAgentId: agentId, status: "active" }).returning(); + if (options.chatCount) await db.insert(chatConversations).values(records.slice(0, options.chatCount).map((issue, n) => ({ companyId, endpointId: endpoint!.id, issueId: issue.id, + externalConversationId: `channel-${n}`, externalThreadId: "thread", externalLabel: "#synthetic", state: options.hostedShape ? n === 0 ? "waiting" : "active" : n % 3 === 0 ? "active" : "waiting", createdAt: new Date("2026-01-01T00:00:00Z") }))); + if (options.hostedShape) { + const [githubConnection] = await db.insert(toolConnections).values({ companyId, applicationId: application!.id, name: "Projection reviews", uid: randomUUID(), transport: "chat_sdk", connectionPurpose: "channel" }).returning(); + const [github] = await db.insert(chatEndpoints).values({ companyId, connectionId: githubConnection!.id, provider: "github", publicId: randomUUID(), assignedAgentId: agentId, status: "active" }).returning(); + await db.insert(chatConversations).values(records.slice(options.chatCount, options.chatCount + 642).map((issue, n) => ({ companyId, endpointId: github!.id, issueId: issue.id, + externalConversationId: `github-${n}`, externalThreadId: "thread", externalLabel: "Synthetic review", state: "active", createdAt: new Date("2026-01-01") }))); + await db.insert(chatConversations).values(records.slice(options.chatCount + 642, options.chatCount + 642 + 53).map((issue, n) => ({ companyId, endpointId: endpoint!.id, issueId: issue.id, + externalConversationId: `removed-${n}`, externalThreadId: "thread", externalLabel: "#removed", state: "endpoint_removed", createdAt: new Date("2026-01-01") }))); + } + const wakeNoise = options.hostedShape ? randomBytes(1024).toString("base64") : noise; + await db.$client.unsafe(`insert into heartbeat_runs(company_id,agent_id,status,context_snapshot,native_issue_id,created_at,finished_at) + select $1::uuid,$2::uuid,'succeeded',case when n%3=0 then jsonb_build_object('issueId',id::text,'noise',$4::text) + when n%3=1 then jsonb_build_object('taskId',id::text,'noise',$4::text) else jsonb_build_object('noise',$4::text) end, + case when n%3=2 then id end,'2026-01-02'::timestamptz+n*interval '1 second','2026-01-03'::timestamptz + from unnest($3::uuid[]) task(id) cross join generate_series(1,$5::integer)n`, [companyId, agentId, records.map(row => row.id), noise, options.history]); + await db.$client.unsafe(`insert into agent_wakeup_requests(company_id,agent_id,source,status,payload,created_at) + select $1::uuid,$2::uuid,'synthetic','consumed',case when n%3=0 then jsonb_build_object('issueId',id::text,'noise',$4::text) + when n%3=1 then jsonb_build_object('taskId',id::text,'noise',$4::text) + else jsonb_build_object('_paperclipWakeContext',jsonb_build_object('issueId',id::text),'noise',$4::text) end,'2026-01-02'::timestamptz+n*interval '1 second' + from unnest($3::uuid[]) task(id) cross join generate_series(1,$5::integer)n`, [companyId, agentId, records.map(row => row.id), wakeNoise, options.hostedShape ? 28 : options.history]); + const liveCount = options.hostedShape ? 210 : options.liveCount ?? 0; + const liveIds = records.slice(options.hostedShape ? options.chatCount : 0).map(row => row.id); + if (liveCount) { + await db.$client.unsafe(`insert into heartbeat_runs(company_id,agent_id,status,context_snapshot,native_issue_id,created_at) + select $1::uuid,$2::uuid,case when $7::boolean then 'running' else case n%3 when 0 then 'queued' when 1 then 'running' else 'scheduled_retry' end end, + case n%3 when 0 then jsonb_build_object('issueId',id::text,'noise',$4::text) + when 1 then jsonb_build_object('taskId',id::text,'noise',$4::text) else jsonb_build_object('noise',$4::text) end, + case when n%3=2 then id end,'2026-01-04'::timestamptz + from generate_series(1,$5::integer)n join unnest($3::uuid[]) with ordinality task(id,ordinal) + on ordinal=1+(n%least($6::integer,12))`, [companyId, agentId, liveIds, noise, options.hostedShape ? 2 : liveCount, liveIds.length, options.hostedShape ?? false]); + await db.$client.unsafe(`insert into agent_wakeup_requests(company_id,agent_id,source,status,payload,created_at) + select $1::uuid,$2::uuid,'synthetic',case when $7::boolean then case when n<=139 then 'claimed' else 'deferred_issue_execution' end else case n%3 when 0 then 'queued' when 1 then 'claimed' else 'deferred_issue_execution' end end, + case n%3 when 0 then jsonb_build_object('issueId',id::text,'noise',$4::text) + when 1 then jsonb_build_object('taskId',id::text,'noise',$4::text) + else jsonb_build_object('_paperclipWakeContext',jsonb_build_object('issueId',id::text),'noise',$4::text) end, + '2026-01-04'::timestamptz + from generate_series(1,$5::integer)n join unnest($3::uuid[]) with ordinality task(id,ordinal) + on ordinal=1+(n%least($6::integer,12))`, [companyId, agentId, liveIds, wakeNoise, liveCount, liveIds.length, options.hostedShape ?? false]); + } + if (options.activityPerIssue) await db.$client.unsafe(`insert into activity_log(company_id,entity_type,entity_id,actor_type,actor_id,action,details,created_at) + select $1::uuid,'issue',id::text,'agent','synthetic-worker','issue.updated',jsonb_build_object('noise',left($3,4096)), + '2026-01-03'::timestamptz+n*interval '1 second' + from unnest($2::uuid[]) task(id) cross join generate_series(1,$4::integer)n order by n,id`, + [companyId, records.map(row => row.id), noise, options.activityPerIssue]); + if (options.commentsPerIssue) await db.$client.unsafe(`insert into issue_comments(company_id,issue_id,author_type,author_user_id,body,created_at) + select $1::uuid,id,'user','synthetic-reader',left($3,4096),'2026-01-05'::timestamptz+n*interval '1 second' + from unnest($2::uuid[]) task(id) cross join generate_series(1,$4::integer)n order by n,id`, + [companyId, records.map(row => row.id), noise, options.commentsPerIssue]); + if (options.activityDecoys) await db.$client.unsafe(`insert into activity_log(company_id,entity_type,entity_id,actor_type,actor_id,action,details,created_at) + select $1::uuid,'heartbeat_run',gen_random_uuid()::text,'agent','synthetic-worker','heartbeat.run.created', + jsonb_build_object('noise',left($2,256)),'2026-01-03'::timestamptz+n*interval '1 second' + from generate_series(1,$3::integer)n`, [companyId, noise, options.activityDecoys]); + await db.$client.unsafe("ANALYZE"); + return { companyId, otherCompanyId, agentId, endpointId: endpoint!.id, records, options, payloadBytes: Buffer.byteLength(noise) }; +} diff --git a/server/src/__tests__/task-list-projection-query.test.ts b/server/src/__tests__/task-list-projection-query.test.ts new file mode 100644 index 00000000000..3b9ec9c90ee --- /dev/null +++ b/server/src/__tests__/task-list-projection-query.test.ts @@ -0,0 +1,69 @@ +import { sql } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { activityLog, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, issueComments, issues } from "@paperclipai/db"; +import { issueService } from "../services/issues.js"; +import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { captureTaskListQuery, seedTaskListProjectionFixture, type ListFilters } from "./helpers/task-list-projection-fixture.js"; + +describe("main task-list canonical activity index", () => { + let database: Awaited>; + let db: ReturnType; + beforeAll(async () => { database = await startEmbeddedPostgresTestDatabase("task-projection-"); db = createDb(database.connectionString); }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); + afterAll(async () => { await db?.$client.end({ timeout: 0 }); await database?.cleanup(); }); + + it("stops each latest-activity probe for full and index projections at one qualifying row", async () => { + const f = await seedTaskListProjectionFixture(db, { count: 1200, chatCount: 6, history: 0, payloadBytes: 3072, descriptionBytes: 4096, activityPerIssue: 8, activityDecoys: 1000 }); + const cases: ListFilters[] = [{ limit: 100, sortField: "updated", sortDir: "desc" }, { limit: 100, listProjection: "index" }]; + for (const filters of cases) { + const query = await captureTaskListQuery(db, issueService, f.companyId, filters); + const [explain] = await db.$client.unsafe(`explain (analyze, buffers, format json) ${query.sql}`, query.params as never[]); + const inspected: { rows: number; loops: number }[] = []; + const visit = (node: Record) => { + if (node["Relation Name"] === "activity_log") inspected.push({ rows: Number(node["Actual Rows"]), loops: Number(node["Actual Loops"]) }); + for (const child of node.Plans as Record[] ?? []) visit(child); + }; + visit(explain["QUERY PLAN"][0].Plan); + expect(inspected.length).toBeGreaterThan(0); + expect(Math.max(...inspected.map(row => row.rows))).toBeLessThanOrEqual(1); + expect(Math.max(...inspected.map(row => row.loops))).toBeGreaterThanOrEqual(1000); + expect(await issueService(db).list(f.companyId, filters)).toHaveLength(100); + } + }); + + it("rejects incomplete hosted conversation controls before writing fixture rows", async () => { + const [before] = await db.$client.unsafe("select count(*)::int as count from companies"); + await expect(seedTaskListProjectionFixture(db, { count: 1200, chatCount: 600, history: 0, payloadBytes: 0, hostedShape: true })).rejects.toThrow("Hosted shape needs chatCount + 695 tasks"); + const [after] = await db.$client.unsafe("select count(*)::int as count from companies"); + expect(after!.count).toBe(before!.count); + }); + + it("preserves canonical activity exclusions, text bindings, ties and all list order modes", async () => { + const f = await seedTaskListProjectionFixture(db, { count: 8, chatCount: 0, history: 0, payloadBytes: 0 }); + const id = (n: number) => f.records[n]!.id; + const log = (n: number, day: number, extra: Record = {}) => ({ companyId: f.companyId, actorType: "user", actorId: "reader", entityType: "issue", entityId: id(n), action: "issue.updated", createdAt: new Date(`2026-01-${String(day).padStart(2, "0")}T00:00:00Z`), ...extra }); + await db.insert(activityLog).values([ + log(0, 3), + ...["issue.read_marked", "issue.read_unmarked", "issue.inbox_archived", "issue.inbox_unarchived"].map(action => log(0, 20, { action })), + log(0, 21, { action: "issue.recovery_action_resolved", details: { source: "source_revalidation", trigger: "read_projection" } }), + log(1, 22, { entityType: "heartbeat_run" }), + log(2, 5), log(2, 6, { action: "issue.recovery_action_resolved", details: { source: "source_revalidation", trigger: "explicit_decision" } }), + log(4, 23, { entityId: id(4).toUpperCase() }), + log(4, 24, { entityId: `{${id(4)}}` }), + log(4, 25, { entityId: "not-a-task" }), + log(5, 26, { companyId: f.otherCompanyId }), + log(6, 7, { action: "issue.recovery_action_resolved", details: null }), log(7, 7), + ] as never); + await db.insert(issueComments).values({ companyId: f.companyId, issueId: id(1), authorUserId: "reader", body: "Canonical comment", createdAt: new Date("2026-01-04") }); + const svc = issueService(db), descending = [7, 6, 2, 1, 0, 5, 4, 3].map(id); + expect((await svc.list(f.companyId, { sortField: "updated", sortDir: "desc" })).map(row => row.id)).toEqual(descending); + expect((await svc.list(f.companyId, { sortField: "updated", sortDir: "asc" })).map(row => row.id)).toEqual([...descending].reverse()); + expect((await svc.list(f.companyId)).map(row => row.id)).toEqual([6, 0, 3, 7, 2, 1, 5, 4].map(id)); + expect((await svc.list(f.companyId, { sortField: "updated", sortDir: "desc", limit: 2, offset: 2 })).map(row => row.id)).toEqual(descending.slice(2, 4)); + expect((await svc.list(f.companyId, { sortField: "updated", sortDir: "desc", limit: 2, readCondition: sql`${issues.id} <> ${id(7)}` })).map(row => row.id)).toEqual([id(6), id(2)]); + expect((await svc.list(f.companyId, { sortField: "updated", sortDir: "desc", issueIds: [id(0), id(2)] })).map(row => row.id)).toEqual([id(2), id(0)]); + expect(await svc.list(f.companyId, { readCondition: sql`false`, limit: 2 })).toEqual([]); + await db.update(issues).set({ parentId: id(0) }).where(sql`${issues.id} in (${id(6)},${id(7)})`); + expect((await svc.list(f.companyId, { sortField: "updated", sortDir: "desc", oneChildPerParent: true })).map(row => row.id)).toEqual([id(7), id(2)]); + }); + +});