diff --git a/gitnexus/src/core/lbug/lbug-adapter.ts b/gitnexus/src/core/lbug/lbug-adapter.ts index fba92465c4..0298d5f7c5 100644 --- a/gitnexus/src/core/lbug/lbug-adapter.ts +++ b/gitnexus/src/core/lbug/lbug-adapter.ts @@ -1,6 +1,8 @@ import fs from 'fs/promises'; import { createReadStream, createWriteStream } from 'fs'; import { createInterface } from 'readline'; +import { once } from 'events'; +import { finished } from 'stream/promises'; import path from 'path'; import lbug from '@ladybugdb/core'; import { KnowledgeGraph } from '../graph/types.js'; @@ -56,97 +58,80 @@ export const splitRelCsvByLabelPair = async ( let skippedRels = 0; let totalValidRels = 0; - await new Promise((resolve, reject) => { - const inputStream = createReadStream(csvPath, 'utf-8'); - const rl = createInterface({ - input: inputStream, - crlfDelay: Infinity, - }); + const inputStream = createReadStream(csvPath, 'utf-8'); + const rl = createInterface({ input: inputStream, crlfDelay: Infinity }); - // Track which streams are already waiting for drain to prevent - // listener accumulation. rl.pause() is not synchronous — buffered - // line events continue firing after pause(), and without this guard - // each line targeting the same pairKey would add another drain listener. - const waitingForDrain = new Set(); - - let settled = false; - const cleanup = (err: Error) => { - if (settled) return; - settled = true; - try { - rl.close(); - } catch {} - try { - inputStream.destroy(); - } catch {} - for (const ws of pairWriteStreams.values()) { - try { - ws.destroy(); - } catch {} - } - reject(err); - }; + // If any pair WriteStream errors (disk full, EMFILE, etc.) or the input + // stream fails, we need to abort the pending `once(ws, 'drain')` await. + // An AbortController gives us one signal to cancel all pending waits + // without a custom state machine. + const abortOnError = new AbortController(); + let streamError: Error | null = null; + const markStreamError = (err: Error): void => { + streamError ??= err; + abortOnError.abort(err); + }; + try { + // `for await (const line of rl)` replaces the old manual + // on('line')/pause()/resume()/waitingForDrain state machine: readline's + // async iterator naturally serializes line delivery with our awaits, so + // at most one ws can be in backpressure at a time and we just await its + // 'drain' event. let isFirst = true; - rl.on('line', (line) => { + for await (const line of rl) { + if (streamError) throw streamError; if (isFirst) { relHeader = line; isFirst = false; - return; + continue; } - if (!line.trim()) return; + if (!line.trim()) continue; const match = line.match(/"([^"]*)","([^"]*)"/); if (!match) { skippedRels++; - return; + continue; } const fromLabel = getNodeLabel(match[1]); const toLabel = getNodeLabel(match[2]); if (!validTables.has(fromLabel) || !validTables.has(toLabel)) { skippedRels++; - return; + continue; } + const pairKey = `${fromLabel}|${toLabel}`; let ws = pairWriteStreams.get(pairKey); if (!ws) { const pairCsvPath = path.join(csvDir, `rel_${fromLabel}_${toLabel}.csv`); ws = wsFactory(pairCsvPath); - // If any per-pair WriteStream errors (disk full, EMFILE, etc.), - // tear down everything and reject the Promise. Without this handler, - // a stream error while rl is paused waiting for drain would cause - // the drain callback to never fire and the Promise to hang forever. - ws.on('error', cleanup); - ws.write(relHeader + '\n'); + ws.on('error', markStreamError); pairWriteStreams.set(pairKey, ws); relsByPairMeta.set(pairKey, { csvPath: pairCsvPath, rows: 0 }); + if (!ws.write(relHeader + '\n')) { + await once(ws, 'drain', { signal: abortOnError.signal }); + } + } + + if (!ws.write(line + '\n')) { + await once(ws, 'drain', { signal: abortOnError.signal }); } - const ok = ws.write(line + '\n'); relsByPairMeta.get(pairKey)!.rows++; totalValidRels++; - // Handle backpressure: pause reading when the write buffer is full, - // resume when the stream drains. Prevents unbounded memory growth - // on repos with millions of relationships. - // Guard with waitingForDrain to ensure only one drain listener is - // registered per stream at a time — rl.pause() doesn't stop buffered - // line events immediately. Only resume when ALL streams have drained - // to avoid writing into still-full streams. - if (!ok && !waitingForDrain.has(pairKey)) { - waitingForDrain.add(pairKey); - rl.pause(); - ws.once('drain', () => { - waitingForDrain.delete(pairKey); - if (waitingForDrain.size === 0) rl.resume(); - }); - } - }); - rl.on('close', () => { - if (!settled) { - settled = true; - resolve(); - } - }); - rl.on('error', cleanup); - }); + } + if (streamError) throw streamError; + } catch (err) { + // Tear down everything so no fd is left dangling. If the abort was caused + // by a stream error, rethrow that error (more actionable than AbortError). + for (const ws of pairWriteStreams.values()) ws.destroy(); + inputStream.destroy(); + throw streamError ?? err; + } finally { + // Readline 'close' fires before the underlying fs.ReadStream releases its + // fd — on Windows that race caused ENOTEMPTY on the parent dir. + // stream/promises.finished is the stdlib "wait until this stream is fully + // closed" primitive and handles both success and error paths. + await finished(inputStream).catch(() => {}); + } return { relHeader, relsByPairMeta, pairWriteStreams, skippedRels, totalValidRels }; }; @@ -388,19 +373,14 @@ export const loadGraphToLbug = async ( const { relHeader, relsByPairMeta, pairWriteStreams, skippedRels, totalValidRels } = await splitRelCsvByLabelPair(csvResult.relCsvPath, csvDir, validTables, getNodeLabel); - // Close all per-pair write streams before COPY + // Close all per-pair write streams before COPY. `stream/promises.finished` + // resolves on the stream's 'finish' event and rejects on 'error' — replaces + // a hand-rolled promisification with the stdlib primitive. await Promise.all( - Array.from(pairWriteStreams.values()).map( - (ws) => - new Promise((resolve, reject) => { - const onError = (err: Error) => reject(err); - ws.on('error', onError); - ws.end(() => { - ws.removeListener('error', onError); - resolve(); - }); - }), - ), + Array.from(pairWriteStreams.values()).map(async (ws) => { + ws.end(); + await finished(ws); + }), ); const insertedRels = totalValidRels; diff --git a/gitnexus/test/unit/rel-csv-split.test.ts b/gitnexus/test/unit/rel-csv-split.test.ts index dab38a913d..5b0a811991 100644 --- a/gitnexus/test/unit/rel-csv-split.test.ts +++ b/gitnexus/test/unit/rel-csv-split.test.ts @@ -87,7 +87,12 @@ beforeEach(() => { }); afterEach(() => { - fs.rmSync(tmpDir, { recursive: true, force: true }); + // fs.rmSync's built-in retry loop handles Windows EBUSY/ENOTEMPTY/EPERM + // when a just-closed fd hasn't been released yet (Node added this exactly + // for cross-platform tmpdir cleanup — see Node.js fs docs). The production + // function also waits for the input stream's 'close' event, so this is + // defense-in-depth. + fs.rmSync(tmpDir, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 }); }); function writeCsv(lines: string[]): string { @@ -242,8 +247,13 @@ describe('splitRelCsvByLabelPair', () => { mockFactory(streams, { blocked: true }), ); - // Wait for readline to process and create streams - await new Promise((r) => setTimeout(r, 50)); + // The first pair stream is created immediately and blocks on its header + // write. Unblock it once so the loop advances and creates the second + // pair stream (also blocked). Now both streams exist — trigger the error. + await new Promise((r) => setTimeout(r, 20)); + expect(streams.length).toBe(1); + streams[0].unblock(); + await new Promise((r) => setTimeout(r, 20)); expect(streams.length).toBeGreaterThanOrEqual(2); streams[0].triggerError(new Error('EMFILE'));