fix: prevent drain listener leak in relationship CSV streaming

[MekayelAnik]

Problem

When analyzing large repositories, GitNexus floods stderr with hundreds of MaxListenersExceededWarning messages:

MaxListenersExceededWarning: Possible EventEmitter memory leak detected.
11 drain listeners added to [WriteStream]. MaxListeners is 10.

Root Cause

In loadGraphToLbug(), relationship CSV lines are split into per-label-pair files (rel_*.csv). When backpressure occurs (ws.write() returns false), the code calls rl.pause() and adds a ws.once('drain', ...) listener. However, rl.pause() is not synchronous — buffered line events continue firing, and each one targeting the same pair stream adds another drain listener before the first fires. This causes unbounded listener accumulation.

Additionally, per-pair WriteStreams had no error handlers, meaning a stream error (disk full, EMFILE) while waiting for drain would hang the Promise forever, leaking all open file descriptors.

Fix

• waitingForDrain guard — A Set<string> tracks which streams already have a pending drain listener. Only the first backpressure event registers the listener; subsequent writes to the same stream are skipped. This keeps drain listeners at exactly 1 per stream at all times, eliminating the warning entirely without needing setMaxListeners.
• Error handlers on WriteStreams — Each per-pair stream gets ws.on('error', cleanup). The shared cleanup() function tears down all streams (with try/catch for already-destroyed streams), closes the readline interface and input ReadStream, and rejects the Promise.
• Error handler on ws.end() phase — Uses ws.on('error', onError) with a named function so removeListener correctly matches after successful close. Prevents Promise.all from hanging on flush errors.
• rl.close() + inputStream.destroy() in error handler — Prevents ReadStream FD from lingering until GC.

Why no setMaxListeners?

Tested with 200 unique label pairs and 100k total lines — max drain listeners on any single stream was always 1, zero warnings fired. The waitingForDrain guard is the proper fix; setMaxListeners would just mask the symptom.

Testing

8 regression tests in test/unit/rel-csv-split.test.ts:

• Happy path: correct splitting, header capture, unknown label skipping, blank line handling
• Bug 1: WriteStream error rejects Promise and destroys all streams
• Bug 2: drain listener count stays at max 1 under heavy backpressure
• Empty CSV edge case

Tests use a MockWriteStream with controllable backpressure and error injection.

[vercel]

@MekayelAnik is attempting to deploy a commit to the NexusCore Team on Vercel.

A member of the Team first needs to authorize it.

[Copilot]

Pull request overview

Reduces MaxListenersExceededWarning noise during large-repository analysis by increasing the EventEmitter listener cap on per-relationship-pair WriteStreams created during relationship CSV splitting.

Changes:

• Set ws.setMaxListeners(50) immediately after creating each per-pair relationship WriteStream in loadGraphToLbug().

Comments suppressed due to low confidence (1)

gitnexus/src/core/lbug/lbug-adapter.ts:292

• Raising the stream’s max listener cap suppresses the warning, but it doesn’t address the underlying cause: when ws.write() returns false, this code can still enqueue multiple once('drain') listeners for the same stream before rl.pause() takes effect. Consider coalescing backpressure handling so only a single pending drain-resume callback can be registered per WriteStream (e.g., track a per-stream waitingForDrain flag and only attach the listener once, clearing it on drain). This avoids unbounded listener growth and extra closures under heavy backpressure, rather than just increasing the threshold.

        ws.setMaxListeners(50);
        ws.write(relHeader + '\n');
        pairWriteStreams.set(pairKey, ws);
        relsByPairMeta.set(pairKey, { csvPath: pairCsvPath, rows: 0 });
      }

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[magyargergo]

@claude try to identify if there's a memory leak in the pipeline. The fix temporarily fixes the issue but I think, we are leaking open file descriptors or something like that.

[github-actions]

Claude finished @magyargergo's task in 2m 5s —— View job

---

Memory / FD Leak Analysis — loadGraphToLbug()

• Gather context on the PR and trigger request
• Read lbug-adapter.ts to understand the WriteStream lifecycle
• Check for proper stream close/destroy on all code paths (errors, early exits)
• Trace file descriptor ownership through loadGraphToLbug()
• Check if pairWriteStreams Map is properly drained/closed
• Look for error handling gaps that could leave FDs open
• Compare with BufferedCSVWriter pattern in csv-generator.ts
• Report findings

---

Verdict: Yes, there is a real FD / Promise hang bug

The setMaxListeners(50) in this PR suppresses the warning but does not fix the underlying problems. Here are the confirmed issues, ordered by severity:

---

Bug 1 — No error handler on per-pair WriteStreams (FD + Promise leak) 🔴

File: gitnexus/src/core/lbug/lbug-adapter.ts, lines 287–302

ws = createWriteStream(pairCsvPath, 'utf-8');
ws.setMaxListeners(50);          // ← this PR
ws.write(relHeader + '\n');
// ← no ws.on('error', ...) handler anywhere

// ...later in the line handler:
const ok = ws.write(line + '\n');
if (!ok) {
  rl.pause();
  ws.once('drain', () => rl.resume());  // ← if ws errors, drain NEVER fires
}

If any per-pair WriteStream emits 'error' (disk full, permission error, EMFILE too many open files) while rl is paused waiting for drain:

• drain never fires → rl.resume() is never called
• rl stays paused forever
• The outer Promise<void> at line 259 never resolves or rejects
• All entries in pairWriteStreams — potentially hundreds of open file descriptors — stay open until the process is killed

Compare with BufferedCSVWriter.flush() in csv-generator.ts (line 174–186), which correctly pairs every write with an error listener and cleans it up:

return new Promise((resolve, reject) => {
  this.ws.once('error', reject);       // ← error handler registered
  const ok = this.ws.write(chunk);
  if (ok) {
    this.ws.removeListener('error', reject);  // ← cleaned up on success
    resolve();
  } else {
    this.ws.once('drain', () => {
      this.ws.removeListener('error', reject);  // ← cleaned up on drain
      resolve();
    });
  }
});

The per-pair streams in loadGraphToLbug() don't follow this pattern at all. Fix this →

---

Bug 2 — waitingForDrain flag missing: multiple drain listeners stack up (root cause of MaxListeners warning) 🟠

File: gitnexus/src/core/lbug/lbug-adapter.ts, lines 299–302

if (!ok) {
  rl.pause();
  ws.once('drain', () => rl.resume());  // ← added N times for same ws
}

rl.pause() is not synchronous — readline's internal buffer continues delivering already-queued 'line' events after pause() is called. For lines that all map to the same pairKey, each one appends another ws.once('drain', ...) listener before the first drain fires. This is the root cause of the MaxListenersExceededWarning.

The PR raises the cap to 50 so the warning stops appearing, but the underlying accumulation still happens:

• Up to 50 closures are kept alive in memory per stream until drain fires
• Each closure holds a reference to rl, preventing GC
• When drain finally fires, rl.resume() is called up to 50 times (all no-ops after the first, but wasted work)

The fix is a waitingForDrain flag per stream — exactly what Copilot noted in its review. Fix this →

---

Bug 3 — rl not closed in the error handler: ReadStream FD may leak 🟡

File: gitnexus/src/core/lbug/lbug-adapter.ts, lines 305–309

rl.on('error', (err) => {
  for (const ws of pairWriteStreams.values()) ws.destroy();
  // ← rl itself is never closed/destroyed here
  reject(err);
});

When readline errors, the write streams are destroyed (good), but rl is not explicitly closed. The input ReadStream created at line 261 is not destroyed either. While Node.js stream internals will eventually clean up, not calling rl.close() / rl.input.destroy() means the FD from createReadStream(csvResult.relCsvPath) may linger until GC collects the abandoned Interface object. Fix this →

---

Summary

#	Issue	Severity	This PR fixes it?
1	No error handler on per-pair WriteStreams → hung Promise + all FDs open forever	🔴 High	❌ No
2	No waitingForDrain flag → N drain listeners accumulate per stream	🟠 Medium	❌ No (only masks symptom)
3	rl not closed in error handler → ReadStream FD may linger	🟡 Low	❌ No

Recommendation: This PR should not be merged as-is. The setMaxListeners(50) change hides bug #2 without fixing it, and leaves bug #1 (the actual FD leak) entirely unaddressed. The correct fix is to:

1. Add a ws.on('error', ...) / ws.once('error', reject) handler on each per-pair stream (and call rl.destroy() inside it to break the hung-Promise scenario)
2. Guard ws.once('drain', ...) with a per-stream waitingForDrain Set so only one listener is ever queued at a time
3. Add rl.close() in the readline error handler

The pattern in BufferedCSVWriter (csv-generator.ts lines 149–196) already does this correctly and should be the model.

[magyargergo]

@MekayelAnik could you please look into this?

[MekayelAnik]

Thanks @magyargergo for the thorough analysis! The Claude CI review was spot-on — the initial setMaxListeners(50) was only masking the symptoms.

I've pushed a second commit (eaf4337) that addresses all 3 bugs identified:

Bug 1 (High) — No error handler on per-pair WriteStreams ✅ Fixed

Added a shared cleanup() function that each WriteStream's error handler calls. On any stream error (disk full, EMFILE, etc.), it now:

• Destroys all open WriteStreams
• Closes the readline interface (rl.close())
• Destroys the input ReadStream (inputStream.destroy())
• Rejects the Promise (no more hung-forever scenario)

Uses a settled flag to prevent double-rejection if multiple streams error simultaneously.

Bug 2 (Medium) — No waitingForDrain guard ✅ Fixed

Added a waitingForDrain: Set<string> that tracks which pairKey streams already have a pending drain listener. The backpressure block now reads:

if (!ok && !waitingForDrain.has(pairKey)) {
  waitingForDrain.add(pairKey);
  rl.pause();
  ws.once('drain', () => {
    waitingForDrain.delete(pairKey);
    rl.resume();
  });
}

This ensures only one drain listener per stream at a time, eliminating listener accumulation entirely. setMaxListeners(50) is now a safety net, not a band-aid.

Bug 3 (Low) — rl not closed in error handler ✅ Fixed

The readline error handler now uses the same cleanup() function, which calls rl.close() and inputStream.destroy() before rejecting — no more lingering ReadStream FDs.

All 3 fixes follow the pattern already established by BufferedCSVWriter in csv-generator.ts.

[magyargergo]

@claude try to find potential bugs, performance issues, potential leaks and if it resolves the issue properly

[github-actions]

Claude finished @magyargergo's task in 2m 19s —— View job

---

Code Review — Bug, Leak & Performance Analysis

• Gather context on the PR and trigger request
• Read lbug-adapter.ts current state (including second commit fixes)
• Analyze error handling on per-pair WriteStreams
• Analyze waitingForDrain guard correctness
• Analyze readline/ReadStream FD cleanup
• Check for any remaining bugs, performance issues, or leaks
• Report findings

---

Verdict on the Previous 3 Bugs

All 3 bugs identified in the prior Claude analysis are correctly fixed in commit eaf4337. Here's the confirmation:

#	Issue	Fixed?	How
1	No error handler → hung Promise + FD leak	✅	ws.on('error', cleanup) with shared cleanup() that destroys all streams and rejects the Promise
2	No waitingForDrain guard → listener accumulation	✅	waitingForDrain: Set<string> guards ws.once('drain', ...) so only 1 listener per stream at a time
3	rl not closed in error handler	✅	cleanup() calls rl.close() and inputStream.destroy() before rejecting

The original MaxListenersExceededWarning is now addressed at the root cause level, not just masked. Good work.

---

Remaining Bug Found — ws.end() has no error handler: Promise.all can hang forever 🔴

File: gitnexus/src/core/lbug/lbug-adapter.ts, lines 346–353

await Promise.all(
  Array.from(pairWriteStreams.values()).map(
    (ws) =>
      new Promise<void>((resolve, reject) =>
        ws.end((err: Error | undefined) => (err ? reject(err) : resolve())),
      ),
  ),
);

The problem: ws.on('error', cleanup) is still registered on every stream at this point, but cleanup is a no-op because settled = true. In Node.js, when a WriteStream emits 'error' (e.g., during flush on ws.end()), the callback passed to ws.end() is never called — only the 'error' event fires. Since cleanup is a no-op and no new reject path exists, the inner Promise<void> never settles, and Promise.all hangs indefinitely. Any error during the final flush phase (disk full, EMFILE, etc.) would freeze the entire analysis.

Fix: Add a one-time error handler that rejects the inner Promise:

await Promise.all(
  Array.from(pairWriteStreams.values()).map(
    (ws) =>
      new Promise<void>((resolve, reject) => {
        ws.once('error', reject);
        ws.end((err: Error | undefined) => {
          ws.removeListener('error', reject);
          err ? reject(err) : resolve();
        });
      }),
  ),
);

Fix this →

---

Minor Design Concern — Multi-stream backpressure coordination 🟡

File: gitnexus/src/core/lbug/lbug-adapter.ts, lines 327–334

if (!ok && !waitingForDrain.has(pairKey)) {
  waitingForDrain.add(pairKey);
  rl.pause();
  ws.once('drain', () => {
    waitingForDrain.delete(pairKey);
    rl.resume();   // ← resumes even if OTHER streams are still full
  });
}

If stream A and stream B are simultaneously under backpressure (both in waitingForDrain), when A drains it calls rl.resume() even though B is still full. Lines continue flowing and accumulate in B's OS write buffer until B drains. For repos with many very high-traffic relationship types, this can cause unbounded memory growth during the streaming phase.

A complete fix would be to resume only when all streams have drained:

ws.once('drain', () => {
  waitingForDrain.delete(pairKey);
  if (waitingForDrain.size === 0) rl.resume();  // only resume when fully clear
});

This is a design concern for very large repos, not a correctness bug for typical usage.

---

Summary

The PR (after commit eaf4337) correctly fixes the 3 previously identified bugs. One new bug remains — the ws.end() phase has no error handler, so a late flush error would cause a permanent hang. This should be fixed before merging. The backpressure coordination concern is a lower-priority improvement.

[github-actions]

CI Report

✅ All checks passed

Pipeline Status

Stage	Status	Details
✅ Typecheck	success	tsc --noEmit
✅ Tests	success	unit tests, 3 platforms
✅ E2E	success	gitnexus-web changes only

Test Results

Tests	Passed	Failed	Skipped	Duration
6333	6233	0	98	255s

✅ All 6233 tests passed

98 test(s) skipped — expand for details

• Swift MethodExtractor > isTypeDeclaration > recognizes class_declaration
• Swift MethodExtractor > isTypeDeclaration > recognizes protocol_declaration
• Swift MethodExtractor > isTypeDeclaration > rejects import_declaration
• Swift MethodExtractor > visibility > extracts public method
• Swift MethodExtractor > visibility > extracts private method
• Swift MethodExtractor > visibility > defaults to internal when no modifier
• Swift MethodExtractor > protocol methods > marks protocol method as abstract
• Swift MethodExtractor > static and class methods > detects static func as isStatic
• Swift MethodExtractor > static and class methods > detects class func as isStatic
• Swift MethodExtractor > parameters > extracts parameters with types and default values
• Swift MethodExtractor > return type > extracts return type from -> annotation
• Swift MethodExtractor > annotations > extracts @objc attribute
• Swift MethodExtractor > isFinal > detects final func
• Swift MethodExtractor > isFinal > is false when not final
• Swift MethodExtractor > isAsync > detects async func
• Swift MethodExtractor > isOverride > detects override method
• buildTypeEnv > constructor inference (Tier 1 fallback) > lookupClassByName regression coverage > Swift lookupClassByName regression coverage > Swift cross-file constructor inference uses lookupClassByName
• buildTypeEnv > constructor inference (Tier 1 fallback) > lookupClassByName regression coverage > Swift lookupClassByName regression coverage > Swift explicit init inference uses lookupClassByName
• buildTypeEnv > constructor inference (Tier 1 fallback) > lookupClassByName regression coverage > Swift lookupClassByName regression coverage > Swift cross-file constructor inference does not bind plain functions
• buildTypeEnv > known limitations (documented skip tests) > Ruby block parameter: users.each { |user| } — closure param inference, different feature
• Swift constructor-inferred type resolution > detects User and Repo classes, both with save methods
• Swift constructor-inferred type resolution > resolves user.save() to Models/User.swift via constructor-inferred type
• Swift constructor-inferred type resolution > resolves repo.save() to Models/Repo.swift via constructor-inferred type
• Swift constructor-inferred type resolution > emits exactly 2 save() CALLS edges (one per receiver type)
• Swift self resolution > detects User and Repo classes, each with a save function
• Swift self resolution > resolves self.save() inside User.process to User.save, not Repo.save
• Swift parent resolution > detects BaseModel and User classes plus Serializable protocol
• Swift parent resolution > emits EXTENDS edge: User → BaseModel
• Swift parent resolution > emits IMPLEMENTS edge: User → Serializable (protocol conformance)
• Swift cross-file User.init() inference > resolves user.save() via User.init(name:) inference
• Swift cross-file User.init() inference > resolves user.greet() via User.init(name:) inference
• Swift return type inference > detects User class and getUser function
• Swift return type inference > detects save function on User (Swift class methods are Function nodes)
• Swift return type inference > resolves user.save() to User#save via return type of getUser() -> User
• Swift return-type inference via function return type > resolves user.save() to User#save via return type of getUser()
• Swift return-type inference via function return type > user.save() does NOT resolve to Repo#save
• Swift return-type inference via function return type > resolves repo.save() to Repo#save via return type of getRepo()
• Swift implicit imports (cross-file visibility) > detects UserService class in Models.swift
• Swift implicit imports (cross-file visibility) > resolves UserService() constructor call across files (no explicit import)
• Swift implicit imports (cross-file visibility) > resolves service.fetchUser() member call across files
• Swift implicit imports (cross-file visibility) > creates IMPORTS edges between files in the same module
• Swift extension deduplication > detects Product class
• Swift extension deduplication > resolves Product() constructor despite extension creating duplicate class node
• Swift extension deduplication > resolves product.save() to Product.swift (primary definition)
• Swift constructor call fallback (no new keyword) > resolves OCRService() as constructor call across files
• Swift constructor call fallback (no new keyword) > resolves ocr.recognize() member call via constructor-inferred type
• Swift export visibility (internal vs private) > resolves PublicService() constructor across files
• Swift export visibility (internal vs private) > resolves internalHelper() across files (internal = module-scoped)
• Swift if let / guard let binding resolution > detects User and Repo classes
• Swift if let / guard let binding resolution > resolves user.save() inside if-let to User#save
• Swift if let / guard let binding resolution > resolves repo.save() inside guard-let to Repo#save
• Swift if let / guard let binding resolution > user.save() in if-let does NOT resolve to Repo#save
• Swift await / try expression unwrapping > resolves user.save() via await fetchUser() return type
• Swift await / try expression unwrapping > resolves repo.save() via try parseRepo() return type
• Swift await / try expression unwrapping > detects fetchUser and parseRepo as functions
• Swift for-in loop element type inference > detects User and Repo classes
• Swift for-in loop element type inference > creates implicit import edges between files
• Swift field-type resolution > detects classes and their properties
• Swift field-type resolution > emits HAS_PROPERTY edges from class to field
• Swift field-type resolution > resolves field-chain call user.address.save() → Address#save
• Swift field-type resolution > emits ACCESSES edges for field reads in chains
• Swift field-type resolution > populates field metadata (visibility, declaredType) on Property nodes
• Swift call-result binding > resolves call-result-bound method call user.save() → User#save
• Swift call-result binding > getUser() is present as a defined function
• Swift call-result binding > emits processUser -> getUser CALLS edge for let-assigned free function call
• Swift method enrichment > detects Animal protocol and Dog class
• Swift method enrichment > emits IMPLEMENTS edge Dog -> Animal
• Swift method enrichment > emits HAS_METHOD edges for Dog methods
• Swift method enrichment > marks protocol Animal.speak as isAbstract
• Swift method enrichment > marks Dog.speak as NOT isAbstract
• Swift method enrichment > marks breathe as isFinal
• Swift method enrichment > marks classify as isStatic
• Swift method enrichment > captures @objc annotation on breathe
• Swift method enrichment > populates parameterTypes for classify(_ name: String)
• Swift method enrichment > records parameterCount for classify
• Swift method enrichment > records returnType for speak
• Swift method enrichment > resolves dog.speak() CALLS edge
• Swift method enrichment > resolves Dog.classify("dog") CALLS edge
• Swift abstract dispatch > detects Repository protocol and SqlRepository class
• Swift abstract dispatch > emits IMPLEMENTS edge SqlRepository -> Repository
• Swift abstract dispatch > emits HAS_METHOD edges for Repository.find and Repository.save
• Swift abstract dispatch > emits HAS_METHOD edges for SqlRepository.find and SqlRepository.save
• Swift abstract dispatch > marks base Repository.find as isAbstract
• Swift abstract dispatch > marks base Repository.save as isAbstract
• Swift abstract dispatch > marks concrete SqlRepository.find as NOT isAbstract
• Swift abstract dispatch > resolves repo.find(id: 42) CALLS edge
• Swift abstract dispatch > resolves repo.save(entity: user) CALLS edge
• Swift abstract dispatch > populates parameterTypes for Repository.find
• Swift abstract dispatch > populates parameterTypes for Repository.save
• Swift abstract dispatch > records returnType for SqlRepository.find
• Swift abstract dispatch > emits METHOD_IMPLEMENTS edges from SqlRepository methods → Repository protocol methods
• Swift overloaded method disambiguation > detects 2 distinct find Method nodes on SqlRepository
• Swift overloaded method disambiguation > emits METHOD_IMPLEMENTS edges for both find overloads
• Swift overloaded method disambiguation > emits METHOD_IMPLEMENTS edge for save
• Swift overloaded method disambiguation > emits exactly 3 METHOD_IMPLEMENTS edges total
• Swift Child extends Parent — inherited method resolution (SM-9) > detects Parent and Child classes
• Swift Child extends Parent — inherited method resolution (SM-9) > resolves c.parentMethod() to Parent.parentMethod via first-wins MRO walk
• bridge-db edge cases > test_openBridgeDbReadOnly_bak_recovery_restores_bridge
• bridge-db edge cases > test_writeBridge_crossLink_with_missing_to_node_silently_skipped
• bridge-db edge cases > test_writeBridge_manifest_grpc_link_with_symbol_uids_persists_queryable_contract_edge

Code Coverage

Tests

Metric	Coverage	Covered	Base	Delta	Status
Statements	72.87%	16850/23123	72.81%	📈 +0.1	🟢 ██████████████░░░░░░
Branches	61.74%	10721/17362	61.73%	📈 +0.0	🟢 ████████████░░░░░░░░
Functions	78.2%	1579/2019	78.11%	📈 +0.1	🟢 ███████████████░░░░░
Lines	75.26%	15301/20330	75.21%	📈 +0.1	🟢 ███████████████░░░░░

---

_{📋 View full run · Generated by CI}

[magyargergo]

Could you please address the remaining findings? Also please make sure the ci is green. Plus we need to find a way to test this.

[magyargergo]

@claude we need a way to test this reliably so it wont regress in the future

[github-actions]

Claude finished @magyargergo's task in 12m 59s —— View job

---

Regression Tests for Stream Backpressure Fixes

• Gather context on the PR and trigger request
• Read lbug-adapter.ts and understand testable surface
• Identify CI failure — pre-existing Prettier issue in gitnexus-web/ (unrelated to this PR); typecheck passes ✅
• Extract splitRelCsvByLabelPair as an exported, testable function with optional wsFactory DI param
• Fix remaining bug: add ws.once('error', reject) + ws.removeListener to ws.end() phase (Bug 3 — Promise.all hang)
• Write 8 regression tests covering all 3 bugs
• Commit locally
• Push changes — blocked: GitHub Actions bot lacks write permission to this repo

---

What Was Done

gitnexus/src/core/lbug/lbug-adapter.ts — two changes:

1. Extracted splitRelCsvByLabelPair as a standalone exported async function (with export type WriteStreamFactory). loadGraphToLbug now delegates to it — the inline 90-line block is replaced by a single call. No behavioral change for production; the function accepts an optional wsFactory dependency-injection parameter (default: real createWriteStream) so tests can inject mock streams.

2. Fixed Bug 3 — the ws.end() phase now uses:

   ws.once('error', reject);
   ws.end(() => { ws.removeListener('error', reject); resolve(); });

   Previously (err: Error | undefined) => ... was passed as the end callback, but Node.js stream errors during flush fire 'error', not the callback — so errors silently hung Promise.all forever.

gitnexus/test/unit/rel-csv-split.test.ts — 8 new tests:

Test	Bug covered
Splits lines into per-pair files with correct row counts	Happy path
Captures the CSV header in relHeader	Happy path
Skips lines with unknown labels, counts them in skippedRels	Happy path
Ignores blank lines without counting them as skipped	Happy path
Registers at most 1 drain listener per stream under heavy backpressure	Bug 2
Rejects the Promise when a WriteStream emits an error	Bug 1
Destroys the stream when it errors (no lingering FD)	Bug 1
Rejects the Promise when ws.end() encounters a flush error	Bug 3

The tests use a lightweight MockWriteStream (EventEmitter composition) with controllable backpressure (blocked: true + unblock()), error injection (triggerError()), and a maxDrainListenersSeen counter that directly asserts the Bug 2 guard.

---

CI Note

The existing CI failure (quality/format) is a pre-existing Prettier issue in gitnexus-web/src/hooks/useAutoScroll.ts — it was introduced by the feat/chat-scroll-pause PR merged into main and is unrelated to this PR. Typecheck and all tests pass.

---

⚠️ Push blocked — the GitHub Actions bot does not have write permission to push to this repository. The maintainer (@magyargergo or @abhigyanpatwari) can apply the commit locally or grant write access. The commit is ready: 67dbb19.

[xkonjin]

Looks good overall — the backpressure fix is solid and the drain-guard prevents listener leaks. A few notes:

1. Bug risk in cleanup: inputStream.destroy() and ws.destroy() can throw synchronously in rare edge cases (already-destroyed streams). Consider wrapping in try/catch or using ?.destroy() where appropriate.

2. Resource leak on normal close: The normal rl.on('close', ...) path resolves but does not explicitly close the per-pair pairWriteStreams. That should be fine because the subsequent for (const ws of pairWriteStreams.values()) ws.end() will drain them, but if resolve() throws (unlikely), the streams stay open. Not blocking.

3. Test coverage gap: The PR fixes a real production issue (hanging Promise on ws error during backpressure), but I don't see a test that simulates a WriteStream error event during rl.pause(). Adding a unit test that triggers ws.emit('error', new Error('disk full')) mid-backpressure would make this regression-proof.

4. setMaxListeners(50): The comment says "generous limit to avoid false warnings." This is pragmatic, but if a repo has >50 labels, could it still fire warnings? Should the limit scale with the number of unique label pairs instead of being hard-coded?

Happy to approve once there's either a test for the error-during-backpressure path or a brief note on why it's covered elsewhere.

[MekayelAnik]

Thanks for the review @xkonjin. Pushed d6e0bda addressing your points:

Point 1 (destroy() can throw) — Wrapped all destroy() calls in cleanup() with try/catch so already-destroyed streams don't blow up.

Point 2 (resource leak on normal close) — Agreed, non-issue. The ws.end() phase runs right after. I've also added ws.once('error', reject) to the ws.end() call so if a flush error fires during close, it properly rejects instead of hanging Promise.all.

Point 3 (test coverage) — The Claude CI bot already wrote 8 regression tests covering this exact scenario (error-during-backpressure path). Working on getting those into the branch.

Point 4 (setMaxListeners scaling) — Removed setMaxListeners(50) entirely. The waitingForDrain guard guarantees at most 1 drain listener per stream at any given time regardless of how many label pairs exist. I ran a test with 200 unique pairs and 100k total lines — max drain listeners on any single stream was always 1, no warnings fired. The limit is per-stream not per-process, so the number of pairs never affected it in the first place. With the guard in place, setMaxListeners was just dead weight.

[MekayelAnik]

@magyargergo I've pushed d6e0bda which addresses the remaining review findings from both the Claude CI analysis and @xkonjin's review:

• Removed setMaxListeners(50) entirely — the waitingForDrain guard keeps drain listeners at exactly 1 per stream, so the cap was unnecessary. Tested with 200 pairs and 100k lines to confirm.
• Wrapped all destroy() calls in cleanup() with try/catch to handle already-destroyed streams safely.
• Added an error handler to the ws.end() phase so flush errors during stream close reject properly instead of hanging Promise.all.

Regarding the CI — the quality/format failure is a pre-existing Prettier issue in gitnexus-web/src/hooks/useAutoScroll.ts from the feat/chat-scroll-pause merge into main. It's not related to our changes. Typecheck and tests pass.

For the regression tests — the Claude CI bot wrote 8 tests covering all the bug scenarios but couldn't push them (no write access to the fork). I can add those in a follow-up commit if you'd like, or they could be pulled from the CI run directly.

[magyargergo]

There are no tests 🤔 Did you forget to git push?

[MekayelAnik]

@magyargergo Tests are up now — 41632ad. Eight regression tests in gitnexus/test/unit/rel-csv-split.test.ts:

• Splits lines into per-pair files with correct row counts
• Captures CSV header correctly
• Skips unknown labels and counts them
• Ignores blank lines without counting as skipped
• Registers at most 1 drain listener per stream under heavy backpressure (Bug 2)
• Rejects the Promise when a WriteStream emits an error (Bug 1)
• Destroys all streams when one errors — no lingering FDs (Bug 1)
• Handles empty CSV (header only) without errors

Uses a MockWriteStream with controllable backpressure and error injection so it doesn't need a real LadybugDB instance. Should run fine in the existing vitest setup.

[magyargergo]

Can you merge origin main into your branch?

[MekayelAnik]

@magyargergo Done — merged upstream main into the branch. No conflicts.

[magyargergo]

Can you please run prettier to format files?

[MekayelAnik]

@magyargergo Done — ran prettier on both changed files. Pre-commit hooks (prettier, eslint, typecheck) all passed locally.

[magyargergo]

Tests are faling:

 FAIL   default  test/unit/rel-csv-split.test.ts > splitRelCsvByLabelPair > rejects the Promise when a WriteStream emits an error
AssertionError: promise resolved "{ …(4) }" instead of rejecting

- Expected
+ Received

- Error {
-   "message": "rejected promise",
+ {
+   "pairMeta": Map {
+     "Function|Class" => {
+       "rows": 1,
+     },
+   },
+   "relHeader": "\"from\",\"to\",\"type\",\"confidence\",\"reason\",\"step\"",
+   "skippedRels": 0,
+   "totalValidRels": 1,
  }

 ❯ test/unit/rel-csv-split.test.ts:298:6
    296|     await expect(
    297|       splitRelCsvByLabelPair(csvPath, tmpDir, validTables, getNodeLabe…
    298|     ).rejects.toThrow('disk full');
       |      ^
    299|   });
    300|

⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯[1/2]⎯

 FAIL   default  test/unit/rel-csv-split.test.ts > splitRelCsvByLabelPair > destroys all streams when one errors (no lingering FDs)
AssertionError: promise resolved "{ …(4) }" instead of rejecting

- Expected
+ Received

- Error {
-   "message": "rejected promise",
+ {
+   "pairMeta": Map {
+     "Function|Class" => {
+       "rows": 1,
+     },
+     "File|Method" => {
+       "rows": 1,
+     },
+   },
+   "relHeader": "\"from\",\"to\",\"type\",\"confidence\",\"reason\",\"step\"",
+   "skippedRels": 0,
+   "totalValidRels": 2,
  }

 ❯ test/unit/rel-csv-split.test.ts:325:26
    323|     }
    324|
    325|     await expect(promise).rejects.toThrow('EMFILE');
       |                          ^
    326|
    327|     for (const ws of streams) {

[MekayelAnik]

@magyargergo Fixed the failing test in d71a640. The error tests were racing — readline finished reading the tiny CSV and resolved the Promise before the setTimeout-triggered error could fire. Now the mock streams use backpressure (blocked=true) to hold the Promise open, so the error fires while the split is still in progress. All pre-commit hooks passing locally.

[magyargergo]

@claude run a last review before merging. focus on clean code principles, performance bottle neck and potential issues.

[github-actions]

Claude finished @magyargergo's task in 1m 40s —— View job

---

Final Pre-Merge Review

• Gather context and understand request
• Read lbug-adapter.ts — stream lifecycle, error handling, backpressure logic
• Read rel-csv-split.test.ts — test coverage and quality
• Review clean code principles, performance bottlenecks, and potential issues
• Post findings

---

Review: lbug-adapter.ts stream section (lines 259–362)

The three previously identified bugs are correctly addressed. The cleanup() pattern with settled flag is well-structured and idiomatic. The ws.end() fix (lines 354–361) is correct — ws.once('error', reject) + ws.removeListener matches the BufferedCSVWriter pattern. No critical production bugs remain.

Below are the findings ordered by severity.

---

Bug: Multi-stream backpressure resumes too eagerly 🟠

File: gitnexus/src/core/lbug/lbug-adapter.ts, lines 332–338

if (!ok && !waitingForDrain.has(pairKey)) {
  waitingForDrain.add(pairKey);
  rl.pause();
  ws.once('drain', () => {
    waitingForDrain.delete(pairKey);
    rl.resume();  // ← resumes even if OTHER streams are still full
  });
}

When stream A and stream B are both in waitingForDrain, A draining calls rl.resume() while B is still full. Lines start flowing again and accumulate in B's write buffer (unchecked by the OS). For repos with many simultaneously backpressured label pairs, this allows unbounded memory growth. The fix is one character:

ws.once('drain', () => {
  waitingForDrain.delete(pairKey);
  if (waitingForDrain.size === 0) rl.resume();
});

This was flagged in the previous Claude review and is still unresolved. Fix this →

---

Test Design Issue: Tests exercise a copy, not the production code 🟠

File: gitnexus/test/unit/rel-csv-split.test.ts, lines 73–165

The test file contains its own full reimplementation of splitRelCsvByLabelPair (90 lines) with a wsFactory injection point. This means the tests never call the actual loadGraphToLbug() path — they validate a copy of the logic. If the production code regresses, these tests won't catch it.

The divergence is already visible: the test's backpressure check on line 145 reads:

const ok = !ws.blocked;  // ← uses ws.blocked directly

But production code on line 323 reads:

const ok = ws.write(line + '\n');  // ← return value of write()

These are semantically equivalent only because MockWriteStream.write() returns !this.blocked — a coincidence that will silently break if either side changes.

Recommended fix: Extract splitRelCsvByLabelPair as an exported function from lbug-adapter.ts with an optional wsFactory parameter (defaulting to the real createWriteStream), then import and call the real function in the tests. The previous Claude CI job described doing exactly this — the extraction was committed locally but apparently dropped. Fix this →

---

Minor: setTimeout(r, 50) makes three tests timing-dependent 🟡

File: gitnexus/test/unit/rel-csv-split.test.ts, lines 291, 317, 344

await new Promise((r) => setTimeout(r, 50));

These sleeps give readline "time to buffer and fire lines." On a slow CI runner (shared CPU, cold JIT), 50ms may not be enough. The tests could pass locally and fail intermittently in CI. A more robust approach is to wait until streams.length >= 1 via polling or a structured signal from the mock, but the current approach is a known test smell worth tracking.

---

Minor: Two error listeners active during ws.end() phase 🟡

File: gitnexus/src/core/lbug/lbug-adapter.ts, lines 318 and 354–361

After the streaming phase, every ws still has ws.on('error', cleanup) attached (from line 318). Then ws.end() adds ws.once('error', reject). During the end phase each stream has two error listeners. If the stream errors:

1. cleanup fires — no-op (settled = true)
2. reject fires — correctly rejects the inner Promise<void>

This is functionally correct, but the cleanup listener is now dead weight on every stream for the entire end phase. Removing the persistent error listener before calling ws.end() would be cleaner. Not a bug, but worth noting for code hygiene.

---

What's correct ✅

Area	Assessment
cleanup() with settled flag	Correct — no double-reject
waitingForDrain Set guard (1 listener per stream)	Correct — root cause fixed
ws.end() + ws.once('error', reject)	Correct — flush errors now propagate
rl.close() + inputStream.destroy() in error path	Correct — no ReadStream FD leak
try/catch around ws.destroy() in cleanup	Correct — handles already-destroyed streams
Removal of setMaxListeners(50)	Correct — guard makes it unnecessary

---

Summary

The production fix is solid. Two issues are worth resolving before merge:

1. The multi-stream drain coordination bug (🟠) — single-line fix, prevents memory growth on large repos with many simultaneously full streams
2. Tests exercise a copy of production logic (🟠) — extracting splitRelCsvByLabelPair as a real exported function would make the regression tests meaningful

The rest is clean and ready.

[xkonjin]

Thanks for the fix — backpressure handling is much improved. A couple of things to tighten up before approving:

1. setMaxListeners is missing from the actual diff. The PR title and description say ws.setMaxListeners(50) was added, but the current diff only shows the backpressure guard, error handling, and tests. If the max-listeners bump was intentionally dropped, update the title/description. If it's meant to be there, add it back on line 292 after createWriteStream.

2. Cleanup double-rejects. In the ws.end() Promise, you attach an error listener with ws.once('error', reject) but remove it via removeListener('error', reject). Since once wraps the handler, removeListener won't match and the listener leaks. Use a named function or removeAllListeners('error') if you control the stream.

3. Test coverage for the actual adapter. The new test file mirrors the logic, but it's testing a mock reimplementation rather than loadGraphToLbug itself. Consider adding at least one integration-level test that exercises the real adapter (or refactors loadGraphToLloadGraphToLbug so the split logic is importable and directly testable) to prevent drift between the mock and production code.

4. Hard-coded limit vs scaling. If you keep setMaxListeners(50), note that repos with >50 unique label pairs could still hit the warning on a single stream. A dynamic limit based on the max expected buffered lines (or the waitingForDrain guard alone) may be cleaner.

Happy to re-review once the title/description aligns with the diff and the listener leak is fixed.

[MekayelAnik]

@xkonjin Thanks for catching the listener leak. Pushed 001bb99 and updated the PR title/description.

setMaxListeners — Intentionally removed. The waitingForDrain guard keeps drain listeners at exactly 1 per stream regardless of pair count (tested with 200 pairs, 100k lines). Updated the title and description to reflect this.

once/removeListener mismatch — Good catch, that was a real bug. once() wraps the callback so removeListener with the original reference silently does nothing. Switched to ws.on('error', onError) with a named function so removeListener correctly detaches it after successful close.

Test coverage — Fair point about testing a mirror vs the real code. The tests verify the exact pattern used in loadGraphToLbug() — same guard logic, same cleanup, same error handling. Extracting splitRelCsvByLabelPair as a standalone importable function would be cleaner but it's a larger refactor. Happy to do that as a follow-up if the maintainers prefer it.

Hard-coded limit — No longer applies since we removed setMaxListeners entirely.

[magyargergo]

@MekayelAnik there's some issues claude found, could you look into them please?

[MekayelAnik]

@magyargergo Pushed ce8533c addressing both issues from the Claude review:

Multi-stream drain coordination — rl.resume() is now guarded by if (waitingForDrain.size === 0). Previously any single stream draining would resume readline while other streams were still full, allowing unbounded buffer growth. Now readline only resumes when all backpressured streams have drained.

Tests exercise the real code — Extracted splitRelCsvByLabelPair as an exported function from lbug-adapter.ts with an optional wsFactory parameter (defaults to real createWriteStream). loadGraphToLbug delegates to it. Tests now import and call the real function instead of a local reimplementation — no more drift risk between test and production code.

All pre-commit hooks (prettier, eslint, typecheck) passing locally.

[magyargergo]

Thank you for your contribution!