codegraph/__tests__/parse-pool.test.ts

/**
 * ParseWorkerPool — the worker pool that parses files across cores during a full
 * `codegraph index` (issue #1015). These tests drive the pool's queue / growth /
 * recycle / crash-recovery / timeout / teardown logic with INJECTED fake
 * workers, so they exercise the real scheduling code without spawning threads or
 * needing a built dist.
 *
 * End-to-end behavior with real worker threads (each worker owns a tree-sitter
 * WASM heap and runs extractFromSource) is covered by the extraction suite
 * against a real temp project; here we pin the orchestration that makes the
 * parallelism safe.
 */
import { describe, it, expect } from 'vitest';
import { ParseWorkerPool, resolveParseBudgetMs, resolveParsePoolSize, resolveParseTimeoutMs, type ParsePoolWorker, type ParseTask } from '../src/extraction/parse-pool';
import type { Language, ExtractionResult } from '../src/types';

const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

interface ParseMsg { type: 'parse'; id: number; filePath: string; content: string; language: Language }
type Action = { result: ExtractionResult } | { crash: true } | { hang: true } | { wait: Promise<ExtractionResult> };

/**
 * Fake worker speaking the same {load-grammars → grammars-loaded} /
 * {parse → parse-result} protocol as the real parse-worker. `behavior` decides
 * per parse whether to return a result, crash (exit≠0), hang (never reply —
 * exercises the timeout), or wait on a promise (hold a parse in-flight to
 * observe concurrency). Emits 'grammars-loaded' on a macrotask so the pool has
 * wired its listeners first.
 */
class FakeWorker implements ParsePoolWorker {
  private msgCb?: (m: unknown) => void;
  private exitCb?: (code: number) => void;
  alive = true;
  constructor(private behavior: (m: ParseMsg) => Action, private onTerminate?: () => void) {}
  on(event: string, cb: (...args: any[]) => void): void {
    if (event === 'message') this.msgCb = cb;
    else if (event === 'exit') this.exitCb = cb;
    // 'error' unused by the fakes
  }
  private reply(id: number, result: ExtractionResult): void {
    if (this.alive) this.msgCb?.({ type: 'parse-result', id, result });
  }
  postMessage(msg: unknown): void {
    const m = msg as { type: string } & Partial<ParseMsg>;
    if (m.type === 'load-grammars') {
      setTimeout(() => { if (this.alive) this.msgCb?.({ type: 'grammars-loaded' }); }, 0);
      return;
    }
    if (m.type !== 'parse') return;
    const action = this.behavior(m as ParseMsg);
    if ('crash' in action) {
      this.alive = false;
      setTimeout(() => this.exitCb?.(1), 0); // simulate a WASM-OOM exit(1)
      return;
    }
    if ('hang' in action) return; // never reply → timeout path
    if ('wait' in action) { void action.wait.then((r) => this.reply(m.id!, r)); return; }
    setTimeout(() => this.reply(m.id!, action.result), 0);
  }
  terminate(): Promise<number> { this.alive = false; this.onTerminate?.(); return Promise.resolve(0); }
}

const task = (filePath: string, content = 'code'): ParseTask => ({ filePath, content, language: 'typescript' as Language });
const result = (tag = 0): ExtractionResult => ({ nodes: [], edges: [], unresolvedReferences: [], errors: [], durationMs: tag });

/** Build a pool with a counting fake-worker factory. */
function makePool(
  size: number,
  behavior: (m: ParseMsg) => Action,
  opts: Partial<{ recycleInterval: number; parseTimeoutMs: number }> = {}
) {
  let spawned = 0, terminated = 0;
  const pool = new ParseWorkerPool({
    languages: ['typescript'] as Language[],
    size,
    recycleInterval: opts.recycleInterval,
    parseTimeoutMs: opts.parseTimeoutMs,
    createWorker: () => { spawned++; return new FakeWorker(behavior, () => { terminated++; }); },
  });
  return { pool, counts: () => ({ spawned, terminated }) };
}

describe('resolveParseTimeoutMs', () => {
  it('honors a positive numeric override (CODEGRAPH_PARSE_TIMEOUT_MS)', () => {
    expect(resolveParseTimeoutMs('30000')).toBe(30000);
    expect(resolveParseTimeoutMs('1500.9')).toBe(1500);
  });
  it('falls back to the 10s default when unset/blank/non-numeric/non-positive', () => {
    expect(resolveParseTimeoutMs(undefined)).toBe(10_000);
    expect(resolveParseTimeoutMs('')).toBe(10_000);
    expect(resolveParseTimeoutMs('abc')).toBe(10_000);
    expect(resolveParseTimeoutMs('0')).toBe(10_000);
    expect(resolveParseTimeoutMs('-5')).toBe(10_000);
  });
});

describe('resolveParseBudgetMs', () => {
  it('caps near-limit blob headers at a 20s soft / 60s hard window (#1555)', () => {
    expect(resolveParseBudgetMs(10_000, 940_800)).toBe(20_000);
    expect(resolveParseBudgetMs(10_000, 857_376)).toBe(20_000);
  });

  it('does not clamp an explicit larger base timeout', () => {
    expect(resolveParseBudgetMs(45_000, 940_800)).toBe(45_000);
  });
});

describe('resolveParsePoolSize', () => {
  it('treats explicit 0 and 1 as a single worker (the rollback path)', () => {
    expect(resolveParsePoolSize('0', 8)).toBe(1);
    expect(resolveParsePoolSize('1', 8)).toBe(1);
  });
  it('honors a numeric override, capped at the hard ceiling', () => {
    expect(resolveParsePoolSize('4', 8)).toBe(4);
    expect(resolveParsePoolSize('999', 8)).toBe(16);
  });
  it('defaults to clamp(cores-1, 1, 8) when unset/blank/non-numeric', () => {
    expect(resolveParsePoolSize(undefined, 8)).toBe(7);
    expect(resolveParsePoolSize('', 8)).toBe(7);
    expect(resolveParsePoolSize('abc', 8)).toBe(7);
    expect(resolveParsePoolSize(undefined, 1)).toBe(1);   // never zero
    expect(resolveParsePoolSize(undefined, 2)).toBe(1);   // leave a core
    expect(resolveParsePoolSize(undefined, 64)).toBe(8);  // never above the default cap
  });
});

describe('ParseWorkerPool', () => {
  it('parses a file and returns the worker result', async () => {
    const { pool } = makePool(1, () => ({ result: result(42) }));
    const res = await pool.requestParse(task('a.ts'));
    expect(res.durationMs).toBe(42);
    await pool.destroy();
  });

  it('runs N parses in parallel across the pool (not serialized)', async () => {
    let active = 0, maxActive = 0;
    let release!: () => void;
    const gate = new Promise<void>((r) => { release = r; });
    const { pool } = makePool(4, () => ({
      wait: (async () => { active++; maxActive = Math.max(maxActive, active); await gate; active--; return result(); })(),
    }));
    const ps = [0, 1, 2, 3].map((i) => pool.requestParse(task(`f${i}.ts`)));
    await sleep(60); // let the pool grow to size and dispatch all four
    expect(maxActive).toBe(4);
    release();
    await Promise.all(ps);
    await pool.destroy();
  });

  it('grows lazily — a single parse does not spawn the whole pool', async () => {
    const { pool, counts } = makePool(8, () => ({ result: result() }));
    await pool.requestParse(task('only.ts'));
    expect(counts().spawned).toBe(1); // just the eager warm worker
    await pool.destroy();
  });

  it('recycles a worker after recycleInterval parses', async () => {
    const { pool, counts } = makePool(1, () => ({ result: result() }), { recycleInterval: 3 });
    for (let i = 0; i < 4; i++) await pool.requestParse(task(`f${i}.ts`));
    // 3 parses on the first worker → recycle (terminate + respawn); the 4th runs
    // on the fresh worker.
    expect(counts().spawned).toBe(2);
    expect(counts().terminated).toBeGreaterThanOrEqual(1);
    await pool.destroy();
  });

  it('rejects a parse whose worker crashes (retry-pass-recognisable message) and keeps serving', async () => {
    const { pool, counts } = makePool(1, (m) => (m.filePath === 'poison.ts' ? { crash: true } : { result: result(7) }));
    // The message must contain "Worker exited" so the orchestrator's retry pass
    // re-attempts it (that's the filter it uses).
    await expect(pool.requestParse(task('poison.ts'))).rejects.toThrow(/Worker exited/);
    const ok = await pool.requestParse(task('good.ts'));
    expect(ok.durationMs).toBe(7);
    expect(counts().spawned).toBe(2); // respawned after the crash
    await pool.destroy();
  });

  it('times out a hung parse (at the hard-kill backstop) and stays usable', async () => {
    const { pool } = makePool(1, (m) => (m.filePath === 'hang.ts' ? { hang: true } : { result: result(9) }), { parseTimeoutMs: 30 });
    const t0 = Date.now();
    // The base timer (30ms) only marks the job late; the kill happens at the
    // 3× backstop (90ms), and the message carries the full window.
    await expect(pool.requestParse(task('hang.ts'))).rejects.toThrow(/timed out after 90ms/i);
    expect(Date.now() - t0).toBeGreaterThanOrEqual(80);
    const ok = await pool.requestParse(task('ok.ts'));
    expect(ok.durationMs).toBe(9);
    await pool.destroy();
  });

  it('accepts a result that arrives after the base timeout instead of killing the worker (#1231 false-timeout fix)', async () => {
    // Simulates the HDD stall: the parse "finished" but its result is only
    // delivered after the base timer fired. Old behaviour killed the worker and
    // rejected; now the late result is accepted and the worker keeps serving.
    const { pool, counts } = makePool(
      1,
      (m) => (m.filePath === 'late.ts' ? { wait: sleep(80).then(() => result(11)) } : { result: result(9) }),
      { parseTimeoutMs: 50 }
    );
    const res = await pool.requestParse(task('late.ts')); // base timer 50ms < delivery 80ms < backstop 150ms
    expect(res.durationMs).toBe(11);
    expect(counts().terminated).toBe(0); // no kill…
    expect(counts().spawned).toBe(1);    // …and no respawn churn
    const ok = await pool.requestParse(task('next.ts'));
    expect(ok.durationMs).toBe(9); // same worker still serving
    await pool.destroy();
  });

  it('forwards pre-read grammar WASM bytes to every spawned worker (#1231 respawn I/O fix)', async () => {
    const grammarBuffers = { typescript: new Uint8Array([1, 2, 3]) };
    const loadMsgs: Array<{ grammarBuffers?: Record<string, Uint8Array> }> = [];
    let worker!: FakeWorker;
    const pool = new ParseWorkerPool({
      languages: ['typescript'] as Language[],
      size: 1,
      grammarBuffers,
      createWorker: () => {
        worker = new FakeWorker(() => ({ result: result() }));
        const orig = worker.postMessage.bind(worker);
        worker.postMessage = (msg: unknown) => {
          const m = msg as { type: string; grammarBuffers?: Record<string, Uint8Array> };
          if (m.type === 'load-grammars') loadMsgs.push(m);
          orig(msg);
        };
        return worker;
      },
    });
    await pool.requestParse(task('a.ts'));
    expect(loadMsgs).toHaveLength(1);
    expect(loadMsgs[0].grammarBuffers).toBe(grammarBuffers);
    await pool.destroy();
  });

  it('serves a queue larger than the pool size', async () => {
    const { pool } = makePool(2, (m) => ({ result: result(Number(m.filePath.replace(/\D/g, ''))) }));
    const ps = Array.from({ length: 10 }, (_, i) => pool.requestParse(task(`${i}.ts`)));
    const res = await Promise.all(ps);
    expect(res.map((r) => r.durationMs).sort((a, b) => a - b)).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9]);
    await pool.destroy();
  });

  it('destroy() rejects in-flight and subsequent parses', async () => {
    const { pool } = makePool(1, () => ({ hang: true }));
    const p = pool.requestParse(task('x.ts'));
    p.catch(() => {}); // avoid an unhandled rejection before we assert
    await sleep(10);
    await pool.destroy();
    await expect(p).rejects.toThrow(/destroyed/);
    await expect(pool.requestParse(task('y.ts'))).rejects.toThrow(/destroyed/);
  });
});