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)]); + }); + +});