From fb1d712657dcf3914ba2ed96137d8d64346ffe17 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gerg=C5=91=20Magyar?= Date: Wed, 15 Apr 2026 08:59:09 +0100 Subject: [PATCH] fix(lbug): wait for read stream close in splitRelCsvByLabelPair (Windows ENOTEMPTY) (#832) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(lbug): wait for read stream close in splitRelCsvByLabelPair (Windows ENOTEMPTY) The windows-latest CI job intermittently failed: FAIL test/unit/rel-csv-split.test.ts > splitRelCsvByLabelPair > handles empty CSV (header only) without errors Error: ENOTEMPTY: directory not empty, rmdir 'C:\Users\RUNNER~1\AppData\Local\Temp\rel-csv-test-XW5KOu' Cause: splitRelCsvByLabelPair resolved its Promise on readline's 'close' event, but the underlying fs.ReadStream's file descriptor is released asynchronously after that — especially on Windows. For the empty-CSV test the function returns so quickly that afterEach fires rmSync while the relations.csv fd is still held, so Windows reports ENOTEMPTY on the directory. Fixes: - Production: after readline 'close', wait for inputStream 'close' (or resolve immediately if already closed/destroyed). Call inputStream .destroy() defensively so we never hang if the fd never emits 'close'. - Test: afterEach now retries rmSync up to 5 times on ENOTEMPTY/EBUSY/ EPERM with a brief back-off — defense-in-depth so the test doesn't flake on slow CI runners independent of the production change. The production fix benefits every caller, not just the test: any code that deletes the CSV's parent directory right after the Promise resolves previously hit the same race on Windows. * refactor(lbug): replace custom stream state machines with stdlib primitives Full audit of splitRelCsvByLabelPair's stream usage after the original ENOTEMPTY fix. Replaced three hand-rolled mechanisms with their standard-library equivalents — 147 -> 71 lines in the function, and the caller's WriteStream closure dropped from 13 lines to 5. - readline: 'on(line)' + pause/resume/waitingForDrain state machine -> 'for await (const line of rl)'. Async-iterator delivery naturally serializes line processing with our awaits, so at most one ws is in backpressure at a time. We just 'await once(ws, "drain")' when 'write()' returns false — the custom Set, the settled flag and the 'only resume when all streams have drained' logic all go away. - Multi-stream error coordination: hand-rolled cleanup() that had to be entered exactly once and had to destroy the inputStream and every pair ws -> single AbortController shared across every 'once(ws, 'drain', { signal })'. Any stream error aborts every pending wait. - 'stream/promises.finished(inputStream)' in the 'finally' block replaces the manual 'rl.on('close', () => inputStream.once('close', ...))' dance, and covers both the success and error paths with the same primitive. This closes the Windows ENOTEMPTY race root cause — we never return while the fd might still be in flight. - Caller closure: 'new Promise((res, rej) => ws.end(cb) + remove listener on error)' -> 'ws.end(); await finished(ws)'. - Test 'afterEach': custom retry loop -> 'fs.rmSync(..., { maxRetries: 5, retryDelay: 50 })' (Node added these options specifically for cross-platform tmpdir cleanup). - Test 'destroys all streams when one errors': old code leaked backpressure and created multiple pair streams before the first blocked; new strict serial backpressure doesn't, so the test now unblocks the first stream once to advance the loop and create the second stream before triggering the error. --- gitnexus/src/core/lbug/lbug-adapter.ts | 134 ++++++++++------------- gitnexus/test/unit/rel-csv-split.test.ts | 16 ++- 2 files changed, 70 insertions(+), 80 deletions(-) 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'));