Merge qualified retention performance repair

Co-Authored-By: Paperclip <noreply@paperclip.ing>

* commit '57f914747c0d2aefaca9ceccdeeddb55489dad67':
  Bound blocker workspace history reads by candidate workspace

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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
+  let db: ReturnType<typeof createDb>;
+  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);
+  });
+
+});

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<string>`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),