From a11a439002dd0d7bda8683fb8da2b03deec956e6 Mon Sep 17 00:00:00 2001 From: Colby Mchenry Date: Fri, 10 Jul 2026 03:42:30 -0500 Subject: [PATCH] =?UTF-8?q?fix(indexing):=20HDD-class=20storage=20?= =?UTF-8?q?=E2=80=94=20false=20parse=20timeouts,=20dropped=20files,=20and?= =?UTF-8?q?=20WAL=20checkpoint=20write-back=20(#1231)=20(#1242)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Parse timeouts are now judged by the worker's own clock: the base timer only marks a job late (after a long synchronous store stall, Node runs the timers phase before the poll phase, so the timer fired before an already-delivered result was processed — killing workers over parses that took milliseconds, even on 0-byte files); a result arriving before a 3× hard-kill backstop is accepted, timed-out files are retried, and CODEGRAPH_PARSE_TIMEOUT_MS overrides the budget. Grammar WASM bytes are pre-read once on the main thread and handed to every worker, so spawns/respawns load grammars from memory instead of re-reading a saturated disk. Bulk indexing defers WAL auto-checkpointing for the whole run: the default 1000-page interval re-writes hot B-tree/FTS pages into the main DB file over and over — ~95% of all disk I/O under throttled measurement. A WalCheckpointValve bounds WAL growth with off-thread PASSIVE backfill passes (never blocking the writer or the #850 watchdog heartbeat), pauses the writer for a full backfill if the disk truly can't keep up, and folds the WAL at the parse→resolution boundary so post-parse reads never page a bulk-write-sized WAL. Opt out with CODEGRAPH_NO_WAL_DEFER=1; tune with CODEGRAPH_WAL_VALVE_MB. Measured at 150 IOPS (HDD class): commons-lang 1526s → 59s with 0 dropped files (was 8); guava-scale completes in 7.6 min with a full graph where v1.3.1 needed 25 min for a repo 5× smaller. Unthrottled: no change. Co-authored-by: Claude Fable 5 --- .dockerignore | 7 + CHANGELOG.md | 9 ++ __tests__/grammar-wasm-bytes.test.ts | 43 ++++++ __tests__/parse-pool.test.ts | 67 ++++++++- __tests__/wal-deferral.test.ts | 217 +++++++++++++++++++++++++++ src/db/index.ts | 120 ++++++++++++++- src/db/wal-valve.ts | 206 +++++++++++++++++++++++++ src/extraction/grammars.ts | 133 ++++++++++------ src/extraction/index.ts | 39 ++++- src/extraction/parse-pool.ts | 96 +++++++++++- src/extraction/parse-worker.ts | 13 +- src/index.ts | 51 ++++++- 12 files changed, 928 insertions(+), 73 deletions(-) create mode 100644 .dockerignore create mode 100644 __tests__/grammar-wasm-bytes.test.ts create mode 100644 __tests__/wal-deferral.test.ts create mode 100644 src/db/wal-valve.ts diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..5e7e5a0 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,7 @@ +node_modules +dist +.git +.codegraph +.kommandr +docs +assets diff --git a/CHANGELOG.md b/CHANGELOG.md index 06d0d23..07e3e07 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,15 @@ and adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## [Unreleased] +### New Features + +- Indexing is dramatically faster on slow storage — mechanical HDDs, network folders, and virtualized disks. The database no longer folds its write journal back into the main file thousands of times during a bulk index (that folding was ~95% of all disk activity); it now streams writes sequentially and folds them back in a few large, coalesced passes that run off the main thread. In a disk-throttled benchmark matching the reported hardware, a mid-size Java project went from over 25 minutes to under a minute, and there is no change on fast disks. Opt out with `CODEGRAPH_NO_WAL_DEFER=1`; tune the fold-back threshold with `CODEGRAPH_WAL_VALVE_MB`. (#1231) +- New `CODEGRAPH_PARSE_TIMEOUT_MS` environment variable to raise the per-file parse budget on unusually slow storage, the same way `CODEGRAPH_PARSE_WORKERS` already tunes the worker count. (#1231) + +### Fixes + +- Indexing on slow storage (mechanical HDDs, network folders) no longer collapses into false "parse timeout" failures. When disk writes stalled the coordinating thread, parses that had already finished — including empty files — were being misjudged as hung, their workers killed, and the files silently dropped from the index. A parse result is now judged by the worker's own clock, so a stalled coordinator accepts the finished result instead of killing the worker; only a genuinely hung parse is terminated (after a wider grace window). Files that do hit the timeout are retried at the end of indexing instead of being silently lost. Thanks @KnifeOfLife for the exceptional report. (#1231) +- Parse workers now receive their grammar files from memory instead of each re-reading them from disk on spawn, eliminating a feedback loop on slow disks where every worker restart added more disk contention — and making worker restarts cheaper everywhere. (#1231) ## [1.3.1] - 2026-07-09 diff --git a/__tests__/grammar-wasm-bytes.test.ts b/__tests__/grammar-wasm-bytes.test.ts new file mode 100644 index 0000000..e2daa27 --- /dev/null +++ b/__tests__/grammar-wasm-bytes.test.ts @@ -0,0 +1,43 @@ +/** + * readGrammarWasmBytes + bytes-based grammar loading (#1231, Phase 2.1). + * + * The orchestrator pre-reads each needed grammar's WASM once on the main + * thread and hands the bytes to every parse worker, so a worker respawn loads + * grammars from memory instead of re-reading them from a (possibly slow) disk. + * These tests pin that the byte reader resolves the same artifacts the loader + * would, and that web-tree-sitter genuinely accepts the bytes. + */ +import { describe, it, expect } from 'vitest'; +import { Parser, Language as WasmLanguage } from 'web-tree-sitter'; +import { readGrammarWasmBytes } from '../src/extraction/grammars'; + +describe('readGrammarWasmBytes', () => { + it('reads bytes for a tree-sitter-wasms grammar and a vendored grammar', async () => { + const bytes = await readGrammarWasmBytes(['typescript', 'lua']); + expect(bytes.typescript).toBeInstanceOf(Uint8Array); // from tree-sitter-wasms + expect(bytes.typescript.byteLength).toBeGreaterThan(10_000); + expect(bytes.lua).toBeInstanceOf(Uint8Array); // vendored under src/extraction/wasm/ + expect(bytes.lua.byteLength).toBeGreaterThan(10_000); + }); + + it('expands delegating languages to the grammars they need (svelte → ts/js)', async () => { + const bytes = await readGrammarWasmBytes(['svelte']); + expect(Object.keys(bytes).sort()).toEqual(['javascript', 'typescript']); + }); + + it('omits languages without a WASM grammar instead of failing', async () => { + const bytes = await readGrammarWasmBytes(['yaml', 'unknown']); + expect(Object.keys(bytes)).toEqual([]); + }); + + it('produces bytes web-tree-sitter can load into a working parser', async () => { + await Parser.init(); + const bytes = await readGrammarWasmBytes(['javascript']); + const language = await WasmLanguage.load(bytes.javascript); + const parser = new Parser(); + parser.setLanguage(language); + const tree = parser.parse('function hello() { return 1; }'); + expect(tree!.rootNode.hasError).toBe(false); + expect(tree!.rootNode.toString()).toContain('function_declaration'); + }); +}); diff --git a/__tests__/parse-pool.test.ts b/__tests__/parse-pool.test.ts index 72b212f..641d24d 100644 --- a/__tests__/parse-pool.test.ts +++ b/__tests__/parse-pool.test.ts @@ -11,7 +11,7 @@ * parallelism safe. */ import { describe, it, expect } from 'vitest'; -import { ParseWorkerPool, resolveParsePoolSize, type ParsePoolWorker, type ParseTask } from '../src/extraction/parse-pool'; +import { ParseWorkerPool, 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)); @@ -80,6 +80,20 @@ function makePool( 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('resolveParsePoolSize', () => { it('treats explicit 0 and 1 as a single worker (the rollback path)', () => { expect(resolveParsePoolSize('0', 8)).toBe(1); @@ -150,14 +164,61 @@ describe('ParseWorkerPool', () => { await pool.destroy(); }); - it('times out a hung parse and stays usable', async () => { + 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 }); - await expect(pool.requestParse(task('hang.ts'))).rejects.toThrow(/timed out/i); + 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 }> = []; + 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 }; + 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`))); diff --git a/__tests__/wal-deferral.test.ts b/__tests__/wal-deferral.test.ts new file mode 100644 index 0000000..e9a1598 --- /dev/null +++ b/__tests__/wal-deferral.test.ts @@ -0,0 +1,217 @@ +/** + * WAL checkpoint deferral during bulk indexing (#1231). + * + * The default 1000-page wal_autocheckpoint re-writes hot pages into the main + * DB over and over during a bulk index (~95% of all disk I/O on slow + * storage). indexAll defers auto-checkpointing for the whole run, a + * WalCheckpointValve bounds WAL growth via off-thread PASSIVE checkpoints, + * and the interval is restored afterwards. These tests pin the DB helpers, + * the valve's trigger/dedupe/backpressure logic, and the end-to-end indexAll + * behavior (identical graph with and without deferral; interval restored). + */ +import { describe, it, expect, beforeEach, afterEach } from 'vitest'; +import * as fs from 'fs'; +import * as os from 'os'; +import * as path from 'path'; +import { DatabaseConnection } from '../src/db'; +import { WalCheckpointValve, resolveWalValveMb } from '../src/db/wal-valve'; +import CodeGraph from '../src/index'; + +let tmpDir: string; + +beforeEach(() => { + tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), 'cg-wal-deferral-')); +}); + +afterEach(() => { + fs.rmSync(tmpDir, { recursive: true, force: true }); +}); + +function openDb(): DatabaseConnection { + return DatabaseConnection.initialize(path.join(tmpDir, 'test.db')); +} + +/** Grow the WAL: with autocheckpoint off, every commit appends and nothing folds back. */ +function writeRows(db: DatabaseConnection, rows: number): void { + const raw = db.getDb(); + raw.exec('CREATE TABLE IF NOT EXISTS t (id INTEGER PRIMARY KEY, blob TEXT)'); + const stmt = raw.prepare('INSERT INTO t (blob) VALUES (?)'); + for (let i = 0; i < rows; i++) stmt.run('x'.repeat(4096)); +} + +describe('resolveWalValveMb', () => { + it('honors a positive numeric override and falls back otherwise', () => { + expect(resolveWalValveMb('64')).toBe(64); + expect(resolveWalValveMb('64.9')).toBe(64); + expect(resolveWalValveMb(undefined)).toBe(256); + expect(resolveWalValveMb('')).toBe(256); + expect(resolveWalValveMb('abc')).toBe(256); + expect(resolveWalValveMb('0')).toBe(256); + expect(resolveWalValveMb('-5')).toBe(256); + }); +}); + +describe('DatabaseConnection WAL helpers', () => { + it('reads and writes the wal_autocheckpoint interval', () => { + const db = openDb(); + expect(db.getWalAutocheckpoint()).toBe(1000); // SQLite default + db.setWalAutocheckpoint(0); + expect(db.getWalAutocheckpoint()).toBe(0); + db.setWalAutocheckpoint(1000); + expect(db.getWalAutocheckpoint()).toBe(1000); + db.close(); + }); + + it('reports WAL size that grows with deferred commits', () => { + const db = openDb(); + db.setWalAutocheckpoint(0); + const before = db.getWalSizeBytes(); + writeRows(db, 200); + expect(db.getWalSizeBytes()).toBeGreaterThan(before); + db.close(); + }); + + it('checkpointWalPassive backfills the WAL from a worker connection and reports the result', async () => { + const db = openDb(); + db.setWalAutocheckpoint(0); + writeRows(db, 500); + const dbFile = path.join(tmpDir, 'test.db'); + const mainSizeBefore = fs.statSync(dbFile).size; + const res = await db.checkpointWalPassive(); + // Backfill moves the committed pages into the main DB file… + expect(fs.statSync(dbFile).size).toBeGreaterThan(mainSizeBefore); + // …and reports a full backfill (idle DB: every WAL frame checkpointed). + expect(res).not.toBeNull(); + expect(res!.busy).toBe(0); + expect(res!.log).toBeGreaterThan(0); + expect(res!.checkpointed).toBe(res!.log); + db.close(); + }); +}); + +describe('WalCheckpointValve', () => { + it('check() fires an off-thread checkpoint once growth passes the soft threshold', async () => { + const db = openDb(); + db.setWalAutocheckpoint(0); + writeRows(db, 500); // WAL well past a ~10-byte threshold + const valve = new WalCheckpointValve(db, 0.00001); // ~10 bytes soft + const dbFile = path.join(tmpDir, 'test.db'); + const mainSizeBefore = fs.statSync(dbFile).size; + valve.check(); + await valve.drain(); + expect(fs.statSync(dbFile).size).toBeGreaterThan(mainSizeBefore); + db.close(); + }); + + it('advances its baseline on a full backfill — a wrapped WAL does not retrigger it', async () => { + const db = openDb(); + db.setWalAutocheckpoint(0); + writeRows(db, 500); + const valve = new WalCheckpointValve(db, 0.00001); + valve.check(); + await valve.drain(); // full backfill on an idle DB → baseline = current file size + // The WAL file keeps its high-water size, but growth is now 0: neither + // the timer path nor backpressure may fire again (the pre-fix bug fired + // on raw size forever and serialized every store behind a checkpoint). + expect(valve.backpressure()).toBeNull(); + valve.check(); + await valve.drain(); // no-op drain: nothing in flight + // New commits recycle wrapped frames — file size is flat, still no trigger. + writeRows(db, 5); + expect(valve.backpressure()).toBeNull(); + db.close(); + }); + + it('does not fire below the soft threshold', async () => { + const db = openDb(); + db.setWalAutocheckpoint(0); + writeRows(db, 5); + const valve = new WalCheckpointValve(db, 1024); // 1GB soft — never reached + const dbFile = path.join(tmpDir, 'test.db'); + const mainSizeBefore = fs.statSync(dbFile).size; + valve.check(); + await valve.drain(); + expect(fs.statSync(dbFile).size).toBe(mainSizeBefore); + db.close(); + }); + + it('backpressure() is null under the hard cap and a promise above it', async () => { + const db = openDb(); + db.setWalAutocheckpoint(0); + writeRows(db, 500); + const relaxed = new WalCheckpointValve(db, 1024); + expect(relaxed.backpressure()).toBeNull(); + const strict = new WalCheckpointValve(db, 0.0000001); // hard cap ~0.4 bytes + const bp = strict.backpressure(); + expect(bp).toBeInstanceOf(Promise); + await bp; + await strict.drain(); + db.close(); + }); + + it('foldNow() backfills everything at a phase boundary and resets growth', async () => { + const db = openDb(); + db.setWalAutocheckpoint(0); + writeRows(db, 500); + const valve = new WalCheckpointValve(db, 1024); // thresholds never reached on their own + const dbFile = path.join(tmpDir, 'test.db'); + const mainSizeBefore = fs.statSync(dbFile).size; + await valve.foldNow(); + expect(fs.statSync(dbFile).size).toBeGreaterThan(mainSizeBefore); // pages backfilled + expect(valve.backpressure()).toBeNull(); // baseline advanced — growth is zero + await valve.foldNow(); // second fold is a no-op (growth 0), must not spin + db.close(); + }); + + it('dedupes concurrent fires into one in-flight checkpoint', () => { + const db = openDb(); + db.setWalAutocheckpoint(0); + writeRows(db, 500); + const valve = new WalCheckpointValve(db, 0.00001); + valve.check(); + const first = valve.backpressure(); + const second = valve.backpressure(); + expect(second).toBe(first); // same in-flight promise, not a second worker + db.close(); + return first ?? undefined; + }); +}); + +describe('indexAll WAL deferral end-to-end', () => { + function writeFixtureProject(): void { + fs.mkdirSync(path.join(tmpDir, 'src'), { recursive: true }); + for (let i = 0; i < 8; i++) { + fs.writeFileSync( + path.join(tmpDir, 'src', `mod${i}.ts`), + `export function fn${i}(x: number): number { return helper${i}(x) + ${i}; }\n` + + `function helper${i}(x: number): number { return x * ${i}; }\n` + ); + } + } + + it('produces the same graph with and without deferral, and restores the interval', async () => { + writeFixtureProject(); + + const cg1 = CodeGraph.initSync(tmpDir); + const r1 = await cg1.indexAll(); + expect(r1.success).toBe(true); + // Deferral is scoped to the run: the connection is back on the default. + const conn1 = (cg1 as unknown as { db: DatabaseConnection }).db; + expect(conn1.getWalAutocheckpoint()).toBe(1000); + const counts1 = { nodes: r1.nodesCreated, edges: r1.edgesCreated }; + await cg1.close(); + + fs.rmSync(path.join(tmpDir, '.codegraph'), { recursive: true, force: true }); + + process.env.CODEGRAPH_NO_WAL_DEFER = '1'; + try { + const cg2 = CodeGraph.initSync(tmpDir); + const r2 = await cg2.indexAll(); + expect(r2.success).toBe(true); + expect({ nodes: r2.nodesCreated, edges: r2.edgesCreated }).toEqual(counts1); + await cg2.close(); + } finally { + delete process.env.CODEGRAPH_NO_WAL_DEFER; + } + }); +}); diff --git a/src/db/index.ts b/src/db/index.ts index 80d55d6..0cd2823 100644 --- a/src/db/index.ts +++ b/src/db/index.ts @@ -189,6 +189,98 @@ export class DatabaseConnection { return stats.size; } + /** + * Size of the `-wal` sidecar file in bytes. 0 when it doesn't exist (non-WAL + * journal mode, in-memory DB, or no write since the last checkpoint+reset). + */ + getWalSizeBytes(): number { + if (!this.dbPath || this.dbPath === ':memory:') return 0; + try { + return fs.statSync(`${this.dbPath}-wal`).size; + } catch { + return 0; + } + } + + /** Current `wal_autocheckpoint` interval in pages (0 = disabled). */ + getWalAutocheckpoint(): number { + const v = this.db.pragma('wal_autocheckpoint', { simple: true }); + const n = Number(v); + return Number.isFinite(n) ? n : 0; + } + + /** + * Set the connection's `wal_autocheckpoint` interval (pages; 0 disables). + * Bulk indexing defers checkpoints entirely (#1231): the default 1000-page + * auto-checkpoint re-writes hot B-tree/FTS pages into the main DB file over + * and over — measured at ~95% of ALL disk I/O during a bulk index, and the + * difference between 45s and 19+ minutes on HDD-class storage. During + * deferral a {@link WalCheckpointValve} bounds WAL growth off-thread. + */ + setWalAutocheckpoint(pages: number): void { + this.db.pragma(`wal_autocheckpoint = ${Math.max(0, Math.floor(pages))}`); + } + + /** + * `PRAGMA wal_checkpoint(PASSIVE)` on a worker thread with its own + * connection. PASSIVE never blocks the writer, and running it off-thread + * means the main thread — and the #850 watchdog heartbeat — keep turning + * even when the backfill is minutes of I/O on slow storage (a synchronous + * checkpoint that exceeds the watchdog's 60s window gets a healthy index + * SIGKILLed — observed in the #1231 repro). + * + * Returns SQLite's checkpoint result row — `log === checkpointed` with + * `busy === 0` means the ENTIRE WAL was backfilled, so the writer's next + * commit restarts the WAL from the top and the file stops growing. The + * WAL valve needs that signal because a WAL file's SIZE never shrinks: + * after the first wrap, raw file size says nothing about the un-backfilled + * backlog. Best-effort: returns null on any failure (including worker + * threads being unavailable — a potentially minutes-long checkpoint must + * never run inline on the main thread). + */ + async checkpointWalPassive(): Promise<{ busy: number; log: number; checkpointed: number } | null> { + if (!this.dbPath || this.dbPath === ':memory:') { + try { + const row = this.db.prepare('PRAGMA wal_checkpoint(PASSIVE)').get() as Record | undefined; + return row ? { busy: Number(row.busy), log: Number(row.log), checkpointed: Number(row.checkpointed) } : null; + } catch { + return null; + } + } + try { + const { Worker } = await import('node:worker_threads'); + const workerSource = ` + const { workerData, parentPort } = require('node:worker_threads'); + let row = null; + try { + const { DatabaseSync } = require('node:sqlite'); + const db = new DatabaseSync(workerData.dbPath); + try { row = db.prepare('PRAGMA wal_checkpoint(PASSIVE)').get(); } catch {} + try { db.close(); } catch {} + } catch {} + parentPort.postMessage({ row }); + `; + return await new Promise((resolve) => { + let settled = false; + const finish = (row?: Record | null): void => { + if (settled) return; + settled = true; + resolve(row ? { busy: Number(row.busy), log: Number(row.log), checkpointed: Number(row.checkpointed) } : null); + }; + try { + const worker = new Worker(workerSource, { eval: true, workerData: { dbPath: this.dbPath } }); + worker.once('message', (m: { row?: Record | null }) => { void worker.terminate(); finish(m?.row ?? null); }); + worker.once('error', () => { void worker.terminate(); finish(null); }); + worker.once('exit', () => finish(null)); + } catch { + finish(null); + } + }); + } catch { + return null; + } + } + /** * Optimize database (vacuum and analyze) */ @@ -233,6 +325,22 @@ export class DatabaseConnection { try { this.db.exec('PRAGMA wal_checkpoint(PASSIVE)'); } catch { /* ignore */ } return; } + await this.runPragmasOffThread( + ['PRAGMA analysis_limit=1000', 'PRAGMA optimize', 'PRAGMA wal_checkpoint(PASSIVE)'], + // Worker threads unavailable — bounded in-line fallback, no checkpoint. + ['PRAGMA analysis_limit=1000', 'PRAGMA optimize'] + ); + } + + /** + * Run pragmas on a worker thread against its own connection to this DB + * (shared machinery for {@link runMaintenance} and + * {@link checkpointWalPassive}). Each pragma is individually best-effort; + * the whole call is best-effort. `inlineFallback` (if any) runs on THIS + * connection only when worker threads are unavailable — keep it to pragmas + * that are safe to run synchronously on the main thread. + */ + private async runPragmasOffThread(pragmas: string[], inlineFallback: string[] = []): Promise { try { const { Worker } = await import('node:worker_threads'); const workerSource = ` @@ -240,9 +348,7 @@ export class DatabaseConnection { try { const { DatabaseSync } = require('node:sqlite'); const db = new DatabaseSync(workerData.dbPath); - try { db.exec('PRAGMA analysis_limit=1000'); } catch {} - try { db.exec('PRAGMA optimize'); } catch {} - try { db.exec('PRAGMA wal_checkpoint(PASSIVE)'); } catch {} + for (const p of workerData.pragmas) { try { db.exec(p); } catch {} } try { db.close(); } catch {} } catch {} parentPort.postMessage('done'); @@ -253,7 +359,7 @@ export class DatabaseConnection { if (!settled) { settled = true; resolve(); } }; try { - const worker = new Worker(workerSource, { eval: true, workerData: { dbPath: this.dbPath } }); + const worker = new Worker(workerSource, { eval: true, workerData: { dbPath: this.dbPath, pragmas } }); worker.once('message', () => { void worker.terminate(); finish(); }); worker.once('error', () => { void worker.terminate(); finish(); }); worker.once('exit', finish); @@ -262,9 +368,9 @@ export class DatabaseConnection { } }); } catch { - // Worker threads unavailable — bounded in-line fallback, no checkpoint. - try { this.db.exec('PRAGMA analysis_limit=1000'); } catch { /* ignore */ } - try { this.db.exec('PRAGMA optimize'); } catch { /* ignore */ } + for (const p of inlineFallback) { + try { this.db.exec(p); } catch { /* ignore */ } + } } } diff --git a/src/db/wal-valve.ts b/src/db/wal-valve.ts new file mode 100644 index 0000000..b509527 --- /dev/null +++ b/src/db/wal-valve.ts @@ -0,0 +1,206 @@ +/** + * WAL checkpoint valve — bounds WAL growth while auto-checkpointing is + * deferred during a bulk index (#1231). + * + * Why deferral: SQLite's default `wal_autocheckpoint` (1000 pages) re-writes + * hot B-tree/FTS pages into the main DB file over and over during a bulk + * index — measured at ~95% of ALL disk I/O, and the difference between 45s + * and 19+ minutes on HDD-class storage (150 random IOPS). Deferring + * checkpoints turns the store into pure sequential WAL appends; each backfill + * pass writes distinct pages once, in page order (≈ sequential). + * + * Why a valve: unbounded deferral is its own failure mode, both measured in + * the #1231 repro. The WAL duplicates hot pages per COMMIT, so it grows far + * faster than the DB (5.9GB WAL for a ~340MB DB on a 3.3k-file index) — + * filling the disk, and poisoning every subsequent read that must page + * through it (the first resolution-phase read blocked the main thread >60s + * and the #850 liveness watchdog killed the healthy index). The valve + * watches WAL growth on a timer and, past a soft threshold, backfills with + * `PRAGMA wal_checkpoint(PASSIVE)` on a worker-thread connection — PASSIVE + * never blocks the writer, and off-thread means the main thread (and the + * watchdog heartbeat) keep turning regardless of how long a backfill takes. + * + * The load-bearing subtlety: a WAL file's SIZE never shrinks. After a full + * backfill, the writer's next commit RESTARTS the WAL from the top and the + * frames recycle inside the same file — so raw size says nothing about the + * un-backfilled backlog, and a size-triggered valve degenerates into firing + * (and pausing the writer) forever once the file passes its threshold + * (measured: guava crawled at ~9min per 160 files). Instead the valve + * tracks `sizeAtLastFullBackfill` — refreshed whenever a checkpoint reports + * `log === checkpointed` (everything backfilled) — and triggers on GROWTH + * beyond that baseline, which only happens when genuinely un-backfilled + * frames push past the file's high-water mark. + * + * Backpressure: if the writer outruns the checkpointer past a hard cap of + * growth (2× soft), {@link backpressure} pauses the writer (at a safe, + * between-transactions boundary) until a FULL backfill lands. One in-flight + * pass is not enough: on a disk saturated by the writer, every concurrent + * PASSIVE pass is already stale by the time it finishes (the writer appended + * past its snapshot), so neither SQLite's WAL wrap nor the baseline ever + * trigger and the WAL grows without bound (measured: 5.9GB on guava at 150 + * IOPS, then a >60s read stall and a watchdog kill). With the writer parked, + * the next pass covers everything, the WAL wraps on the following commit, + * and the pause is the disk's honest catch-up cost — the correct terminal + * mode when hardware genuinely can't keep up with the append rate. + */ + +import type { DatabaseConnection } from './index'; + +/** Soft WAL-growth threshold (MB) that triggers an off-thread passive checkpoint. */ +const DEFAULT_WAL_VALVE_MB = 256; +/** Hard cap = this × soft threshold; past it the writer pauses for a full backfill. */ +const HARD_CAP_MULTIPLIER = 2; +/** Passes attempted per writer pause before giving up (a pinned reader could stall forever). */ +const MAX_PAUSED_BACKFILL_PASSES = 20; +/** How often the timer looks at the WAL file size. */ +const CHECK_INTERVAL_MS = 2000; + +/** + * Resolve the valve's soft threshold from the `CODEGRAPH_WAL_VALVE_MB` + * override; non-numeric / non-positive values fall back to the default. + */ +export function resolveWalValveMb(envVal: string | undefined): number { + if (envVal !== undefined && envVal !== '') { + const n = Number(envVal); + if (Number.isFinite(n) && n > 0) return Math.floor(n); + } + return DEFAULT_WAL_VALVE_MB; +} + +export class WalCheckpointValve { + private timer: ReturnType | null = null; + private inflight: Promise | null = null; + /** Writer pause in progress (hard cap breached): passes loop until a full backfill. */ + private pause: Promise | null = null; + /** + * WAL file size observed when a checkpoint last reported the ENTIRE WAL + * backfilled. Growth is measured against this baseline — see the header + * comment for why absolute size cannot be used. + */ + private sizeAtLastFullBackfill = 0; + private readonly softBytes: number; + private readonly hardBytes: number; + + constructor( + private readonly db: DatabaseConnection, + softMb: number = resolveWalValveMb(process.env.CODEGRAPH_WAL_VALVE_MB), + private readonly intervalMs: number = CHECK_INTERVAL_MS, + private readonly log: (msg: string) => void = () => {} + ) { + this.softBytes = softMb * 1024 * 1024; + this.hardBytes = this.softBytes * HARD_CAP_MULTIPLIER; + } + + private mb(n: number): string { + return `${Math.round(n / 1024 / 1024)}MB`; + } + + /** Un-backfilled growth estimate: bytes the WAL has grown past the last full backfill. */ + private growthBytes(): number { + return this.db.getWalSizeBytes() - this.sizeAtLastFullBackfill; + } + + /** Begin watching the WAL. Idempotent; the timer never holds the loop open. */ + start(): void { + if (this.timer) return; + this.timer = setInterval(() => this.check(), this.intervalMs); + this.timer.unref?.(); + } + + /** Stop watching. Any in-flight checkpoint keeps running — await drain(). */ + stop(): void { + if (this.timer) { + clearInterval(this.timer); + this.timer = null; + } + } + + /** One poll: fire an off-thread passive checkpoint when growth passes the soft threshold. */ + check(): void { + if (!this.pause && !this.inflight && this.growthBytes() > this.softBytes) this.fire(); + } + + /** + * Writer-side backstop, called at a between-transactions boundary. Returns + * null (no wait) while growth is under the hard cap; past it, returns a + * promise that resolves only once a FULL backfill has landed — see the + * header comment for why a single pass is not enough on a saturated disk. + */ + backpressure(): Promise | null { + if (this.pause) return this.pause; + if (this.growthBytes() <= this.hardBytes) return null; + this.log(`backpressure: wal=${this.mb(this.db.getWalSizeBytes())} baseline=${this.mb(this.sizeAtLastFullBackfill)} — pausing writer for full backfill`); + const t0 = Date.now(); + this.pause = this.backfillFully().finally(() => { + this.pause = null; + this.log(`backpressure released after ${Date.now() - t0}ms: wal=${this.mb(this.db.getWalSizeBytes())} baseline=${this.mb(this.sizeAtLastFullBackfill)}`); + }); + return this.pause; + } + + /** Await any in-flight checkpoint and writer pause. */ + async drain(): Promise { + while (this.pause || this.inflight) { + if (this.pause) await this.pause; + if (this.inflight) await this.inflight; + } + } + + /** + * Phase-boundary fold: backfill the ENTIRE WAL now (off-thread, awaited). + * Called between bulk phases — e.g. after parsing, before resolution's + * first reads — so the next phase never pages a bulk-write-sized WAL on + * the main thread (the post-parse read against a multi-GB WAL is what + * blew the #850 watchdog's 60s window in the #1231 repro). The await + * keeps the event loop (and the watchdog heartbeat) turning. + */ + async foldNow(): Promise { + await this.drain(); + if (this.growthBytes() <= 0) return; + this.log(`foldNow: wal=${this.mb(this.db.getWalSizeBytes())} baseline=${this.mb(this.sizeAtLastFullBackfill)}`); + this.pause = this.backfillFully().finally(() => { this.pause = null; }); + await this.pause; + } + + /** + * With the writer parked on the returned promise, loop passive passes until + * one reports the entire WAL backfilled (typically the second: the first + * drains the pass that was already running against a stale snapshot). Gives + * up after a bounded number of passes — e.g. a reader pinning the WAL — + * because unbounded WAL growth degrades; a wedged writer never recovers. + */ + private async backfillFully(): Promise { + for (let i = 0; i < MAX_PAUSED_BACKFILL_PASSES; i++) { + if (this.inflight) await this.inflight; // fold in the stale in-flight pass first + const res = await this.db.checkpointWalPassive(); + if (!res) return; // checkpoint machinery unavailable — don't spin + this.log(`backfill pass ${i + 1}: busy=${res.busy} log=${res.log} checkpointed=${res.checkpointed} wal=${this.mb(this.db.getWalSizeBytes())}`); + if (res.busy === 0 && res.log === res.checkpointed) { + this.sizeAtLastFullBackfill = this.db.getWalSizeBytes(); + return; + } + } + this.log(`backfill gave up after ${MAX_PAUSED_BACKFILL_PASSES} passes — WAL stays unbounded this cycle`); + } + + private fire(): void { + const p = this.db + .checkpointWalPassive() + .then((res) => { + // Full backfill (busy 0, every log frame checkpointed) ⇒ the writer's + // next commit wraps the WAL; the file's current size becomes the new + // growth baseline. A partial pass (writer appended during it, or a + // read transaction pinned frames) leaves the baseline alone, so the + // next tick fires again and copies the remainder. In non-WAL mode + // SQLite reports log = checkpointed = -1, which is harmless here. + if (res && res.busy === 0 && res.log === res.checkpointed) { + this.sizeAtLastFullBackfill = this.db.getWalSizeBytes(); + } + }) + .catch(() => { /* best-effort */ }) + .finally(() => { + if (this.inflight === p) this.inflight = null; + }); + this.inflight = p; + } +} diff --git a/src/extraction/grammars.ts b/src/extraction/grammars.ts index 5a1b9ed..a26d232 100644 --- a/src/extraction/grammars.ts +++ b/src/extraction/grammars.ts @@ -7,6 +7,7 @@ */ import * as path from 'path'; +import * as fsp from 'fs/promises'; import { Parser, Language as WasmLanguage } from 'web-tree-sitter'; import { Language } from '../types'; @@ -248,31 +249,101 @@ export async function initGrammars(): Promise { parserInitialized = true; } +/** + * Grammars that ship their own vendored WASMs under `dist/extraction/wasm/` + * (not in tree-sitter-wasms, or the tree-sitter-wasms build is too old). + * Lua: tree-sitter-wasms ships an ABI-13 build that corrupts the shared WASM + * heap under web-tree-sitter 0.25 (drops nested calls/imports on every file + * after the first); we vendor the upstream ABI-15 wasm instead. C#: the + * tree-sitter-wasms build (ABI 13) has no primary-constructor support and + * parses `class Foo(...)` as an ERROR that swallows the whole class (#237); we + * vendor the upstream ABI-15 tree-sitter-c-sharp 0.23.5 wasm, which parses + * primary constructors natively. Terraform: tree-sitter-wasms does not ship + * HCL/Terraform at all, so we vendor the prebuilt tree-sitter-terraform.wasm + * from @tree-sitter-grammars/tree-sitter-hcl 1.2.0 (Apache-2.0) — + * byte-identical to the npm package's artifact. ArkTS: tree-sitter-wasms + * doesn't ship it either; we vendor the prebuilt tree-sitter-arkts.wasm from + * the tree-sitter-arkts 0.2.0 npm package (harmony-contrib/tree-sitter-arkts, + * MIT) — byte-identical to the npm tarball's artifact. It extends the + * tree-sitter-javascript grammar the same way tree-sitter-typescript does, + * adding `struct_declaration` and the `arkui_component_expression` build() + * DSL. Nix: tree-sitter-wasms doesn't ship it; we vendor a wasm built from + * nix-community/tree-sitter-nix @ 3d0173d (MIT) with tree-sitter-cli 0.25.10 + * (`generate` + `build --wasm`, ABI 15 — upstream's checked-in parser.c is + * still ABI 13; all 54 upstream corpus tests pass on the regenerated parser). + */ +const VENDORED_WASM_LANGS: ReadonlySet = new Set([ + 'pascal', 'scala', 'lua', 'luau', 'csharp', 'r', 'cfml', 'cfscript', 'cfquery', + 'cobol', 'vbnet', 'erlang', 'terraform', 'arkts', 'nix', +]); + +/** Absolute path of a language's grammar WASM (vendored or tree-sitter-wasms). */ +function resolveWasmPath(lang: GrammarLanguage): string { + const wasmFile = WASM_GRAMMAR_FILES[lang]; + return VENDORED_WASM_LANGS.has(lang) + ? path.join(__dirname, 'wasm', wasmFile) + : require.resolve(`tree-sitter-wasms/out/${wasmFile}`); +} + +/** + * Expand an index set's languages to the grammars actually needed to parse it. + * SFC languages (svelte/vue/astro) have no grammar of their own — their + * extractors delegate