node:stream: stop 'data' and 'end' after destroy() on a native-backed Readable

Merged
RO
1mo ago
+215-12

Related to #36169

Problem

  • child.stdout.destroy() inside a 'data' listener is followed by one more 'data' event (destroyed === true), and by 'end' if EOF was due. Node emits only 'close'. Same for child.stderr and Readable.fromWeb() over a native stream.
  • exec() and execFile() stop a stream with destroy() at maxBuffer, as Node does. The extra chunk gives the callback up to one chunk too many.
  • Cause: _read() in src/js/internal/streams/native-readable.ts pushes synchronously, so a flowing Readable holds the next chunk while a 'data' listener runs. flow() emits it after destroy(). EOF follows a tick later.

Fix

  • destroy() drops what a flowing stream has buffered, and EOF is not pushed to a destroyed stream. The events now match Node: 'data', then 'close'.
  • A paused or ended stream keeps its buffer, as Node's does. A destroy from inside _read() is the source failing, and the bytes read before the error still reach 'data'.
  • Verified: test/js/node/stream/node-stream.test.js (six cases fail on main), test/js/node/child_process/child_process.test.ts (two fail on main). Node v26.3.0 gives every expected sequence.

Background

  • NativeReadable is the Readable over a native ReadableStream source. child_process uses it for stdout and stderr, Readable.fromWeb() for a native web stream.
  • In flowing mode, Readable.read() calls _read() before it removes the chunk it is about to emit. A synchronous push() lands behind that chunk, so the source stays one chunk ahead.
  • Node's child stdio (net.Socket) and fromWeb() push asynchronously, so their buffer is empty while a 'data' listener runs.
Notes

Repro from the report. Node v26.3.0 and Bun 1.3.14 print 0. Bun 1.4.0 to main print 1.

const { spawn } = require("child_process");
const c = spawn("python3", ["-c", "import os\nb=b'x'*65536\nfor i in range(48): os.write(1, b)"]);
let at = -1, n = 0, after = 0;
c.stdout.on("data", d => {
  n += d.length;
  if (at >= 0) return void after++;
  if (n > 1024 * 1024) { at = n; c.stdout.destroy(); }
});
c.on("close", () => console.log("'data' events after destroy():", after));
  • The extra chunk does not come from the pending pull() promise: push() already returns false once the stream is destroyed. An instrumented run shows the chunk in _readableState.buffer when destroy() runs, and flow() emits it from emitReadable_.
  • Deterministic form: Readable.fromWeb(new Blob([Buffer.alloc(1 << 20)]).stream()) with destroy() in the first 'data' listener. main: data, data (destroyed=true), end, close. Node and this branch: data, close. A 100 byte body on main: data, end, close.
  • execFile with maxBuffer 1 MiB against a 3 MiB writer, debug build: 11 of 32 callbacks over the cap on main (1,077,568 to 1,114,112 bytes), 0 of 32 on this branch. Release main with the python writer above: 7 of 16. Node: 0 of 16. #36169 latches the overflow in the exec handler. With this change the handler needs no latch, as in Node.
  • Read errors: pull() throws a stored read error from inside _read(), and the stream is destroyed with it while a chunk is still buffered. #41456 delivers the bytes and then the error. destroy() keeps the buffer in that case (state.sync). The new case in test/js/bun/spawn/spawn-stdio-syscall-error.test.ts counts the bytes the parent received before an injected EIO and expects all of them in 'data'. It passes on main, and it fails (about 180 KiB lost) when the state.sync check is removed.
  • The read path does not change. To push each pull result on a later tick also fixes this, but it costs a tick for each chunk on the hot path, and the EOF check is still needed.
  • Readable itself is unchanged. A pure JS Readable that pushes synchronously emits one more 'data' after destroy() in Node too, and Node keeps buffered data across destroy().
  • One deliberate difference from Node: chunks buffered while paused (source not ended), then resume(), then destroy() in the first 'data' listener. Node and main emit the rest of the buffered chunks after destroy(). This branch drops them. To keep them needs a record of which buffered chunks came from a pause, on the hot path.
  • Not changed here: Duplex.from(nativeWebStream) emits one 'data' after destroy() on main and on this branch (its own _read loop finds several chunks ready). When the source ended with two or more chunks buffered, Bun emits 'end' before 'close' after destroy() and Node does not.
  • No event after destroy() on main for: http client response, http server request, fs.createReadStream, zlib gunzip, net.Socket, child.stdio[3].
  • On Windows the child stdio cases pass with and without the fix (reads there complete asynchronously). The fromWeb cases fail without it.
  • Suites run on the debug build: test/js/node/child_process/, test/js/node/stream/, readline.node.test.ts, node-fetch.test.js, undici.test.ts, test/js/web/streams/streams.test.js, fetch-backpressure.test.ts, body.test.ts, process-stdin.test.ts, spawn.test.ts (stdio subset), spawn-stdio-syscall-error.test.ts, spawn-streaming-stdout.test.ts, test/regression/issue/1632.test.ts, 09555.test.ts, and about 55 test-child-process-*, test-stdio-*, test-stream-* and fromWeb files from test/js/node/test/parallel/.