stream: clean up when pipeline throws synchronously

[shani-singh1]

pipelineImpl() wires the streams together in a loop that can throw synchronously. The most common case is ERR_STREAM_UNABLE_TO_PIPE, raised when the next stream is already closed or destroyed, which happens routinely when a destination goes away first (for example pipeline(fs.createReadStream(file), res) after the HTTP client disconnected).

Each stream the loop adopts registers a destroy function in destroys. finishImpl() is the only code that drains destroys, disposes the listener added to the caller's AbortSignal and calls ac.abort(), and it never runs when the loop throws. Every stream already wired up is therefore left undestroyed and its resources leak. For an fs.ReadStream source that is a leaked file descriptor.

The loop has six synchronous throw sites: one ERR_STREAM_UNABLE_TO_PIPE, three ERR_INVALID_RETURN_VALUE and two ERR_INVALID_ARG_TYPE. This wraps the loop so the same teardown runs before the error propagates. The error is still thrown, so the observable failure mode is unchanged.

Most of the diff is the re-indentation of the existing loop. Reviewing with git diff -w shows the actual change is 13 lines.

Before

callback form throws : ERR_STREAM_UNABLE_TO_PIPE
  sources undestroyed : 50 / 50  (expected 0)
  fds still open      : 50 / 50  (expected 0)
promise form rejects : ERR_STREAM_UNABLE_TO_PIPE
  sources undestroyed : 50 / 50  (expected 0)
  fds still open      : 50 / 50  (expected 0)
  abort listeners     : 50 / 50  (expected 0)
control (ENOENT)     : rs.destroyed = true | mid.destroyed = true  (both expected true)

After

callback form throws : ERR_STREAM_UNABLE_TO_PIPE
  sources undestroyed : 0 / 50  (expected 0)
  fds still open      : 0 / 50  (expected 0)
promise form rejects : ERR_STREAM_UNABLE_TO_PIPE
  sources undestroyed : 0 / 50  (expected 0)
  fds still open      : 0 / 50  (expected 0)
  abort listeners     : 0 / 50  (expected 0)
control (ENOENT)     : rs.destroyed = true | mid.destroyed = true  (both expected true)

The full reproduction is in the linked issue. The added test fails on main (5 failing assertions) and passes with this change.

I also checked the change against a set of ordinary pipeline() usages (happy path, fs read to writable, asynchronous mid-stream error, ENOENT source, async generator transform, abort via an outer signal) and the behaviour is identical before and after. The only observable differences are on the two synchronous-throw paths, where the streams are now destroyed.

Fixes: #65063

[nodejs-github-bot]

Review requested:

• @nodejs/streams