Merge upstream/main into feat/r-language-support

This commit is contained in:
Mick Cooney 2026-10-15 17:25:44 +01:00
commit 52ea6649b0
2 changed files with 51 additions and 41 deletions

View file

@ -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<void> {
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<void> {
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<T>(
): Promise<T[]> {
const run = async (): Promise<T[]> => {
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<BridgeMeta, 'bridgeSize' | 'bridgeMtimeMs'>,
@ -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<Stats> => {
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(),

View file

@ -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'));