gbrain/test/e2e/sync-parallel.test.ts

/**
 * E2E test for parallel sync against real Postgres.
 *
 *   T2 — happy path: 60-file sync at concurrency=4 against PostgresEngine
 *        actually constructs N worker engines, imports correctly, and does
 *        not leak connections (probe pg_stat_activity before/after).
 *   P4 — benchmark: serial vs concurrency=4 timing on the same fixture so
 *        the v0.22.13 CHANGELOG can quote a real number instead of "~4×".
 *
 * Gated on DATABASE_URL. Run via:
 *   docker run -d --name gbrain-test-pg -e POSTGRES_USER=postgres \
 *     -e POSTGRES_PASSWORD=postgres -e POSTGRES_DB=gbrain_test \
 *     -p 5435:5432 pgvector/pgvector:pg16
 *   DATABASE_URL=postgresql://postgres:postgres@localhost:5435/gbrain_test \
 *     bun test test/e2e/sync-parallel.test.ts
 */
import { describe, test, expect, beforeAll, afterAll } from 'bun:test';
import { mkdtempSync, rmSync, writeFileSync, mkdirSync } from 'fs';
import { join } from 'path';
import { tmpdir } from 'os';
import { execSync } from 'child_process';
import { hasDatabase, setupDB, teardownDB, getEngine, getConn } from './helpers.ts';

const skip = !hasDatabase();
const describeE2E = skip ? describe.skip : describe;

if (skip) {
  console.log('Skipping E2E sync-parallel tests (DATABASE_URL not set)');
}

function seedRepo(repoPath: string, fileCount: number): string {
  execSync('git init', { cwd: repoPath, stdio: 'pipe' });
  execSync('git config user.email "test@test.com"', { cwd: repoPath, stdio: 'pipe' });
  execSync('git config user.name "Test"', { cwd: repoPath, stdio: 'pipe' });
  mkdirSync(join(repoPath, 'people'), { recursive: true });
  for (let i = 0; i < fileCount; i++) {
    writeFileSync(join(repoPath, `people/p${i}.md`), [
      '---',
      'type: person',
      `title: Person ${i}`,
      '---',
      '',
      `Person ${i} body — some text long enough to chunk.`,
      `Iteration index ${i}, generated by sync-parallel E2E.`,
    ].join('\n'));
  }
  execSync('git add -A && git commit -m "initial"', { cwd: repoPath, stdio: 'pipe' });
  return execSync('git rev-parse HEAD', { cwd: repoPath, encoding: 'utf-8' }).trim();
}

async function activeConnections(): Promise<number> {
  const conn = getConn();
  const rows = await conn.unsafe(`
    SELECT count(*) AS n FROM pg_stat_activity
    WHERE datname = current_database()
      AND state IS NOT NULL
  `) as Array<{ n: string }>;
  return parseInt(rows[0]?.n ?? '0', 10);
}

describeE2E('E2E sync-parallel: T2 happy path + leak probe', () => {
  let repoPath: string;

  beforeAll(async () => {
    await setupDB();
  });

  afterAll(async () => {
    if (repoPath) rmSync(repoPath, { recursive: true, force: true });
    await teardownDB();
  });

  test('60-file Postgres sync at concurrency=4 imports all + no connection leak', async () => {
    repoPath = mkdtempSync(join(tmpdir(), 'gbrain-e2e-par-'));
    seedRepo(repoPath, 60);

    const before = await activeConnections();

    const { performSync } = await import('../../src/commands/sync.ts');
    const engine = getEngine();
    const result = await performSync(engine, {
      repoPath,
      noPull: true,
      noEmbed: true,
      concurrency: 4,
    });

    // First sync routes through performFullSync (delegates to runImport which
    // also accepts --workers); status is 'first_sync'.
    expect(result.status).toBe('first_sync');

    const after = await activeConnections();

    // Allow some slack — the helper engine + sync's normal pool stay open.
    // Worker engines (4 × 2 = 8 connections) MUST have closed; if they
    // hadn't, after - before would be at least 8.
    expect(after - before).toBeLessThan(4);

    // Verify pages are actually in the DB (via raw SQL — engine API also works).
    const conn = getConn();
    const pageRows = await conn.unsafe(
      `SELECT count(*) AS n FROM pages WHERE slug LIKE 'people/p%'`,
    ) as Array<{ n: string }>;
    const count = parseInt(pageRows[0]?.n ?? '0', 10);
    expect(count).toBe(60);
  }, 60_000);
});

describeE2E('E2E sync-parallel: P4 benchmark serial vs concurrency=4', () => {
  let repoSerial: string;
  let repoParallel: string;

  beforeAll(async () => {
    await setupDB();
  });

  afterAll(async () => {
    if (repoSerial) rmSync(repoSerial, { recursive: true, force: true });
    if (repoParallel) rmSync(repoParallel, { recursive: true, force: true });
    await teardownDB();
  });

  test('120-file benchmark: report serial and parallel wall-clock', async () => {
    // Two separate repos so neither sync's chunks bleed into the other.
    repoSerial = mkdtempSync(join(tmpdir(), 'gbrain-bench-serial-'));
    repoParallel = mkdtempSync(join(tmpdir(), 'gbrain-bench-parallel-'));
    seedRepo(repoSerial, 120);
    seedRepo(repoParallel, 120);

    const { performSync } = await import('../../src/commands/sync.ts');
    const engine = getEngine();

    // Truncate between runs to keep the benchmark honest.
    const conn = getConn();

    const t1 = Date.now();
    await performSync(engine, {
      repoPath: repoSerial,
      noPull: true,
      noEmbed: true,
      concurrency: 1,
    });
    const serialMs = Date.now() - t1;

    // Wipe pages before second run so neither one is "incremental".
    await conn.unsafe(`TRUNCATE pages CASCADE`);
    await conn.unsafe(`TRUNCATE config CASCADE`);

    const t2 = Date.now();
    await performSync(engine, {
      repoPath: repoParallel,
      noPull: true,
      noEmbed: true,
      concurrency: 4,
    });
    const parallelMs = Date.now() - t2;

    const speedup = (serialMs / parallelMs).toFixed(2);
    // Emit as a single line stdout consumers can grep for.
    console.log(`SYNC_PARALLEL_BENCH 120 files | serial=${serialMs}ms | parallel(4)=${parallelMs}ms | speedup=${speedup}x`);

    // Soft assertion: parallel must not be slower than serial. The actual
    // speedup ratio depends heavily on Postgres latency profile and is what
    // the CHANGELOG quotes — don't gate the test on a specific multiplier.
    expect(parallelMs).toBeLessThanOrEqual(serialMs * 1.5); // +50% slack for noisy CI
  }, 120_000);
});