stream.pipeline destroys writable stream when error is occurred

Version:
v11.10.0
Platform:
Mac OS Darwin 16.7.0
Subsystem:
stream, http

stream.pipeline is helpful to handle error and interoperable to Promise.
However, I found a behavior that is not suitable my usecase.

I am creating a web server with stream.pipeline.
If my readable stream emits an error like "file not found", I would like to send error response to my clients.

code example is as follows.

const fs = require('fs')
const http = require('http')
const { pipeline } = require('stream')

const server = http.createServer((req, res) => {
  const r = fs.createReadStream('./foo2.txt')
  pipeline(r, res, (err) => {
    if (err) {
      console.log(err) // No such file
      return res.end("error!!!"); // write error response but this message is not shown.
    }
  })
})

server.listen(9000)

I have investigated nodejs core, stream.pipeline destroys writable stream when error is occurred.
https://github.com/nodejs/node/blob/master/lib/internal/streams/pipeline.js#L42

so the above code cannot send error response.

Question

Is this an expected behaviour?
In my understandings, writable stream should be closed anyway, but in this case, we would like to close writable stream manually.

In this situation, we need to create custom writable stream?

I could send a PR to pass an option like { destroyOnError: false } to avoid destory automatically.

The whole reason of pipeline is to call destroy(). If you don’t want this behavior you can use plain old source.pipe().

hm. I would like to use Promise and Stream seamlessly, stream.pipeline is seems to be helpful for my situation in API doc.

const pipeline = util.promisify(stream.pipeline);

async function run() {
  await pipeline(
    fs.createReadStream('archive.tar'),
    zlib.createGzip(),
    fs.createWriteStream('archive.tar.gz')
  );
  console.log('Pipeline succeeded.');
}

Anyway, I understood stream.pipeline should call destroy() automatically.

destroy function does not catch error object

I have tried to create custom Writable Stream to destroy manually.

const fs = require('fs')
const http = require('http')
const { pipeline, Writable } = require('stream')

class ResponseWriter extends Writable {
  constructor(options) {
    super(options);
    this.res = options.res;
  }

  _write(chunk, encoding, callback) {
    this.res.write(chunk);
    this.data += chunk;
    callback();
  }

  _final(callback) {
    this.data += this.res.end();
    callback();
  }
  
  _destroy(err, callback) {
    console.log("destroy", err); // does not show error.
    this.res.end("ERROR!! " + err);
    this.res.destroy(err);
    callback(err);
  }
}

const server = http.createServer((req, res) => {
  const r = fs.createReadStream('./foo2.txt')
  const w = new ResponseWriter({
    res,
  });
  pipeline(r, w, (err) => {
    console.log(err) // No such file
  });
})

server.listen(9000)

But this solution does not work well. cause _destroy cannot catch the err object.

in node.js core, stream.pipeline does not pass err object to the destroy function.

if (typeof stream.destroy === 'function') return stream.destroy(); // destroy(err) ??

Is this expected behavior? If not, I will send PR.

This is expected behavior. .destroy(err) is supposed to be called with the error that caused the current stream to fail. If you want to always handle the "outer" logic, you should do that in the pipeline callback (or returned promise). Note that you don't want to call stream.destroy(err) because it will cause all streams to emit 'error' events, which can cause nasty side effects because of missing 'error' handlers. I don't think we can support that.

Note that from an http standpoint, adding this.res.end("ERROR!! " + err); after a byte is sent is not correct, as you'll be appending to data that is already flowing.

I think the source of these problems is not pipeline or pipe, but rather that the implementation of HttpResponse is not matching what you'd expect. Essentially if you are piping a stream into a response, you want to handle error conditions on a case-by-cases basis, making pipeline useless for you.

Essentially if you are piping a stream into a response, you want to handle error conditions on a case-by-cases basis, making pipeline useless for you.

I agree with that, but my opinion is different.

pipeline would be better to pass an option (pipeline(...streams, [option], callback)) to handle case-by-case basis like Stream.finished that has an option to handle case-by-case basis (finished(stream[, options], callback)).

https://nodejs.org/api/stream.html#stream_stream_finished_stream_options_callback

I know HttpResponse usecase is a little bit different from other stream usecases (http.request, fs.read/write), but other TCP/UDP network server using stream like ftp would return error response to their clients, this usecase is not so rare case.

I'm willing to discuss adding new features/options to pipeline as long as the default behavior stays consistent. Send a PR!

@yosuke-furukawa

@yosuke-furukawa
I believe you can solve your issue with the PassThrough stream. Look for the code below:

app.post('/resize/:height/:width', (req, res) => {
  const { height, width } = req.params
    const resizer = sharp().resize({
      height: Number(height),
      width: Number(height)
    }).jpeg()
  const psToReq = new stream.PassThrough() // <-----------
  promisify(stream.pipeline)(
        req,
        resizer,
        psToReq // <-----------
  ).catch((e) => {
        console.log(e)
        res.sendStatus(400)
    })
  psToReq.pipe(res) // <-----------
})

@mcollina what do you think? does it make any sense to add this trick(with PassThrough) into the documentation? This problem appears quite often

I don't understand how the above trick helps.

@yosuke-furukawa

a changed code sample from @yosuke-furukawa

const fs = require('fs')
const http = require('http')
const stream = require('stream')

const server = http.createServer((req, res) => {
  const r = fs.createReadStream('./foo2.txt')
  const ps = new stream.PassThrough()
  stream.pipeline(
   r,
   ps,
   (err) => {
    if (err) {
      console.log(err) // No such file
      return res.end("error!!!"); // write error response but this message is not shown.
    }
  })
  ps.pipe(res)
})

server.listen(9000)

and we run the server and call:

>curl localhost:9000
error!!!%

the server output:

node index.js
[Error: ENOENT: no such file or directory, open './foo2.txt'] {
  errno: -2,
  code: 'ENOENT',
  syscall: 'open',
  path: './foo2.txt'
}

so, I achieve the expected behavior in a simple way without changing the interface. Do you see my point now?

I still don't understand why adding a passthrough in the middle is going to fix this problem.
I think this is documented behavior to some extent: pipeline destroys all the streams when there is an error. Note that sending res.end() could create issues, because in the generic case we could be in the middle of a transfer - so we should not really send anything.

The correct fix for this issues is to wait that the file is correctly opened, and then use pipeline. If the file cannot be opened correctly, we should send that message to the user. That's what https://www.npmjs.com/package/send does.

@yosuke-furukawa

I think this is documented behavior to some extent: pipeline destroys all the streams when there is an error.

yep. but in a case of HttpResponse you don't want to destroy it.

Note that sending res.end()

I just changed the code from @yosuke-furukawa

The correct fix for this issues

it's not about this particular case. This case is just simple enough to show the concept. There is a lot of cases when you need to pipe some read stream to a transformation stream and to response(e.g., check my image resizing sample). And you need a way to handle possible ALL errors which can appear.

it was probably the reason why the original pump from @mafintosh was returning the last stream in the end.

I am in favour of adding options to pipeline and will send PR if I can.

We'll be happy to evaluate that change, send it over.

What about:

const r = fs.createReadStream('./foo2.txt')
await finished(pipeline(r, transform, (err) => {})
  .on('error', err => {
    console.log(err) // No such file
    res.end("error!!!"); // write error response but this message is not shown. 
  }) 
  .pipe(res)
)

Due to #34133 resolution, this issue is still a thing?

I’m sorry but it does not seem that #34133 is resolved. Why do you think so?

There are some disagreements in this feature and no consensus, looks like it was the reason that #34133 was closed. Otherwise, I can re-create that PR with some adjustments.

@ronag

It seems that PR was closed due to:

Due to lack of sufficient impetus, I will close this PR.

Having a fresh PR would help, I think we can get this landed. Wdyt @ronag?

I agree.

Any good ideas, maybe I can reopen this PR.

There has been no activity on this feature request for 5 months and it is unlikely to be implemented. It will be closed 6 months after the last non-automated comment.

For more information on how the project manages feature requests, please consult the feature request management document.

Closed by #41796

For others like me looking for a workaround for this:

const makeIndestructible = (stream: NodeJS.WritableStream) => {
  return new Writable({ write: stream.write.bind(stream) })
}

const source = createReadStream(filePath)
await pipeline(source, makeIndestructible(sink))

Although the wrapping Writable is destroyed, the underlying sink stream remains open.

fix(send): nodejs/node#26311

Navigation Menu

Repository navigation

stream.pipeline destroys writable stream when error is occurred #26311

Question

destroy function does not catch error object

5 remaining items

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

stream.pipeline destroys writable stream when error is occurred #26311

Description

Question

Activity

mcollina commented on Feb 28, 2019

yosuke-furukawa commented on Mar 1, 2019

destroy function does not catch error object

mcollina commented on Mar 2, 2019

yosuke-furukawa commented on Mar 4, 2019

mcollina commented on Mar 4, 2019

kharandziuk commented on Jul 3, 2019

mcollina commented on Jul 3, 2019

kharandziuk commented on Jul 3, 2019

mcollina commented on Jul 4, 2019

kharandziuk commented on Jul 4, 2019

rickyes commented on Jun 28, 2020

mcollina commented on Jun 28, 2020

ronag commented on Jul 1, 2020

5 remaining items

mcollina commented on Dec 4, 2021

RafaelGSS commented on Dec 4, 2021

mcollina commented on Dec 4, 2021

ronag commented on Dec 4, 2021

rickyes commented on Dec 5, 2021

github-actions commented on Jun 3, 2022

RafaelGSS commented on Jun 3, 2022

Lordfirespeed commented on Aug 10, 2024

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions