From d71a6408c77367fc57411278805dc76cd6a642af Mon Sep 17 00:00:00 2001
From: MekayelAnik <mekayel.anik@gmail.com>
Date: Tue, 14 Apr 2026 16:03:24 +0600
Subject: [PATCH] fix(test): use backpressure to keep promise pending during
 error tests
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The error tests were racing — readline finished reading the tiny CSV
and resolved the Promise before setTimeout fired the error. Now the
mock streams use blocked=true to trigger backpressure, keeping the
Promise pending so the error fires while the split is still in progress.
---
 gitnexus/test/unit/rel-csv-split.test.ts | 54 ++++++++++++++----------
 1 file changed, 31 insertions(+), 23 deletions(-)

diff --git a/gitnexus/test/unit/rel-csv-split.test.ts b/gitnexus/test/unit/rel-csv-split.test.ts
index 95c6ec190..4e8eb0654 100644
--- a/gitnexus/test/unit/rel-csv-split.test.ts
+++ b/gitnexus/test/unit/rel-csv-split.test.ts
@@ -303,39 +303,47 @@ describe('splitRelCsvByLabelPair', () => {
   it('rejects the Promise when a WriteStream emits an error', async () => {
     const csvPath = writeCsv([HEADER, csvLine('Function:a', 'Class:b')]);
 
-    const factory = () => {
-      const ws = new MockWriteStream();
-      // Trigger error after first write
-      setTimeout(() => ws.triggerError(new Error('disk full')), 10);
-      return ws;
-    };
-
-    await expect(
-      splitRelCsvByLabelPair(csvPath, tmpDir, validTables, getNodeLabel, factory),
-    ).rejects.toThrow('disk full');
-  });
-
-  it('destroys all streams when one errors (no lingering FDs)', async () => {
-    const csvPath = writeCsv([
-      HEADER,
-      csvLine('Function:a', 'Class:b'),
-      csvLine('File:c', 'Method:d'),
-    ]);
-
     const streams: MockWriteStream[] = [];
     const factory = () => {
       const ws = new MockWriteStream();
+      ws.blocked = true; // hold the promise open via backpressure
       streams.push(ws);
       return ws;
     };
 
-    // Error the first stream after both are created
     const promise = splitRelCsvByLabelPair(csvPath, tmpDir, validTables, getNodeLabel, factory);
 
-    await new Promise((r) => setTimeout(r, 20));
-    if (streams.length > 0) {
-      streams[0].triggerError(new Error('EMFILE'));
+    // Wait for readline to process, then error while paused on drain
+    await new Promise((r) => setTimeout(r, 50));
+    expect(streams.length).toBeGreaterThan(0);
+    streams[0].triggerError(new Error('disk full'));
+
+    await expect(promise).rejects.toThrow('disk full');
+  });
+
+  it('destroys all streams when one errors (no lingering FDs)', async () => {
+    // Use enough lines to create two different pair streams
+    const lines = [HEADER];
+    for (let i = 0; i < 10; i++) {
+      lines.push(csvLine(`Function:f${i}`, `Class:c${i}`));
+      lines.push(csvLine(`File:e${i}`, `Method:m${i}`));
     }
+    const csvPath = writeCsv(lines);
+
+    const streams: MockWriteStream[] = [];
+    const factory = () => {
+      const ws = new MockWriteStream();
+      ws.blocked = true; // hold all streams open via backpressure
+      streams.push(ws);
+      return ws;
+    };
+
+    const promise = splitRelCsvByLabelPair(csvPath, tmpDir, validTables, getNodeLabel, factory);
+
+    // Wait for readline to process and create streams
+    await new Promise((r) => setTimeout(r, 50));
+    expect(streams.length).toBeGreaterThanOrEqual(2);
+    streams[0].triggerError(new Error('EMFILE'));
 
     await expect(promise).rejects.toThrow('EMFILE');