fix(indexing): HDD-class storage — false parse timeouts, dropped files, and WAL checkpoint write-back (#1231) (#1242)
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 <noreply@anthropic.com>
This commit is contained in:
Colby Mchenry
co-authored by
Claude Fable 5
Claude Fable 5
parent
e76a355df5
commit
a11a439002
@@ -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');
|
||||
});
|
||||
});
|
||||
@@ -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<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`)));
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user