mirror of
https://github.com/abhigyanpatwari/GitNexus.git
synced 2026-10-28 06:22:22 +00:00
fix(lbug): wait for read stream close in splitRelCsvByLabelPair (Windows ENOTEMPTY) (#832)
* 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.
This commit is contained in:
Gergő Magyar
GitHub
parent
e11f7dfbc5
commit
fb1d712657
2 changed files with 70 additions and 80 deletions
|
|
@ -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<void>((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<string>();
|
||||
|
||||
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<void>((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;
|
||||
|
|
|
|||
|
|
@ -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'));
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue