diff --git a/gitnexus/src/core/lbug/lbug-adapter.ts b/gitnexus/src/core/lbug/lbug-adapter.ts index 8b92ac032..5ee32721a 100644 --- a/gitnexus/src/core/lbug/lbug-adapter.ts +++ b/gitnexus/src/core/lbug/lbug-adapter.ts @@ -198,6 +198,20 @@ let conn: lbug.Connection | null = null; // reassigned only at open/close, never mid-load). const isSharedSingletonConn = (c: lbug.Connection): boolean => c === conn; +// True while the manual WAL-checkpoint driver is running (toggled by the driver's +// start/stop). `streamQuery` is deliberately NOT wrapped in withConnLock (its +// per-row callback can re-enter the adapter), so it must NOT run while a CHECKPOINT +// can fire on the unlocked read connection — the exact overlap the lock serializes +// everything else against (#2264). The serve/read path never starts the driver, so +// this stays false there; an in-process analyze overlapping a stream would trip the +// guard in streamQuery instead of silently corrupting native state. +let walDriverActive = false; + +/** Toggled by the WAL-checkpoint driver's start (true) / stop (false). @see streamQuery */ +export const markWalDriverActive = (active: boolean): void => { + walDriverActive = active; +}; + let currentDbPath: string | null = null; let currentDbReadOnly = false; let ftsLoaded = false; @@ -1529,6 +1543,18 @@ export const streamQuery = async ( cypher: string, onRow: (row: any) => void | Promise, ): Promise => { + if (walDriverActive) { + // streamQuery reads rows on the singleton connection WITHOUT withConnLock; if + // the WAL-checkpoint driver is live, those reads could race a CHECKPOINT — the + // #2264 corruption window. Today the serve/read path never runs the driver + // (analyze runs in a forked worker), so this fails loud only if a future + // in-process analyze overlaps a stream. Run analysis in a worker, or stop the + // driver before streaming. See conn-lock.ts. + throw new Error( + 'streamQuery cannot run while the WAL-checkpoint driver is active (it would ' + + 'race a CHECKPOINT on the unlocked read connection — #2264).', + ); + } if (!conn) { throw new Error('LadybugDB not initialized. Call initLbug first.'); } diff --git a/gitnexus/src/core/lbug/wal-checkpoint-driver.ts b/gitnexus/src/core/lbug/wal-checkpoint-driver.ts index 08e48e6f2..5c92bb980 100644 --- a/gitnexus/src/core/lbug/wal-checkpoint-driver.ts +++ b/gitnexus/src/core/lbug/wal-checkpoint-driver.ts @@ -33,7 +33,7 @@ */ import { logger } from '../logger.js'; -import { tryFlushWAL } from './lbug-adapter.js'; +import { tryFlushWAL, markWalDriverActive } from './lbug-adapter.js'; import { isLbugCheckpointIoError } from './lbug-config.js'; /** @@ -162,6 +162,10 @@ export const startWalCheckpointDriver = ( let stopped = false; let inflight: Promise | null = null; + // Arm the streamQuery guard: while this driver runs, an unlocked streamQuery on + // the singleton connection could race a CHECKPOINT (#2264). Cleared in stop(). + markWalDriverActive(true); + const tick = async (): Promise => { if (stopped) return; // Reentrancy guard: setInterval keeps firing on its fixed cadence even when @@ -218,6 +222,9 @@ export const startWalCheckpointDriver = ( /* swallowed in tick() — surface path is the surrounding write */ } } + // Disarm AFTER the in-flight CHECKPOINT drains — clearing it earlier would + // briefly let a streamQuery race the still-finishing CHECKPOINT (#2264). + markWalDriverActive(false); }, }; }; diff --git a/gitnexus/test/unit/stream-query-driver-guard.test.ts b/gitnexus/test/unit/stream-query-driver-guard.test.ts new file mode 100644 index 000000000..3dac28d1c --- /dev/null +++ b/gitnexus/test/unit/stream-query-driver-guard.test.ts @@ -0,0 +1,52 @@ +/** + * Unit tests for the streamQuery WAL-driver guard (#2264). streamQuery is + * deliberately NOT wrapped in withConnLock (its per-row callback re-enters the + * adapter), so it must refuse to run while the WAL-checkpoint driver is live — + * otherwise its unlocked per-row reads could race a CHECKPOINT on the shared + * connection (the corruption window the lock serializes everything else against). + * Today the serve/read path never starts the driver; this guard fails loud if a + * future in-process analyze ever overlaps a stream. + */ +import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest'; +import { streamQuery, markWalDriverActive } from '../../src/core/lbug/lbug-adapter.js'; +import { startWalCheckpointDriver } from '../../src/core/lbug/wal-checkpoint-driver.js'; + +describe('streamQuery WAL-driver guard (#2264)', () => { + beforeEach(() => { + // Manual checkpoint defaults on; pin it so the driver path is deterministic. + vi.stubEnv('GITNEXUS_WAL_MANUAL_CHECKPOINT', '1'); + }); + + afterEach(() => { + markWalDriverActive(false); + vi.unstubAllEnvs(); + }); + + it('throws when the WAL-checkpoint driver is active', async () => { + markWalDriverActive(true); + await expect(streamQuery('RETURN 1 AS one', () => undefined)).rejects.toThrow( + /WAL-checkpoint driver is active/, + ); + }); + + it('passes the guard when inactive (reaching the not-initialized check)', async () => { + markWalDriverActive(false); + await expect(streamQuery('RETURN 1 AS one', () => undefined)).rejects.toThrow( + /not initialized/, + ); + }); + + it('startWalCheckpointDriver arms the guard; stop() disarms it', async () => { + const driver = startWalCheckpointDriver({ periodMs: 1_000_000 }); + try { + await expect(streamQuery('RETURN 1 AS one', () => undefined)).rejects.toThrow( + /WAL-checkpoint driver is active/, + ); + } finally { + await driver.stop(); + } + await expect(streamQuery('RETURN 1 AS one', () => undefined)).rejects.toThrow( + /not initialized/, + ); + }); +}); diff --git a/gitnexus/test/unit/wal-checkpoint-driver-reentrancy.test.ts b/gitnexus/test/unit/wal-checkpoint-driver-reentrancy.test.ts index 289a5a23b..e10b55f1e 100644 --- a/gitnexus/test/unit/wal-checkpoint-driver-reentrancy.test.ts +++ b/gitnexus/test/unit/wal-checkpoint-driver-reentrancy.test.ts @@ -14,6 +14,8 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; vi.mock('../../src/core/lbug/lbug-adapter.js', () => ({ tryFlushWAL: vi.fn(), + // The driver now toggles the streamQuery guard on start/stop (#2264). + markWalDriverActive: vi.fn(), })); import { startWalCheckpointDriver } from '../../src/core/lbug/wal-checkpoint-driver.js';