stream.pipeline does not invoke callback when error happens

[sullenor]

Version

v15.8.0

Platform

Darwin Alekseys-iMac.local 20.5.0 Darwin Kernel Version 20.5.0: Sat May 8 05:10:33 PDT 2021; root:xnu-7195.121.3~9/RELEASE_X86_64 x86_64

Subsystem

stream

What steps will reproduce the bug?

Run the following code:

'use strict'

const { Transform, pipeline } = require('stream')

function createTransformStream (tf, context) {
  return new Transform({
    readableObjectMode: true,
    writableObjectMode: true,

    transform (chunk, encoding, done) {
      tf(chunk, context, done)
    }
  })
}

const ts = createTransformStream((chunk, _, done) => done(new Error('Artificial error')))

pipeline(ts, process.stdout, (err) => {
  if (err) console.log(err)
  console.log('done')
})

console.log('run test')
ts.write('test')

How often does it reproduce? Is there a required condition?

all the time

What is the expected behavior?

I expect pipeline callback to be called with error argument. So I'll have the output like this:

run test
Error: Artificial error
    at /Users/hackerman/Documents/code/wave/demo.js:16:59
    at Transform.transform [as _transform] (/Users/hackerman/Documents/code/wave/demo.js:11:7)
    at Transform._write (node:internal/streams/transform:184:23)
    at writeOrBuffer (node:internal/streams/writable:400:12)
    at _write (node:internal/streams/writable:341:10)
    at Transform.Writable.write (node:internal/streams/writable:345:10)
    at Object.<anonymous> (/Users/hackerman/Documents/code/wave/demo.js:24:4)
    at Module._compile (node:internal/modules/cjs/loader:1105:14)
    at Object.Module._extensions..js (node:internal/modules/cjs/loader:1137:10)
    at Module.load (node:internal/modules/cjs/loader:973:32)
done

What do you see instead?

pipeline callback is not called. Output that I have:

run test

Additional information

When I run mentioned code with --inspect-brk flag — I can achieve expected behaviour in google chrome console. Steps:

• run node --inspect-brk demo.js
• open chrome://inspect/#devices in chrome
• click inspect and resume script execution

[targos]

@nodejs/streams

[mcollina]

The problem is that process.stdout() never gets closed.

'use strict'

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

const w = new Writable({
  write (chunk, enc, cb) {
    cb()
  }
})

function createTransformStream (tf, context) {
  return new Transform({
    readableObjectMode: true,
    writableObjectMode: true,

    transform (chunk, encoding, done) {
      tf(chunk, context, done)
    }
  })
}

const ts = createTransformStream((chunk, _, done) => done(new Error('Artificial error')))

pipeline(ts, w, (err) => {
  if (err) console.log(err)
  console.log('done')
})

console.log('run test')
ts.write('test')

In theory, this should have been solved by #32373, but I guess there is something different from using a child_process and a tty.

@ronag wdyt?

[sullenor]

Oh, makes sense. Thanks @mcollina for clarifying this issue for me.

[ktfth]

@mcollina I have some questions...
Can i work on that issue?
Where is located the entry points for streams "Transformer"?
I can put the test on test folder and the execution happen because the existance of a mechanism for execution?

[ktfth]

I'm made some progress investigating the code base, but if it's possible to talk more about that could be helpful.

[ktfth]

I have made a change who passes on tests with the case described by you @mcollina here some refs: https://github.com/ktfth/node/tree/fix/stream-pipeline-error-callback

[mcollina]

You need to add a test for stdout. Look in https://github.com/nodejs/node/tree/master/test/pseudo-tty.

[ktfth]

Included another test with process.stdout, but looking the reference you shared @mcollina to create more tests

[ktfth]

Do you have an hint to what I can search for to made a good test?

[ronag]

@ktfth Can you please do this as a proper PR?

[ktfth]

I can do that

[ktfth]

PR: #39533

[rluvaton]

I wanted to know if the problem persist when using an async itaretable function, and it does:

const { Transform, pipeline } =  require('stream')

function createTransformStream () {
  return async function*(stream) {
      yield 'test';
      throw new Error('Artificial Error');
  };
}

const ts = createTransformStream()

pipeline(ts, process.stdout, (err) => {
  if (err) console.log(err);
  console.log('done')
})
console.log('run test')

[rluvaton]

If I'm right in my assumption about the error - #39533 (comment) (in 3-4 words - the process.stdout closed)

A workaround would be to just wrap the process.stdout with it's own async iterator / Writable and it should work

[mcollina]

After doing some work, I think this is not a bug. The following pass:

'use strict'

const { Transform, pipeline } = require('stream')

function createTransformStream (tf, context) {
  return new Transform({
    readableObjectMode: true,
    writableObjectMode: true,

    transform (chunk, encoding, done) {
      tf(chunk, context, done)
    }
  })
}

const ts = createTransformStream((chunk, _, done) => done(new Error('Artificial error')))

process.stdout.on('error', function () {
  process._rawDebug('error emitted')
})


process.stdout.on('close', function () {
  process._rawDebug('close emitted')
})

pipeline(ts, process.stdout, (err) => {
  if (err) process._rawDebug(err)
  process._rawDebug('done')
})

console.log('run test')
ts.write('test')

The reason why you do not see the output is that process.stdout is closed. Maybe we should print a warning in this case (or maybe console.log() should bypass it.

[rluvaton]

After doing some work, I think this is not a bug.

[...]

The reason why you do not see the output is that process.stdout is closed. Maybe we should print a warning in this case (or maybe console.log() should bypass it.

According to #7606 (comment) stdio should never be closed

[mcollina]

the actual file descriptor is never closed. However the stream object process.stdout can be closed as it is a stream.
Otherwise the callback to pipeline would never be called.

[ktfth]

I understand and totally agree

[rluvaton]

the actual file descriptor is never closed. However the stream object process.stdout can be closed as it is a stream.
Otherwise the callback to pipeline would never be called.

Closing the stdio can have an unwanted results:

1. If you run the code in the issue above in REPL you wouldn't see any character you type after that ran
2. Loggers that use process.stdout would not output anything if it will close.

I'm sure there are more reasons, this is just at the top of my head.

I think we should fix the pipeline rather than the process.stdout

[mcollina]

Here is an alternative way to fix this: #39670