mirror of
https://github.com/nodejs/node.git
synced 2026-10-28 10:15:07 +08:00
Mahin Anowar
39f427d3cc
When a source is piped to multiple destinations and one destination errors synchronously in `_write()`, `write()` returns false without setting `needDrain` (since the "avoid unnecessary drain for sync stream" change). The pipe cleanup only re-invoked the source's drain handler when the destination still had `needDrain` set, so the errored destination was never removed from the source's `awaitDrainWriters` set. The source stayed paused forever and any remaining healthy destination stopped receiving data. Always invoke the drain handler during cleanup. `pipeOnDrain` only removes this destination from the awaiting-drain set and resumes the source once nothing else is awaiting a drain, so it is safe to call unconditionally and no-ops when this destination was not awaiting a drain. Fixes: https://github.com/nodejs/node/issues/53185 Signed-off-by: Mahin Anowar <86069420+MahinAnowar@users.noreply.github.com> PR-URL: https://github.com/nodejs/node/pull/64310 Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
49 lines
1.5 KiB
JavaScript
49 lines
1.5 KiB
JavaScript
'use strict';
|
|
const common = require('../common');
|
|
|
|
// Regression test for https://github.com/nodejs/node/issues/53185
|
|
// When a source is piped to multiple destinations and one destination errors
|
|
// synchronously in `_write`, the source must keep flowing to the remaining
|
|
// healthy destination(s) instead of stalling forever.
|
|
|
|
const assert = require('assert');
|
|
const { PassThrough, Writable, finished } = require('stream');
|
|
|
|
const source = new PassThrough({ highWaterMark: 16 * 1024 });
|
|
|
|
// A destination that errors synchronously on its first write.
|
|
const failing = new Writable({
|
|
highWaterMark: 16 * 1024,
|
|
write(chunk, encoding, callback) {
|
|
callback(new Error('boom'));
|
|
},
|
|
});
|
|
failing.on('error', common.mustCall());
|
|
|
|
// A healthy destination that counts everything it receives.
|
|
let received = 0;
|
|
const healthy = new Writable({
|
|
highWaterMark: 16 * 1024,
|
|
write(chunk, encoding, callback) {
|
|
received += chunk.length;
|
|
callback();
|
|
},
|
|
});
|
|
|
|
source.pipe(failing);
|
|
source.pipe(healthy);
|
|
|
|
// Two chunks each larger than the highWaterMark so that `write()` returns
|
|
// false and the source is paused awaiting a drain.
|
|
const chunk = Buffer.alloc(20000, 'a');
|
|
source.write(chunk);
|
|
source.write(chunk);
|
|
source.end();
|
|
|
|
finished(healthy, common.mustSucceed(() => {
|
|
// Without the fix, the errored destination stays in the source's
|
|
// awaitDrainWriters set, the source never resumes, and `healthy` only
|
|
// receives the first chunk.
|
|
assert.strictEqual(received, chunk.length * 2);
|
|
}));
|