René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"`stream.pipeline()` leaks file descriptors when it throws synchronously (ERR_STREAM_UNABLE_TO_PIPE)","articleBody":"### Version\n\nv24.11.1, and `main` (`735a09f999a`)\n\n### Platform\n\nReproduced on Windows 11 x64; the code path is platform independent.\n\n### Subsystem\n\nstream\n\n### What steps will reproduce the bug?\n\n`stream.pipeline()` wires the streams together in a loop, and that loop can throw synchronously. The most common case is `ERR_STREAM_UNABLE_TO_PIPE`, raised when the destination is already closed or destroyed. When it throws, every stream `pipeline()` had already taken ownership of is left undestroyed, so its resources leak. For a `fs.ReadStream` source that means a leaked file descriptor.\n\nThe trigger is an ordinary production condition: piping to a destination that has already gone away, e.g. `pipeline(fs.createReadStream(file), res)` after the HTTP client disconnected.\n\n```js\nimport { pipeline, PassThrough, Writable } from 'node:stream';\nimport { pipeline as pipelinePromise } from 'node:stream/promises';\nimport { once, getEventListeners } from 'node:events';\nimport fs from 'node:fs';\nimport os from 'node:os';\nimport path from 'node:path';\n\nconst tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'pipeline-leak-'));\nconst file = path.join(tmp, 'data.bin');\nfs.writeFileSync(file, Buffer.alloc(4096, 'x'));\n\nconst openStream = async () => {\n  const rs = fs.createReadStream(file);\n  await once(rs, 'open');          // ensure the fd is really allocated\n  return rs;\n};\nconst deadWritable = () => {\n  const w = new Writable({ write(c, e, cb) { cb(); } });\n  w.destroy();                     // destination already gone\n  return w;\n};\n\n// 1. callback form\n{\n  const sources = [];\n  for (let i = 0; i \u003c 50; i++) {\n    const rs = await openStream();\n    sources.push(rs);\n    try {\n      pipeline(rs, new PassThrough(), deadWritable(), () => {});\n    } catch (err) {\n      if (i === 0) console.log('callback form throws :', err.code);\n    }\n  }\n  await new Promise((r) => setTimeout(r, 150));\n  console.log('  sources undestroyed :', sources.filter((s) => !s.destroyed).length, '/ 50  (expected 0)');\n  console.log('  fds still open      :', sources.filter((s) => s.fd != null).length, '/ 50  (expected 0)');\n  for (const s of sources) s.destroy();\n}\n\n// 2. promise form, plus the caller's AbortSignal\n{\n  const ac = new AbortController();  // long-lived, e.g. a server shutdown signal\n  const sources = [];\n  for (let i = 0; i \u003c 50; i++) {\n    const rs = await openStream();\n    sources.push(rs);\n    try {\n      await pipelinePromise(rs, new PassThrough(), deadWritable(), { signal: ac.signal });\n    } catch (err) {\n      if (i === 0) console.log('promise form rejects :', err.code);\n    }\n  }\n  await new Promise((r) => setTimeout(r, 150));\n  console.log('  sources undestroyed :', sources.filter((s) => !s.destroyed).length, '/ 50  (expected 0)');\n  console.log('  fds still open      :', sources.filter((s) => s.fd != null).length, '/ 50  (expected 0)');\n  console.log('  abort listeners     :', getEventListeners(ac.signal, 'abort').length, '/ 50  (expected 0)');\n  for (const s of sources) s.destroy();\n}\n\n// 3. control: an ordinary asynchronous failure does clean up correctly\n{\n  const rs = fs.createReadStream(path.join(tmp, 'nope'));\n  const mid = new PassThrough();\n  await new Promise((resolve) => pipeline(rs, mid, new PassThrough(), () => resolve()));\n  console.log('control (ENOENT)     : rs.destroyed =', rs.destroyed, '| mid.destroyed =', mid.destroyed, ' (both expected true)');\n}\n\nfs.rmSync(tmp, { recursive: true, force: true });\n```\n\n### How often does it reproduce? Is there a required condition?\n\nEvery time. The only condition is that `pipeline()` throws while wiring the streams up, after at least one stream has already been wired.\n\n### What is the expected behavior? Why is that the expected behavior?\n\nAll the streams should be destroyed and the file descriptors released, and the listener added to the caller's `AbortSignal` should be removed.\n\n`doc/api/stream.md` states:\n\n> `stream.pipeline()` closes all the streams when an error is raised.\n\nand\n\n> `stream.pipeline()` will call `stream.destroy(err)` on all streams except: `Readable` streams which have emitted `'end'` or `'close'`; `Writable` streams which have emitted `'finish'` or `'close'`.\n\nThe sources here emitted none of those events. The control case in the reproduction shows that an *asynchronous* pipeline error does destroy everything, so the two paths disagree.\n\n### What do you see instead?\n\n```\ncallback form throws : ERR_STREAM_UNABLE_TO_PIPE\n  sources undestroyed : 50 / 50  (expected 0)\n  fds still open      : 50 / 50  (expected 0)\npromise form rejects : ERR_STREAM_UNABLE_TO_PIPE\n  sources undestroyed : 50 / 50  (expected 0)\n  fds still open      : 50 / 50  (expected 0)\n  abort listeners     : 50 / 50  (expected 0)\ncontrol (ENOENT)     : rs.destroyed = true | mid.destroyed = true  (both expected true)\n```\n\n### Additional information\n\nThe wiring loop in `pipelineImpl()` ([`lib/internal/streams/pipeline.js`](https://github.com/nodejs/node/blob/main/lib/internal/streams/pipeline.js)) pushes a destroy function into `destroys` for each stream it adopts, and `finishImpl()` is the only code that drains `destroys`, disposes the `AbortSignal` listener and calls `ac.abort()`.\n\nThe loop is not wrapped in `try`/`finally`, and it can throw at six places:\n\n* `ERR_STREAM_UNABLE_TO_PIPE` when the next stream is already closed or destroyed\n* `ERR_INVALID_RETURN_VALUE` (x3) when a transform function returns something that is not iterable\n* `ERR_INVALID_ARG_TYPE` (x2) when a value cannot be piped into the next stream\n\nWhen any of these fire, `finishImpl()` never runs, so nothing in `destroys` is ever called.\n\nThe `ERR_INVALID_RETURN_VALUE` path leaks in exactly the same way:\n\n```js\nconst source = fs.createReadStream(file);\npipeline(source, () => 42, new PassThrough(), () => {});  // throws\n// source is left open\n```\n\nI have a fix and will open a PR shortly.\n","author":{"url":"https://github.com/shani-singh1","@type":"Person","name":"shani-singh1"},"datePublished":"2026-08-06T01:48:01Z","url":"https://github.com/nodejs/node/issues/65063"}