streams: synchronous unpipe() causes data stalls

[bnoordhuis]

Continuation of nodejs/node-v0.x-archive#7451. Test case:

var assert = require('assert');
var stream = require('stream');

var a = new stream.Duplex;
var b = new stream.Writable;
var chunks = [];

a._read = function() {};
a._write = a.push.bind(a);
b._write = chunks.push.bind(chunks);

a.pipe(b);
a.write('ping');
a.unpipe(b);

process.on('exit', function() {
  assert(chunks.length > 0);
});

From the original issue:

The unpipe() stops the data from propagating down the chain. That's problematic because I, as a streams2 user that gets slotted somewhere in the chain of handlers, don't know if or when unpipe() gets called.

The observable behavior is that sometimes your pipeline stalls without a clue why. It took me long enough to debug so I strongly doubt that an average user could figure it out.

In the other issue I mentioned that it works with v0.10 but it fails with v0.10.41 and v0.10.45 so it seems there's been a regression somewhere. I think I originally tested with v0.10.26 or v0.10.27-pre.

/cc @nodejs/streams

[mcollina]

@bnoordhuis this works:

var assert = require('assert');
var stream = require('stream');

var a = new stream.Duplex;
var b = new stream.Writable;
var chunks = [];

a._read = function() {};
a._write = a.push.bind(a);
b._write = chunks.push.bind(chunks);

a.pipe(b);
a.write('ping');
process.nextTick(function () {
  a.unpipe(b);
})

process.on('exit', function() {
    assert(chunks.length > 0);
});

This happens because resume() is currently wrapped inside nextTick as part of pipe: https://github.com/nodejs/node/blob/master/lib/_stream_readable.js#L717-L734

I am not sure why that was done in that way, and what problem solved.

[addaleax]

I think this is a duplicate of #1041?

[bnoordhuis]

Looks like it's a duplicate of #1041, yes. I'll close out this one.