perf(synthesis): fan dynamic-dispatch passes across the resolver pool, byte-identical graphs (#1321)

The ~36 independent synthesis passes (callback/event/framework wiring) ran
sequentially on the indexer's main thread — 2.0s of a 4,402-file Java repo's
index, and the stage where kernel-class repos die (#1212). They now live in
an explicit registry (SYNTH_PASSES) and, when the resolver pool is alive
(>=150k-ref repos), fan out across its read-only workers: dubbo synthesis
2,024ms -> ~900ms (-55%), total fresh init 13.5s -> 11.9s. Graphs verified
byte-for-byte identical on both the pool path (dubbo) and the sequential
path (excalidraw).

Why this is safe: no pass's edges persist until the ordered merge, so every
pass sees the same committed post-resolution DB state in either mode, and
results merge in registry order regardless of completion order — the
first-seen dedup is unchanged. The pool now survives through synthesis
(destroy moved after it) instead of being torn down moments before the one
stage that could reuse it.

Robustness: a pass that fails on a worker (crash, OOM) is retried on the
main thread — a synthesizer blow-up now costs one worker instead of the
whole index, which is half the #1212 story on very large repos.

Also: ref-row cleanup deletes now run as one transaction with a cached
statement instead of one implicit commit per 500-row chunk (mechanically
fewer WAL commits; matters most on HDD-class storage). A set-based rewrite
of failed-ref parking was tried, measured ~zero on NVMe, and dropped — the
remaining persist cost is edge-index B-tree maintenance, not statement
dispatch.

SYNTH_PROGRESS_STEPS now derives from the registry (passes + fixed marks);
the pin test counts registry entries plus literal __mark sites.

Suite green (2444). Sequential-path timing unchanged on excalidraw.

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

src/resolution/resolver-pool.ts

@@ -13,9 +13,15 @@ import { Worker } from 'worker_threads';
 import * as fs from 'fs';
 import * as path from 'path';
 import * as os from 'os';
-import type { UnresolvedReference } from '../types';
+import type { Edge, UnresolvedReference } from '../types';
 import type { ResolvedRef, UnresolvedRef } from './types';
 
+/** One synthesis pass's output: its edge list + worker-measured wall clock. */
+export interface SynthPassResult {
+  edges: Edge[];
+  ms: number;
+}
+
 export interface ChunkResult {
   resolved: ResolvedRef[];
   unresolved: UnresolvedRef[];
@@ -55,6 +61,7 @@ export class ResolverPool {
   private workers: PoolWorker[] = [];
   private nextId = 0;
   private waiters = new Map<number, { resolve: (r: ChunkResult) => void; reject: (e: Error) => void }>();
+  private synthWaiters = new Map<number, { resolve: (r: SynthPassResult) => void; reject: (e: Error) => void }>();
   private failed: Error | null = null;
 
   /**
@@ -85,7 +92,7 @@ export class ResolverPool {
         readyReject = reject;
       });
       const pw: PoolWorker = { worker, ready, busy: 0 };
-      worker.on('message', (msg: { type: string; id?: number; message?: string } & Partial<ChunkResult>) => {
+      worker.on('message', (msg: { type: string; id?: number; message?: string; edges?: Edge[]; ms?: number } & Partial<ChunkResult>) => {
         if (msg.type === 'ready') {
           readyResolve();
         } else if (msg.type === 'result' && msg.id !== undefined) {
@@ -99,6 +106,11 @@ export class ResolverPool {
             deferredThisMember: msg.deferredThisMember!,
             byMethod: msg.byMethod!,
           });
+        } else if (msg.type === 'synth-result' && msg.id !== undefined) {
+          pw.busy--;
+          const waiter = this.synthWaiters.get(msg.id);
+          this.synthWaiters.delete(msg.id);
+          waiter?.resolve({ edges: msg.edges ?? [], ms: msg.ms ?? 0 });
         } else if (msg.type === 'error') {
           pw.busy--;
           const err = new Error(`resolver worker: ${msg.message}`);
@@ -106,6 +118,10 @@ export class ResolverPool {
             const waiter = this.waiters.get(msg.id)!;
             this.waiters.delete(msg.id);
             waiter.reject(err);
+          } else if (msg.id !== undefined && this.synthWaiters.has(msg.id)) {
+            const waiter = this.synthWaiters.get(msg.id)!;
+            this.synthWaiters.delete(msg.id);
+            waiter.reject(err);
           } else {
             this.fail(err);
           }
@@ -130,6 +146,8 @@ export class ResolverPool {
     if (!this.failed) this.failed = err;
     for (const [, waiter] of this.waiters) waiter.reject(this.failed);
     this.waiters.clear();
+    for (const [, waiter] of this.synthWaiters) waiter.reject(this.failed);
+    this.synthWaiters.clear();
   }
 
   /** Whether this batch is worth fanning out. */
@@ -175,6 +193,23 @@ export class ResolverPool {
     return out;
   }
 
+  /**
+   * Run one synthesis pass (by SYNTH_PASSES name) on the least-busy worker.
+   * The worker reads the committed graph on its own connection and returns
+   * the pass's edge list; the caller merges in canonical order. Rejects on
+   * worker failure — the caller retries the pass on the main thread.
+   */
+  async runSynthPass(passName: string): Promise<SynthPassResult> {
+    if (this.failed) throw this.failed;
+    const id = this.nextId++;
+    const pw = this.workers.reduce((a, b) => (b.busy < a.busy ? b : a));
+    pw.busy++;
+    return new Promise<SynthPassResult>((resolve, reject) => {
+      this.synthWaiters.set(id, { resolve, reject });
+      pw.worker.postMessage({ type: 'synth', id, pass: passName });
+    });
+  }
+
   async destroy(): Promise<void> {
     await Promise.all(
       this.workers.map(