stream.pipeline does not invoke callback when error happens

[mightyaleksey]

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?

[mightyaleksey]

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?