Problem with Error [ERR_MULTIPLE_CALLBACK]: Callback called multiple times after 16.4.2 -> 16.8.0 upgrade

[markddrake]

• Node.js Version: 16.8
• OS:Windows 10
• Scope (install, code, runtime, meta, other?):streams pipleine

I have code that ran without errors in 16.4.2 which now fails in 16.8

The code is failing with
'
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'
}
2021-08-27T00:50:56.485Z [INFO][PIPELINE][STREAM WRITER][character_types][Postgres][Y

This appears to be caused by the fact that the following code has been added to the standard on finish callback

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

I fully understand that my code should probably have never worked in the first place, but how can I trace where the callback is being invoked so I can work out how it's being called twice....

[markddrake]

Looked into this some more.. 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)})```

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>

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>

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. The questions are

1. • Why in 16.8 is the callback apparently being called twice ? As far as I can tell I am only invoking it once.
2. • Is the callback being called twice in 16.4.2 and the error is simply not getting caught ?
3. • Is the code change in 16.8 to detect multiple call backs correct ?
4. • What changes can I make to my code to avoid this problem moving forward ?

[Ayase-252]

Closing as a duplicate of nodejs/node#39923