PR 2841: backpressure for CSV and NDJSON export

Verdict

The PR touches 2 files, src/export/stream.js (+15 -4) and test/export/stream.test.js (+21), about 40 changed lines.

Backpressure in one minute

You asked for focus on the streaming logic, so here is the model before the diff.

A fast producer (the database cursor) can emit rows faster than a slow consumer (the HTTP response to a client on a poor connection) can send them. Node does not stop the producer on its own when you use on('data'). Unsent rows pile up in the response's memory buffer. For a 2 GB export that buffer grows until the process is killed.

The protocol has three parts.

1. res.write(chunk) returns true while the buffer has room and false once it passes its high-water mark (16 KiB by default).
2. On false, the producer must stop: call cursor.pause().
3. When the buffer empties, res emits 'drain'. The producer resumes: cursor.resume().

The same loop as a timeline of one saturation cycle.

Why the PR was needed

The benchmark in the PR description exports a 2 GB table to a client throttled to 5 Mbps. Memory before the fix grows without limit; after it stays flat.

The diff, annotated

Each hunk is followed by the findings that apply to it. Severity is marked by tag and callout level: BLOCKING is a red alert, NIT is a warning, NICE is a note.

src/export/stream.js

Before, every row was written and the return value ignored. After, the function pauses on false and resumes on 'drain'. Read the three hunks in order.

Hunk 1: the signature and the flag

@@ -1,10 +1,21 @@
 'use strict';

-function streamExport(cursor, res) {
-  cursor.on('data', (row) => {
-    res.write(JSON.stringify(row) + '\n');
-  });
+function streamExport(cursor, res, { highWaterMark = 16 * 1024 } = {}) {
+  let paused = false;
+

Hunk 2: the pause and resume

@@ -5,3 +6,10 @@ function streamExport(cursor, res, { highWaterMark = 16 * 1024 } = {}) {
+  cursor.on('data', (row) => {
+    const ok = res.write(JSON.stringify(row) + '\n');
+    if (ok || paused) return;
+    paused = true;
+    cursor.pause();
+    res.once('drain', () => {
+      paused = false;
+      cursor.resume();
+    });
+  });

Hunk 3: end and error

@@ -7,2 +17,4 @@ function streamExport(cursor, res, { highWaterMark = 16 * 1024 } = {}) {
   cursor.on('end', () => res.end());
-}
+  cursor.on('error', (err) => res.destroy(err));
+}

cursor.on('end', () => res.end());
cursor.on('error', (err) => res.destroy(err));
res.once('close', () => cursor.destroy()); // client left: release the connection

test/export/stream.test.js

Hunk 4: the slow-consumer test

@@ -0,0 +1,21 @@
+const test = require('node:test');
+const assert = require('node:assert');
+const { once } = require('node:events');
+const { Readable, Writable } = require('node:stream');
+const { streamExport } = require('../../src/export/stream');
+
+test('keeps the response buffer bounded for a slow consumer', async () => {
+  const cursor = Readable.from(Array.from({ length: 1000 }, (_, i) => ({ i })));
+  let peak = 0;
+  const res = new Writable({
+    highWaterMark: 256,
+    write(_chunk, _enc, cb) { setTimeout(cb, 1); },
+  });
+  const timer = setInterval(() => {
+    peak = Math.max(peak, res.writableLength);
+  }, 5);
+  streamExport(cursor, res);
+  await once(res, 'finish');
+  clearInterval(timer);
+  assert.ok(peak < 1024, `peak buffer ${peak} B`);
+});

write(chunk, _enc, cb) {
  peak = Math.max(peak, this.writableLength);
  setTimeout(cb, 1);
},

All findings

Before you approve

• B1 fixed and the disconnect test added
• Run the new test locally and confirm it passes
• Confirm the DB pool size is shown in the connection metrics after a cancelled export
• Check that highWaterMark has no other caller before removing it

1. 1
   Reproduce B1 with curl --limit-rate 1k and Ctrl-Cnow
   Reviewer
2. 2
   Author applies the close handler and test2026-10-20
   PR author
3. 3
   Re-review the delta only2026-10-21
   Reviewer

Back to the examples