diff --git a/server/src/__tests__/blocker-workspace-history-performance.test.ts b/server/src/__tests__/blocker-workspace-history-performance.test.ts new file mode 100644 index 00000000000..c40a4b46f1e --- /dev/null +++ b/server/src/__tests__/blocker-workspace-history-performance.test.ts @@ -0,0 +1,103 @@ +import { randomUUID } from "node:crypto"; +import { sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { applyPendingMigrations, companies, createDb, issues } from "@paperclipai/db"; +import { issueService } from "../services/issues.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_BLOCKER_HISTORY_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); +type Plan = { "Node Type": string; "Relation Name"?: string; "Actual Rows": number; "Actual Loops": number; + "Rows Removed by Filter"?: number; "Rows Removed by Index Recheck"?: number; Plans?: Plan[] }; + +describe.skipIf(!support.supported)("blocker workspace retained history cost", () => { + let temporary: Awaited> | undefined; + let db: ReturnType; + let capture = false; + const queries: Array<{ query: string; parameters: unknown[] }> = []; + beforeAll(async () => { + if (external) await applyPendingMigrations(external); + else temporary = await startEmbeddedPostgresTestDatabase("blocker-history-"); + db = createDb(external ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + db.$client.options.onnotice = () => {}; + db.$client.options.debug = (_connection, query, parameters) => { + if (capture && query.includes("workspace_operations")) queries.push({ query, parameters: [...parameters] }); + }; + }, 30_000); + afterEach(async () => { capture = false; queries.length = 0; await db.execute(sql`truncate companies cascade`); }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); + + it.each(["finalized", "pending"] as const)("bounds operation-history work for 2100 blockers sharing 5000 %s operations", async kind => { + const companyId = randomUUID(), workspaceId = randomUUID(), unrelatedWorkspaceId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Shared history", issuePrefix: `SH${companyId.slice(0, 6)}` }); + const [root] = await db.insert(issues).values({ companyId, title: "History dependent", status: "blocked" }).returning(); + await db.execute(sql`insert into execution_workspaces (id, company_id, name, mode, strategy_type) + values (${workspaceId}::uuid, ${companyId}::uuid, 'Shared operation history', 'shared', 'project_primary'), + (${unrelatedWorkspaceId}::uuid, ${companyId}::uuid, 'Unrelated history', 'shared', 'project_primary')`); + await db.execute(sql`insert into workspace_operations (company_id, execution_workspace_id, phase, status, started_at) + select ${companyId}::uuid, ${unrelatedWorkspaceId}::uuid, 'worktree_prepare', 'succeeded', + '2025-01-01T00:00:00Z'::timestamptz + n * interval '1 second' from generate_series(1, 50000) n`); + await db.execute(sql`insert into workspace_operations (company_id, execution_workspace_id, phase, status, started_at) + select ${companyId}::uuid, ${workspaceId}::uuid, + case when n = 5000 and ${kind} = 'finalized' then 'workspace_finalize' else 'worktree_prepare' end, + 'succeeded', '2026-01-01T00:00:00Z'::timestamptz + n * interval '1 second' from generate_series(1, 5000) n`); + await db.execute(sql`with history as ( + insert into issues (company_id, title, status, execution_workspace_id) + select ${companyId}::uuid, 'done blocker ' || n, 'done', ${workspaceId}::uuid + from generate_series(1, 2100) n returning id) + insert into issue_relations (company_id, issue_id, related_issue_id, type) + select ${companyId}::uuid, id, ${root.id}::uuid, 'blocks' from history`); + await db.execute(sql`analyze workspace_operations`); + await db.execute(sql`analyze issues`); + await db.execute(sql`analyze issue_relations`); + capture = true; + const batch = await issueService(db).listBlockerAttentionBatch(companyId, [root]); + capture = false; + expect(queries.length).toBeGreaterThan(0); + expect(Math.max(...queries.map(statement => statement.parameters.length))).toBeLessThanOrEqual(1100); + let operationRowsExamined = 0; + const measurements: unknown[] = []; + for (const statement of queries) { + const result = await db.$client.unsafe("EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) " + statement.query, + statement.parameters as never[]); + const explain = (result[0]!["QUERY PLAN"] as Array<{ Plan: Plan }>)[0]!; + const visit = (node: Plan) => { + if (node["Relation Name"] === "workspace_operations") operationRowsExamined += node["Actual Loops"] + * (node["Actual Rows"] + (node["Rows Removed by Filter"] ?? 0) + (node["Rows Removed by Index Recheck"] ?? 0)); + node.Plans?.forEach(visit); + }; + visit(explain.Plan); + measurements.push({ ...statement, explain }); + } + process.stdout.write(JSON.stringify({ sharedWorkspaceHistory: { kind, blockers: 2100, retainedOperations: 5000, unrelatedOperations: 50000, + operationRowsExamined, queryCount: queries.length, measurements } }) + "\n"); + // The finite graph/pair pages may each read the 5,000 candidate rows once. + // Neither 2,100 blockers nor 50,000 unrelated operations may multiply that work. + expect(operationRowsExamined).toBeLessThanOrEqual(55_000); + expect(batch.deferredRootIds.has(root.id)).toBe(kind === "pending"); + if (kind === "finalized") expect(batch.nodeCount).toBe(1); + }, 120_000); + it("preserves attributed precedence and the strict later-finalize boundary", async () => { + const companyId = randomUUID(), workspaceId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Finalize policy", issuePrefix: `FP${companyId.slice(0, 6)}` }); + await db.execute(sql`insert into execution_workspaces (id, company_id, name, mode, strategy_type) + values (${workspaceId}::uuid, ${companyId}::uuid, 'Policy workspace', 'shared', 'project_primary')`); + const roots = await db.insert(issues).values(["older", "equal", "fallback"].map(title => ({ companyId, title, status: "blocked" }))).returning(); + const blockers = await db.insert(issues).values(roots.map(root => ({ companyId, title: root.title + " blocker", status: "done", executionWorkspaceId: workspaceId }))).returning(); + for (let index = 0; index < roots.length; index += 1) { + await db.execute(sql`insert into issue_relations (company_id, issue_id, related_issue_id, type) + values (${companyId}::uuid, ${blockers[index]!.id}::uuid, ${roots[index]!.id}::uuid, 'blocks')`); + } + await db.execute(sql`insert into workspace_operations (company_id, execution_workspace_id, issue_id, phase, status, started_at) values + (${companyId}::uuid, ${workspaceId}::uuid, ${blockers[0]!.id}::uuid, 'worktree_prepare', 'succeeded', '2026-01-01T00:00:01Z'), + (${companyId}::uuid, ${workspaceId}::uuid, ${blockers[1]!.id}::uuid, 'worktree_prepare', 'succeeded', '2026-01-01T00:00:02Z'), + (${companyId}::uuid, ${workspaceId}::uuid, null, 'workspace_finalize', 'succeeded', '2026-01-01T00:00:02Z'), + (${companyId}::uuid, ${workspaceId}::uuid, null, 'worktree_prepare', 'succeeded', '2026-01-01T00:00:03Z')`); + const service = issueService(db); + expect((await service.listBlockerAttentionBatch(companyId, [roots[0]!])).nodeCount).toBe(1); + expect((await service.listBlockerAttentionBatch(companyId, [roots[1]!])).nodeCount).toBe(2); + expect((await service.listBlockerAttentionBatch(companyId, [roots[2]!])).nodeCount).toBe(2); + }); + +}); diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index a1c28f296c2..c7b9f8aebb2 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -2518,24 +2518,55 @@ export async function listUnfinalizedExecutionWorkspaceIds( return unfinalized; } -/** The blocker keeps its own attributed operation barrier; legacy unattributed - * work is a fallback, and a later successful workspace export clears either. */ +/** Workspace-wide facts are shared by blockers, rather than rereading retained + * operation history for every issue pointing at the same workspace. The ordered + * OFFSET boundary keeps the rare finalize filter from turning into a company + * history scan; the existing workspace/started-at index can stream candidates. */ +function blockerWorkspaceFacts(companyId: string, workspaceScope: SQL) { + return sql`( + with candidate_workspaces as materialized ( + select distinct workspace_id from (${workspaceScope}) scoped_workspaces(workspace_id) + ), workspace_facts as materialized ( + select candidate.workspace_id, finalized.started_at as finalized_at, + legacy.phase, legacy.status, legacy.started_at + from candidate_workspaces candidate + left join lateral ( + select started_at from ( + select phase, status, started_at from workspace_operations + where company_id = ${companyId}::uuid + and execution_workspace_id = candidate.workspace_id + order by started_at desc offset 0 + ) workspace_history + where phase = 'workspace_finalize' and status = 'succeeded' + order by started_at desc limit 1 + ) finalized on true + left join lateral ( + select phase, status, started_at from workspace_operations + where company_id = ${companyId}::uuid + and execution_workspace_id = candidate.workspace_id and issue_id is null + order by started_at desc limit 1 + ) legacy on true + ) select * from workspace_facts + ) as blocker_workspace_facts`; +} + +/** An issue's own operation takes precedence over the legacy NULL fallback. + * Only a strictly later successful finalize clears either barrier. */ function pendingFinalizeBlockerCondition(companyId: string, issueId: SQL, workspaceId: SQL) { return sql`${workspaceId} is not null and exists ( - select 1 from lateral ( + select 1 from (values (1)) seed(value) + left join lateral ( select phase, status, started_at from workspace_operations blocker_operation where blocker_operation.company_id = ${companyId}::uuid and blocker_operation.execution_workspace_id = ${workspaceId} - and (blocker_operation.issue_id = ${issueId} or blocker_operation.issue_id is null) - order by (blocker_operation.issue_id is not null) desc, blocker_operation.started_at desc - limit 1 - ) latest_blocker_operation - where not (latest_blocker_operation.phase = 'workspace_finalize' and latest_blocker_operation.status = 'succeeded') - and not exists (select 1 from workspace_operations later_finalize - where later_finalize.company_id = ${companyId}::uuid - and later_finalize.execution_workspace_id = ${workspaceId} - and later_finalize.phase = 'workspace_finalize' and later_finalize.status = 'succeeded' - and later_finalize.started_at > latest_blocker_operation.started_at) + and blocker_operation.issue_id = ${issueId} + order by blocker_operation.started_at desc limit 1 + ) latest_blocker_operation on true + where coalesce(latest_blocker_operation.started_at, blocker_workspace_facts.started_at) is not null + and not (coalesce(latest_blocker_operation.phase, blocker_workspace_facts.phase) = 'workspace_finalize' + and coalesce(latest_blocker_operation.status, blocker_workspace_facts.status) = 'succeeded') + and (blocker_workspace_facts.finalized_at is null + or blocker_workspace_facts.finalized_at <= coalesce(latest_blocker_operation.started_at, blocker_workspace_facts.started_at)) )`; } @@ -2558,10 +2589,13 @@ async function listPendingFinalizeBlockerIssueIds( .for("share", { noWait: true }); } } - for (const chunk of chunkList(pairs, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) { + // Each pair occurs in both the source and the workspace scope. + for (const chunk of chunkList(pairs, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE / 2)) { const values = sql.join(chunk.map(pair => sql`(${pair.blockerIssueId}::uuid, ${pair.executionWorkspaceId}::uuid)`), sql`, `); const rows = await dbOrTx.select({ id: sql`blocker_source.issue_id` }) .from(sql`(values ${values}) as blocker_source(issue_id, workspace_id)`) + .leftJoin(blockerWorkspaceFacts(companyId, sql`select workspace_id from (values ${values}) workspace_scope(issue_id, workspace_id)`), + sql`blocker_workspace_facts.workspace_id = blocker_source.workspace_id`) .where(pendingFinalizeBlockerCondition(companyId, sql`blocker_source.issue_id`, sql`blocker_source.workspace_id`)); for (const row of rows) pending.add(row.id); } @@ -3829,6 +3863,17 @@ async function readIssueBlockerAttentionBatch( // below remains authoritative, without a repeated company-index scan. .from(issues).where(eq(issues.id, issueRelations.issueId)) .limit(1).as("explicit_blocker"); + const workspaceScope = sql`select distinct candidate.execution_workspace_id + from issue_relations workspace_relation + inner join lateral ( + select company_id, status, execution_workspace_id from issues + where id = workspace_relation.issue_id limit 1 + ) candidate on true + where workspace_relation.company_id = ${companyId}::uuid + and workspace_relation.type = 'blocks' + and workspace_relation.related_issue_id in (${sql.join(chunk.map(id => sql`${id}::uuid`), sql`, `)}) + and candidate.company_id = ${companyId}::uuid + and candidate.status = 'done' and candidate.execution_workspace_id is not null`; const explicitBlockerRowsPromise: Promise< IssueBlockerAttentionQueryRow[] > = dbOrTx @@ -3848,6 +3893,8 @@ async function readIssueBlockerAttentionBatch( }) .from(issueRelations) .innerJoinLateral(explicitBlocker, sql`true`) + .leftJoin(blockerWorkspaceFacts(companyId, workspaceScope), + sql`blocker_workspace_facts.workspace_id = ${explicitBlocker.executionWorkspaceId}`) .where( and( eq(issueRelations.companyId, companyId),