stream: pipeline should error if any stream is destroyed

pipeline should immediately fail with ERR_STREAM_DESTROYED when any of the streams have already been destroyed.

Readable might need a little extra consideration since it's possible to read the data after being destroyed. Should maybe check _readableState.errored and/or _readableState.ended.

Refs: #29227 (comment)

I would like to work on this problem

streams: pipeline error if any stream is destroyed

I also bumped into this. Http client can easily cause this to http server which is using pipeline (this could cause really bad things):

let { PassThrough, pipeline } = require("stream");
let http = require("http");

let server = http.createServer(async function(req, res)
{
	await new Promise(r => setTimeout(r, 1000));
	console.log("request destroyed", req.destroyed);
	let pass = new PassThrough();
	pipeline(req, pass, e => console.log("pipeline finished", e));
	for await (let chunk of pass) console.log("received", chunk.length);
	console.log("body processed");
	res.end();
});

(async function()
{
	await new Promise(resolve => server.listen(resolve));
	let req = http.request({ port: server.address().port, method: "post" });
	req.on("error", () => null);
	req.write(Buffer.alloc(10000));
	setTimeout(() => req.destroy(), 500);
}());

request destroyed true
received 10000

@ronag

Can I continue to implement this? @ronag
I see this pr hasn't been updated for a long time: #36791

@ronag

I find that if one stream has been destroyed, the pipeline will call the callback with an ERR_STREAM_PREMATURE_CLOSE now.

I used this test case:

const {
  pipeline,
  PassThrough,
} = require('stream');

{
  const r = new PassThrough();
  const d = new PassThrough();

  d.on('data', (data) => {
    console.log(data);
  });

  r.write('aaa');
  r.destroy();

  // make sure r is destroyed
  process.nextTick(function () {
    pipeline([r, d], (err) => {
      console.log(err);
    });
  });
}

On node version 16.13.2, it will output:

<Buffer 61 61 61>

On the master branch, it will output:

Error [ERR_STREAM_PREMATURE_CLOSE]: Premature close
    at new NodeError (node:internal/errors:372:5)
    at onclose (node:internal/streams/end-of-stream:139:30)
    at processTicksAndRejections (node:internal/process/task_queues:77:11) {
  code: 'ERR_STREAM_PREMATURE_CLOSE'
}

It happened in the pipe function, which calls the end-of-stream: https://github.com/nodejs/node/blob/master/lib/internal/streams/end-of-stream.js#L182

let closed = isClosed(stream);

// ...

if (closed) {
  process.nextTick(onclose);

// in onclose
if (writable && !writableFinished) {
  if (!isWritableFinished(stream, false))
    return callback.call(stream,
                         new ERR_STREAM_PREMATURE_CLOSE());
}

Do I understand this issue correctly? Do we still need to throw a Error directly? as in https://github.com/nodejs/node/pull/36791/files#r665904922

@ronag

@meixg

@meixg Does this work for you with the promisified version of stream.pipeline too?
We've seen behavior where this end up with an unresolvable promise which broke our async-try-await-catch-finaly code because of this tc39/proposal-async-await#89

Not sure if this is the same isue, but interestingly I see a difference between how this behaves inside an async function (which seems to fail fast as expected) vs. how this runs in the REPL calling await.

This one works -

$ node -v
v16.13.2
$ node -e '(async () => {
  const pp = util.promisify(stream.pipeline);
  const r = new stream.PassThrough();
  const w = new stream.PassThrough();
  w.on("data", console.log);
  r.write("la la la");
  r.destroy();
  await pp([r,w]);
})().then(
  () => console.log("DONE"),
  err => console.error("FAILED",err)
)'

FAILED Error [ERR_STREAM_PREMATURE_CLOSE]: Premature close
    at new NodeError (node:internal/errors:371:5)
    at PassThrough.onclose (node:internal/streams/end-of-stream:135:30)
    at PassThrough.emit (node:events:402:35)
    at emitCloseNT (node:internal/streams/destroy:138:10)
    at processTicksAndRejections (node:internal/process/task_queues:82:21) {
  code: 'ERR_STREAM_PREMATURE_CLOSE'
}

But in the REPL it gets stuck like this ...

$ node
Welcome to Node.js v16.13.2.
Type ".help" for more information.
> pp = util.promisify(stream.pipeline);
[Function: pipeline]
> r = new stream.PassThrough();
PassThrough { ... }
> w = new stream.PassThrough();
PassThrough { ... }
> w.on("data", console.log);
PassThrough { ... }
> r.write("la la la");
true
> r.destroy();
PassThrough { ... }
> await pp([r,w]);
<Buffer 6c 61 20 6c 61 20 6c 61>

@@@ this is stuck forever here until I ctrl-c @@@

Uncaught:
Error [ERR_SCRIPT_EXECUTION_INTERRUPTED]: Script execution was interrupted by `SIGINT`
    at __node_internal_captureLargerStackTrace (node:internal/errors:464:5)
    at new NodeError (node:internal/errors:371:5)
    at sigintListener (node:repl:598:27)
    at REPLServer.onSigInt (node:repl:813:9)
    at REPLServer.emit (node:events:390:28)
    at REPLServer.emit (node:domain:475:12)
    at REPLServer.Interface._ttyWrite (node:readline:1081:16)
    at REPLServer.self._ttyWrite (node:repl:984:9)
    at ReadStream.onkeypress (node:readline:288:10)
    at ReadStream.emit (node:events:390:28)
    at ReadStream.emit (node:domain:475:12)
    at emitKeys (node:internal/readline/utils:358:14)
    at emitKeys.next (<anonymous>)
    at ReadStream.onData (node:internal/readline/emitKeypressEvents:61:36)
    at ReadStream.emit (node:events:390:28)
    at ReadStream.emit (node:domain:475:12)
    at addChunk (node:internal/streams/readable:315:12)
    at readableAddChunk (node:internal/streams/readable:289:9)
    at ReadStream.Readable.push (node:internal/streams/readable:228:10)
    at TTY.onStreamRead (node:internal/stream_base_commons:199:23)
    at TTY.callbackTrampoline (node:internal/async_hooks:130:17) {
  code: 'ERR_SCRIPT_EXECUTION_INTERRUPTED'
}

@ronag

I find that if one stream has been destroyed, the pipeline will call the callback with an ERR_STREAM_PREMATURE_CLOSE now.

I used this test case:

const {
  pipeline,
  PassThrough,
} = require('stream');

{
  const r = new PassThrough();
  const d = new PassThrough();

  d.on('data', (data) => {
    console.log(data);
  });

  r.write('aaa');
  r.destroy();

  // make sure r is destroyed
  process.nextTick(function () {
    pipeline([r, d], (err) => {
      console.log(err);
    });
  });
}

On node version 16.13.2, it will output:

<Buffer 61 61 61>

On the master branch, it will output:

Error [ERR_STREAM_PREMATURE_CLOSE]: Premature close
    at new NodeError (node:internal/errors:372:5)
    at onclose (node:internal/streams/end-of-stream:139:30)
    at processTicksAndRejections (node:internal/process/task_queues:77:11) {
  code: 'ERR_STREAM_PREMATURE_CLOSE'
}

It happened in the pipe function, which calls the end-of-stream: https://github.com/nodejs/node/blob/master/lib/internal/streams/end-of-stream.js#L182

let closed = isClosed(stream);

// ...

if (closed) {
  process.nextTick(onclose);

// in onclose
if (writable && !writableFinished) {
  if (!isWritableFinished(stream, false))
    return callback.call(stream,
                         new ERR_STREAM_PREMATURE_CLOSE());
}

Do I understand this issue correctly? Do we still need to throw a Error directly? as in https://github.com/nodejs/node/pull/36791/files#r665904922

@ronag

This looks correct to me. You are destroying the writable before it has ended, hence it will get a premature close.

@guymguym

@guymguym Seems unrelated. Could you maybe open a separate issue?

This looks correct to me. You are destroying the writable before it has ended, hence it will get a premature close.

Does the test case match the issue here?
If it matches, maybe this issue is resolved?

Or if it doesn't match, what the test case should look like? maybe I can work on that.

I don't think this is a problem anymore.

I'm on Node 22 with code that - intermittently - never throws or returns from an await pipeline(). I also am calling readstream.destroy() during the operation, as in the example at #issuecomment-1042881798. It takes something like 500 pipeline operations, reading the initial part of a file into a buffer using Writable, before this happens. I find that a workaround is to add { end: false } to the pipeline options and manually close the destination stream.

Navigation Menu

Repository navigation

stream: pipeline should error if any stream is destroyed #36674

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

stream: pipeline should error if any stream is destroyed #36674

Description

Activity

kalenikalexander commented on Dec 29, 2020

misos1 commented on May 19, 2021

meixg commented on Feb 15, 2022

meixg commented on Feb 17, 2022

guymguym commented on Feb 17, 2022

guymguym commented on Feb 17, 2022

ronag commented on Feb 17, 2022

ronag commented on Feb 17, 2022

meixg commented on Feb 17, 2022

ronag commented on Feb 17, 2022

fozcode commented on Jul 3, 2025

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions