fix: address review feedback — watchdog, TOCTOU race, null guard, platform guard

- Watchdog timer now exempts in-flight queries via activeQueryCount,
  preventing premature stdout restoration during long queries (>1s)
- Stale detection uses reinitPromises Map to prevent TOCTOU race where
  concurrent callers double-close the connection pool
- Throttle meta.json staleness checks to once per 5s per repo
- Add null guard for i.id in IN-clause construction
- Enrichment queries run in parallel on non-arm64 platforms to preserve
  performance; sequential only on arm64 macOS where SIGSEGV occurs
This commit is contained in:
hiromima 2026-03-20 16:52:33 +09:00
parent 999fbf5b11
commit 893f77ae89
2 changed files with 62 additions and 16 deletions

View file

@ -148,6 +148,8 @@ function closeOne(repoId: string): void {
* Create a new Connection from a repo's Database.
* Silences stdout to prevent native module output from corrupting MCP stdio.
*/
let activeQueryCount = 0;
function silenceStdout(): void {
if (stdoutSilenceCount++ === 0) {
process.stdout.write = (() => true) as any;
@ -163,8 +165,10 @@ function restoreStdout(): void {
// Safety watchdog: restore stdout if it gets stuck silenced (e.g. native crash
// inside createConnection before restoreStdout runs).
// Exempts active queries and pre-warm — these legitimately hold silence for
// longer than 1 second (queries can take up to QUERY_TIMEOUT_MS = 30s).
setInterval(() => {
if (stdoutSilenceCount > 0 && !preWarmActive) {
if (stdoutSilenceCount > 0 && !preWarmActive && activeQueryCount === 0) {
stdoutSilenceCount = 0;
process.stdout.write = realStdoutWrite;
}
@ -459,12 +463,14 @@ export const executeQuery = async (repoId: string, cypher: string): Promise<any[
const conn = await checkout(entry);
silenceStdout();
activeQueryCount++;
try {
const queryResult = await withTimeout(conn.query(cypher), QUERY_TIMEOUT_MS, 'Query');
const result = Array.isArray(queryResult) ? queryResult[0] : queryResult;
const rows = await result.getAll();
return rows;
} finally {
activeQueryCount--;
restoreStdout();
checkin(entry, conn);
}
@ -488,6 +494,7 @@ export const executeParameterized = async (
const conn = await checkout(entry);
silenceStdout();
activeQueryCount++;
try {
const stmt = await withTimeout(conn.prepare(cypher), QUERY_TIMEOUT_MS, 'Prepare');
if (!stmt.isSuccess()) {
@ -499,6 +506,7 @@ export const executeParameterized = async (
const rows = await result.getAll();
return rows;
} finally {
activeQueryCount--;
restoreStdout();
checkin(entry, conn);
}

View file

@ -88,6 +88,8 @@ export class LocalBackend {
private repos: Map<string, RepoHandle> = new Map();
private contextCache: Map<string, CodebaseContext> = new Map();
private initializedRepos: Set<string> = new Set();
private reinitPromises: Map<string, Promise<void>> = new Map();
private lastStalenessCheck: Map<string, number> = new Map();
// ─── Initialization ──────────────────────────────────────────────
@ -246,22 +248,43 @@ export class LocalBackend {
// ─── Lazy LadybugDB Init ────────────────────────────────────────────
private async ensureInitialized(repoId: string): Promise<void> {
// If a reinit is already in progress for this repo, wait for it
const pending = this.reinitPromises.get(repoId);
if (pending) return pending;
const handle = this.repos.get(repoId);
if (!handle) throw new Error(`Unknown repo: ${repoId}`);
// Check if the index was rebuilt since we opened the connection (#297).
// Read meta.json's indexedAt and compare to the cached value — if it
// changed, close the stale pool and re-initialize with the fresh index.
// Throttle staleness checks to at most once per 5 seconds per repo to
// avoid an fs.readFile round-trip on every tool invocation.
if (this.initializedRepos.has(repoId) && isLbugReady(repoId)) {
const now = Date.now();
const lastCheck = this.lastStalenessCheck.get(repoId) ?? 0;
if (now - lastCheck < 5000) return; // Checked recently — skip
this.lastStalenessCheck.set(repoId, now);
try {
const metaPath = path.join(handle.storagePath, 'meta.json');
const metaRaw = await fs.readFile(metaPath, 'utf-8');
const meta = JSON.parse(metaRaw);
if (meta.indexedAt && meta.indexedAt !== handle.indexedAt) {
// Index was rebuilt — close stale connection and re-init
await closeLbug(repoId);
this.initializedRepos.delete(repoId);
handle.indexedAt = meta.indexedAt;
// Index was rebuilt — close stale connection and re-init.
// Wrap in reinitPromises to prevent TOCTOU race where concurrent
// callers both detect staleness and double-close the pool.
const reinit = (async () => {
try {
await closeLbug(repoId);
this.initializedRepos.delete(repoId);
handle.indexedAt = meta.indexedAt;
await initLbug(repoId, handle.lbugPath);
this.initializedRepos.add(repoId);
} finally {
this.reinitPromises.delete(repoId);
}
})();
this.reinitPromises.set(repoId, reinit);
return reinit;
} else {
return; // Pool is current
}
@ -1459,33 +1482,48 @@ export class LocalBackend {
// Cap IN-clause to 100 IDs to prevent oversized queries that crash
// the native DB engine on arm64 macOS (#292)
const cappedImpacted = impacted.slice(0, 100);
const allIds = cappedImpacted.map(i => `'${i.id.replace(/'/g, "''")}'`).join(', ');
const allIds = cappedImpacted.map(i => `'${String(i.id ?? '').replace(/'/g, "''")}'`).join(', ');
const d1Items = (grouped[1] || []).slice(0, 100);
const d1Ids = d1Items.map((i: any) => `'${i.id.replace(/'/g, "''")}'`).join(', ');
const d1Ids = d1Items.map((i: any) => `'${String(i.id ?? '').replace(/'/g, "''")}'`).join(', ');
// Run enrichment queries sequentially to avoid concurrent native DB
// access that causes SIGSEGV on arm64 macOS (#285, #290, #292)
const processRows = await executeQuery(repo.id, `
// Enrichment queries: sequential on arm64 macOS to avoid SIGSEGV from
// concurrent native DB access (#285, #290, #292); parallel elsewhere
// to preserve performance on unaffected platforms.
const isArm64Mac = process.platform === 'darwin' && process.arch === 'arm64';
const processQuery = executeQuery(repo.id, `
MATCH (s)-[r:CodeRelation {type: 'STEP_IN_PROCESS'}]->(p:Process)
WHERE s.id IN [${allIds}]
RETURN p.heuristicLabel AS name, COUNT(DISTINCT s.id) AS hits, MIN(r.step) AS minStep, p.stepCount AS stepCount
ORDER BY hits DESC
LIMIT 20
`).catch(() => []);
const moduleRows = await executeQuery(repo.id, `
const moduleQuery = () => executeQuery(repo.id, `
MATCH (s)-[:CodeRelation {type: 'MEMBER_OF'}]->(c:Community)
WHERE s.id IN [${allIds}]
RETURN c.heuristicLabel AS name, COUNT(DISTINCT s.id) AS hits
ORDER BY hits DESC
LIMIT 20
`).catch(() => []);
const directModuleRows = d1Ids
? await executeQuery(repo.id, `
const directModuleQuery = () => d1Ids
? executeQuery(repo.id, `
MATCH (s)-[:CodeRelation {type: 'MEMBER_OF'}]->(c:Community)
WHERE s.id IN [${d1Ids}]
RETURN DISTINCT c.heuristicLabel AS name
`).catch(() => [])
: [];
: Promise.resolve([]);
let processRows: any[], moduleRows: any[], directModuleRows: any[];
if (isArm64Mac) {
// Sequential: avoid concurrent native DB access
processRows = await processQuery;
moduleRows = await moduleQuery();
directModuleRows = await directModuleQuery();
} else {
// Parallel: safe on non-arm64 platforms
processRows = await processQuery;
[moduleRows, directModuleRows] = await Promise.all([moduleQuery(), directModuleQuery()]);
}
affectedProcesses = processRows.map((r: any) => ({
name: r.name || r[0],