diff --git a/gitnexus/src/core/group/bridge-db.ts b/gitnexus/src/core/group/bridge-db.ts index 12b10c1ec..d43a32e2a 100644 --- a/gitnexus/src/core/group/bridge-db.ts +++ b/gitnexus/src/core/group/bridge-db.ts @@ -1,4 +1,3 @@ -import type { Stats } from 'node:fs'; import fsp from 'node:fs/promises'; import path from 'node:path'; import { createHash } from 'node:crypto'; @@ -14,10 +13,10 @@ import type { } from './types.js'; import { BRIDGE_SCHEMA_QUERIES, BRIDGE_SCHEMA_VERSION } from './bridge-schema.js'; import { recordedMatchStages, recordedRepoList } from './completeness.js'; +import { closeQueryResults } from '../lbug/query-result-utils.js'; import { closeLbugConnection, openLbugConnection, - sleep, type LbugConnectionHandle, } from '../lbug/lbug-config.js'; import { dedupeContracts, dedupeCrossLinks } from './normalization.js'; @@ -228,7 +227,7 @@ async function evictBridgeEntry(key: string, entry: CachedBridgeEntry): Promise< async function closeBridgeHandle(handle: BridgeHandle): Promise { if (!handle._readOnly) { try { - await (handle._conn as lbug.Connection).query('CHECKPOINT'); + await closeQueryResults(await (handle._conn as lbug.Connection).query('CHECKPOINT')); } catch { /* ignore — older LadybugDB or schemaless DB may not accept it */ } @@ -548,7 +547,7 @@ export async function ensureBridgeSchema(handle: BridgeHandle): Promise { const conn = handle._conn as lbug.Connection; for (const q of BRIDGE_SCHEMA_QUERIES) { try { - await conn.query(q); + await closeQueryResults(await conn.query(q)); } catch (err: unknown) { const msg = err instanceof Error ? err.message : String(err); if (!msg.includes(LBUG_ALREADY_EXISTS_MSG)) throw err; @@ -563,19 +562,25 @@ export async function queryBridge( ): Promise { const run = async (): Promise => { const conn = handle._conn as lbug.Connection; + let queryResult: lbug.QueryResult | lbug.QueryResult[]; if (params && Object.keys(params).length > 0) { const stmt = await conn.prepare(cypher); if (!stmt.isSuccess()) { const errMsg = await stmt.getErrorMessage(); throw new Error(`Bridge query prepare failed: ${errMsg}`); } - const queryResult = await conn.execute(stmt, params); + queryResult = await conn.execute(stmt, params); + } else { + queryResult = await conn.query(cypher); + } + try { const result = unwrapQueryResult(queryResult); return (await result.getAll()) as T[]; + } finally { + // Native results retain the database even after conn/db.close(). Release + // them before the bridge is swapped and its file stamp is recorded. + await closeQueryResults(queryResult); } - const queryResult = await conn.query(cypher); - const result = unwrapQueryResult(queryResult); - return (await result.getAll()) as T[]; }; // Cached RO handles are shared across concurrent @group callers, so serialize // conn ops per handle (a LadybugDB Connection is not safe for concurrent @@ -669,11 +674,6 @@ export async function writeBridgeMeta(groupDir: string, meta: BridgeMeta): Promi await writeFileAtomic(path.join(groupDir, 'meta.json'), JSON.stringify(persisted, null, 2)); } -const sameFileStamp = ( - left: { size: number; mtimeMs: number }, - right: { size: number; mtimeMs: number }, -): boolean => left.size === right.size && left.mtimeMs === right.mtimeMs; - const stampMatchesStat = ( stat: { size: number; mtimeMs: number }, meta: Pick, @@ -683,33 +683,6 @@ const stampMatchesStat = ( // millisecond must still fail if their filesystem mtimes differ. stat.mtimeMs === meta.bridgeMtimeMs; -/** - * LadybugDB can still flush WAL/shadow into the main file after `close` and - * the atomic rename. Stamping the first `stat` then loses the exact-equality - * check the moment that flush lands. - * - * Wait a quiet interval before every sample — including the first — so an - * initially-stable pair is not stamped on the same tick as close/rename. Two - * consecutive agreeing stats after that interval are treated as settled. A - * flush that arrives later than the retry budget can still miss; this is a - * best-effort wait, not a lock on the file. - */ -const BRIDGE_SETTLE_MS = 10; -const BRIDGE_SETTLE_ATTEMPTS = 10; - -const statSettledBridgeFile = async (filePath: string): Promise => { - let current: Stats | undefined; - for (let i = 0; i < BRIDGE_SETTLE_ATTEMPTS; i++) { - await sleep(BRIDGE_SETTLE_MS); - const next = await fsp.stat(filePath); - if (current && sameFileStamp(next, current)) { - return next; - } - current = next; - } - return current ?? (await fsp.stat(filePath)); -}; - /** * Does `meta` still describe the `bridge.lbug` sitting next to it? * @@ -1416,7 +1389,7 @@ export async function writeBridgeUnlocked( // still belongs together (`bridgeMetaMatchesFile`). A stale meta cannot match // a freshly renamed database, and a sync that fails before the swap leaves a // matching pair untouched. - const finalStat = await statSettledBridgeFile(finalPath); + const finalStat = await fsp.stat(finalPath); await writeBridgeMeta(groupDir, { version: BRIDGE_SCHEMA_VERSION, generatedAt: new Date().toISOString(), diff --git a/gitnexus/test/unit/group/bridge-db.test.ts b/gitnexus/test/unit/group/bridge-db.test.ts index 56e70501e..557c8065f 100644 --- a/gitnexus/test/unit/group/bridge-db.test.ts +++ b/gitnexus/test/unit/group/bridge-db.test.ts @@ -1,8 +1,10 @@ import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest'; +import { spawnSync } from 'node:child_process'; import fsp from 'node:fs/promises'; import path from 'node:path'; import os from 'node:os'; import { cleanupTempDir } from '../../helpers/test-db.js'; +import { tsxLoaderUrl } from '../../helpers/cli-entry.js'; import { openBridgeDb, ensureBridgeSchema, @@ -703,6 +705,41 @@ describe("writeBridge — the swap runs inside the caller's critical section (R9 expect(await bridgeMetaMatchesFile(tmpDir, meta)).toBe(true); }, 30_000); + it('releases the native database before returning, while the writer process is still alive', async () => { + await writeBridge(tmpDir, payload('writer')); + const meta = await readBridgeMeta(tmpDir); + expect(await bridgeMetaMatchesFile(tmpDir, meta)).toBe(true); + + // Block this process while a new writer opens the published bridge. This + // prevents parent-side GC from hiding QueryResults that still own the DB. + const reader = spawnSync( + process.execPath, + [ + '--import', + tsxLoaderUrl(), + '--input-type=module', + '-e', + ` + const { openBridgeDb, queryBridge, closeBridgeDb } = await import(process.argv[1]); + const handle = await openBridgeDb(process.argv[2]); + try { + const rows = await queryBridge(handle, 'MATCH (c:Contract) RETURN c.repo AS repo'); + console.log(JSON.stringify(rows)); + } finally { + await closeBridgeDb(handle); + } + `, + new URL('../../../src/core/group/bridge-db.ts', import.meta.url).href, + path.join(tmpDir, 'bridge.lbug'), + ], + { encoding: 'utf8', timeout: 15_000 }, + ); + + expect(reader.error).toBeUndefined(); + expect(reader.status, reader.stderr).toBe(0); + expect(JSON.parse(reader.stdout)).toEqual([{ repo: 'writer' }]); + }, 30_000); + it('a writer that died mid-swap leaves state the next write recovers from', async () => { await writeBridge(tmpDir, payload('before-the-crash'));