Error [ERR_MULTIPLE_CALLBACK]: Callback called multiple times in 16.8.0 with Stream & pipeline

[markddrake]

Version

16.8.0

Platform

Microsoft Windows NT 10.0.19043.0 x64

Subsystem

Streams pipeline

What steps will reproduce the bug?

Here's a simple testcase

const {PassThrough, pipeline, finished} = require('stream')
const crypto = require('crypto')
const fs = require('fs')
const { createGzip } = require('zlib');

class Switcher extends PassThrough {

  constructor(end) {
	super()
	this.endOption = end
	this.id = crypto.randomBytes(16).toString("hex").toUpperCase();
  }
  
  pipe(os,options) {
    options = options || {}
	options.end = this.endOption
	return super.pipe(os,options)
  }
  
  _transform(data,enc,callback) {
    this.push(data)
    callback()
  }
 
  async _final(callback) {
    console.log(this.constructor.name,'_final()',this.id)
    console.log(callback.toString())
	callback()
  }	   
}

class PipelineTest {

  async testPipeline() {

    const writer = fs.createWriteStream('c:\\temp\\output.zip',{flags :"w"})
    const zipper = createGzip()
    const switcher1 = new Switcher(true)
    const input1 = fs.createReadStream('c:\\temp\\HR.JSON')
    const pipeline1 = [input1,switcher1,zipper,writer]

    const inputFinished = new Promise((resolve,reject) => {
       pipeline(pipeline1,(e)=>{if(e) { reject(e) } else {resolve()}})
	})
    await inputFinished
	
  }

}

async function  main() {
  const t = new PipelineTest()
  await t.testPipeline()
  console.log('X')
}
  
main().then(() => { console.log('Done')}).catch((e) => {console.log(e)})```

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

All the time

What is the expected behavior?

In 16.4.2 the error is not raised

C:\Development\YADAMU>node --version
v16.4.2

C:\Development\YADAMU>node scratch\test1.js
Switcher _final() 2D147E3E5A57F9559FAA7F0EE471D8AD
X
Done

C:\Development\YADAMU>

What do you see instead?

In 16.8.0. this fails with

C:\yadamu>node --version
v16.8.0

C:\yadamu>node test1.js
Switcher _final() 6E61234C5C334338C4C460622EA4D903
Error [ERR_MULTIPLE_CALLBACK]: Callback called multiple times
    at NodeError (node:internal/errors:371:5)
    at onFinish (node:internal/streams/writable:667:37)
    at processTicksAndRejections (node:internal/process/task_queues:82:21) {
  code: 'ERR_MULTIPLE_CALLBACK'
}

C:\yadamu>

Additional information

The change in behavoir seems to arise as a result of a code change to the callback's added by pipeline

In 16.4.2 the callback code generated by node streams pipeline is

(err) => {
    state.pendingcb--;
    if (err) {
      const onfinishCallbacks = state[kOnFinished].splice(0);
      for (let i = 0; i < onfinishCallbacks.length; i++) {
        onfinishCallbacks[i](err);
      }
      errorOrDestroy(stream, err, state.sync);
    } else if (needFinish(state)) {
      state.prefinished = true;
      stream.emit('prefinish');
      // Backwards compat. Don't check state.sync here.
      // Some streams assume 'finish' will be emitted
      // asynchronously relative to _final callback.
      state.pendingcb++;
      process.nextTick(finish, stream, state);
    }
  }

Where as in 16.8.0 it is

function onFinish(err) {
    if (called) {
      errorOrDestroy(stream, err ?? ERR_MULTIPLE_CALLBACK());
      return;
    }
    called = true;

    state.pendingcb--;
    if (err) {
      const onfinishCallbacks = state[kOnFinished].splice(0);
      for (let i = 0; i < onfinishCallbacks.length; i++) {
        onfinishCallbacks[i](err);
      }
      errorOrDestroy(stream, err, state.sync);
    } else if (needFinish(state)) {
      state.prefinished = true;
      stream.emit('prefinish');
      // Backwards compat. Don't check state.sync here.
      // Some streams assume 'finish' will be emitted
      // asynchronously relative to _final callback.
      state.pendingcb++;
      process.nextTick(finish, stream, state);
    }
  }

So the code attached by pipeline has clearly changed and is resulting in a change of behavior causing an error to be raised where no error was raised before.

However I am not clear why node thinks the callback has been called twice. As far as I can tell I am only invoking it once.

As I see it there are 4 possibilities:

1. Node is correct, the callback is being invoked twice and my code is incorrect, in which case any hints as to what needs to changed would be much appreciated.

2. Node has always (incorrectly) caused the callback to be executed twice. However as a result of the code changes between 16.4.2 and 16.8 this is now causing an error to be raised.

3. There is a new problem in Node 16.8 that is causing the callback to be invoked twice.

4. The code added to detect multiple invocations of the callback is incorrect

[Linkgoron]

I think that this is a duplicate of #39535

[ronag]

You can't both return a promise (async) and call the callback.

[markddrake]

Sorry, this example does not really show why I made _final async. In the real world I may be doing something that involves an asynchronous operation, such as invoking a database's commit in my _final() implementation. I need the async to allow me use await with this call. If I remove the async I cannot use await. How would I code a _final() implementation that needs to wait for an asynchronous operation to complete before invoking the callback, do I simply have to revert to a 'then' based model ?

[ronag]

You should either do callback based implementation or promise based. Mixing the two is usually a bad idea.

[markddrake]

If I am implementing a stream I am forced to invoke the callback since that is what the specification defines, I don't have any choice in the matter do I ?. Are you saying that if I am implementing any of the methods defined by the streams API I have to totally abandon the clean coding model offered by async/await. Can I still use promise.then().catch() safely or do I really have to resort to 'callback hell'

[ronag]

You don’t have to call the callback if you return a promise.

[markddrake]

Ah, interesting let me experiment...

[albanm]

Isn't there something missing from the documentation then ? If I understand correctly this should specify that the method can either execute the callback or return a promise https://nodejs.org/api/stream.html#writable_finalcallback

[albanm]

And it looks like this is the behavior for _final, but not _write, _transform, etc.. Shouldn't this be the new default for all stream methods ?

edit: sorry I just realized that the discussion is actually here #39535