Errors on the writable stream are emitted twice on the duplex

[stezu]

With default options, the error on the first stream is emitted twice:

var duplexer = require('duplexer2');
var through = require('through2');

var combinedStream = duplexer(
  through(function(chunk, enc, next) {
    next(new Error('this error is emitted twice.'));
  }),
  through(function(chunk, enc, next) {
    next(null, chunk);
  })
);

combinedStream
  .on('error', function(err) {
    // this event happens twice with the same error
    console.log('\n\n')
    console.log('error:', err);
  })
  .end('beep');

If I set bubbleErrors to false, the first stream throws:

var duplexer = require('duplexer2');
var through = require('through2');

var combinedStream = duplexer({
    bubbleErrors: false
  },
  through(function(chunk, enc, next) {
    // this error is thrown
    next(new Error('this error is emitted twice.'));
  }),
  through(function(chunk, enc, next) {
    next(null, chunk);
  })
);

combinedStream
  .on('error', function(err) {
    console.log('\n\n')
    console.log('error:', err);
  })
  .end('beep');

Interestingly, if I put an error handler on the first stream, both error handlers are called:

var duplexer = require('duplexer2');
var through = require('through2');

var combinedStream = duplexer({
    bubbleErrors: false
  },
  through(function(chunk, enc, next) {
    next(new Error('this error is emitted twice.'));
  })
  .on('error', function(err) {
    // this event happens first
    console.log('\n\n');
    console.log('first stream error', err);
  }),
  through(function(chunk, enc, next) {
    next(null, chunk);
  })
);

combinedStream
  .on('error', function(err) {
    // this event happens second
    console.log('\n\n')
    console.log('error:', err);
  })
  .end('beep');

This happens in Node 0.10, 0.12 and 6.6.0, I assume it happens in all of the versions.

I first brought this up in stream-combiner2, but the problem appears to happen here (or maybe even lower down).

Do you know what's happening and how it can be fixed? The only thing I can think of doing on my end is using the last code snippet with a noop on the first stream's error handler, but that feels like a hack.

[jcuffe]

This is a very old issue, but I recently had need for stream composition in my project so I wanted to investigate to make sure our code wasn't vulnerable to any side effects of this bug.

Long story short: the done callback supplied to DuplexWrapper on write calls is passed to the enclosed writable. When the writable invokes the callback with an error, this error is emitted directly from the DuplexWrapper:

DuplexWrapper.prototype._write = function _write(input, encoding, done) {
  this._writable.write(input, encoding, done);
};

For some reason, this error is also being emitted from the Writable, which can be verified by replacing the through2 writable with a native instance and listening to its errors:

const writable = new Writable({
  write(chunk, enc, callback) {
    callback(new Error('emitted from writable'));
  },
});
writer.on('error', (err) => console.log('writable error', err.message))

From what I can see, since an error in write() will always be emitted on the enclosed writable stream, the simplest solution is to ignore the error provided to the DuplexWrapper's _write() callback:

  DuplexWrapper.prototype._write = function _write(input, encoding, done) {
-    this._writable.write(input, encoding, done);    
+    this._writable.write(input, encoding, () => done);
  };

edit: My solution had me pretty nervous, so I found some additional context to verify that, if re-emitting errors from the writable, there is no need to also propagate write errors with the done callback: Writable internal write callback

This is the callback provided to your custom _write function. Any callback supplied by callers of write() is captured and passed in to this method as state. When called with an error, the callback emits an error on the stream in addition to passing the error along to the captured callback.

Short story long: if you're re-emitting all errors from the writable, you can ignore any error passed to the callback for write().