From 0a9e96e86cd50a55c913f5931c2b2b0e5bd320e5 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Wed, 21 Jun 2017 20:04:12 +0200 Subject: [PATCH] stream: finish must always follow error When _write completes with an Error, 'finish' was emitted before 'error' if the callback was asynchronous. This commit restore the previous behavior. The logic is still less then ideal, because we call the write() callback before emitting error if asynchronous, but after if synchronous. This commit do not try to change the behavior. This commit fixes a regression introduced by: https://github.com/nodejs/node/pull/13195. Fixes: https://github.com/nodejs/node/issues/13812 PR-URL: https://github.com/nodejs/node/pull/13850 Reviewed-By: James M Snell Reviewed-By: Anna Henningsen Reviewed-By: Calvin Metcalf Reviewed-By: Colin Ihrig --- lib/_stream_writable.js | 30 ++-- ...est-stream-writable-write-writev-finish.js | 156 ++++++++++++++++++ .../test-stream-writable-writev-finish.js | 53 ------ 3 files changed, 175 insertions(+), 64 deletions(-) create mode 100644 test/parallel/test-stream-writable-write-writev-finish.js delete mode 100644 test/parallel/test-stream-writable-writev-finish.js diff --git a/lib/_stream_writable.js b/lib/_stream_writable.js index 615f5a37624f31..78ab13d9063c67 100644 --- a/lib/_stream_writable.js +++ b/lib/_stream_writable.js @@ -374,18 +374,26 @@ function doWrite(stream, state, writev, len, chunk, encoding, cb) { function onwriteError(stream, state, sync, er, cb) { --state.pendingcb; - if (sync) - process.nextTick(afterError, stream, state, cb, er); - else - afterError(stream, state, cb, er); - - stream._writableState.errorEmitted = true; - stream.emit('error', er); -} -function afterError(stream, state, cb, err) { - cb(err); - finishMaybe(stream, state); + if (sync) { + // defer the callback if we are being called synchronously + // to avoid piling up things on the stack + process.nextTick(cb, er); + // this can emit finish, and it will always happen + // after error + process.nextTick(finishMaybe, stream, state); + stream._writableState.errorEmitted = true; + stream.emit('error', er); + } else { + // the caller expect this to happen before if + // it is async + cb(er); + stream._writableState.errorEmitted = true; + stream.emit('error', er); + // this can emit finish, but finish must + // always follow error + finishMaybe(stream, state); + } } function onwriteStateUpdate(state) { diff --git a/test/parallel/test-stream-writable-write-writev-finish.js b/test/parallel/test-stream-writable-write-writev-finish.js new file mode 100644 index 00000000000000..c4aaa36606c5f4 --- /dev/null +++ b/test/parallel/test-stream-writable-write-writev-finish.js @@ -0,0 +1,156 @@ +'use strict'; + +const common = require('../common'); +const assert = require('assert'); +const stream = require('stream'); + +// ensure consistency between the finish event when using cork() +// and writev and when not using them + +{ + const writable = new stream.Writable(); + + writable._write = (chunks, encoding, cb) => { + cb(new Error('write test error')); + }; + + let firstError = false; + writable.on('finish', common.mustCall(function() { + assert.strictEqual(firstError, true); + })); + + writable.on('prefinish', common.mustCall()); + + writable.on('error', common.mustCall((er) => { + assert.strictEqual(er.message, 'write test error'); + firstError = true; + })); + + writable.end('test'); +} + +{ + const writable = new stream.Writable(); + + writable._write = (chunks, encoding, cb) => { + setImmediate(cb, new Error('write test error')); + }; + + let firstError = false; + writable.on('finish', common.mustCall(function() { + assert.strictEqual(firstError, true); + })); + + writable.on('prefinish', common.mustCall()); + + writable.on('error', common.mustCall((er) => { + assert.strictEqual(er.message, 'write test error'); + firstError = true; + })); + + writable.end('test'); +} + +{ + const writable = new stream.Writable(); + + writable._write = (chunks, encoding, cb) => { + cb(new Error('write test error')); + }; + + writable._writev = (chunks, cb) => { + cb(new Error('writev test error')); + }; + + let firstError = false; + writable.on('finish', common.mustCall(function() { + assert.strictEqual(firstError, true); + })); + + writable.on('prefinish', common.mustCall()); + + writable.on('error', common.mustCall((er) => { + assert.strictEqual(er.message, 'writev test error'); + firstError = true; + })); + + writable.cork(); + writable.write('test'); + + setImmediate(function() { + writable.end('test'); + }); +} + +{ + const writable = new stream.Writable(); + + writable._write = (chunks, encoding, cb) => { + setImmediate(cb, new Error('write test error')); + }; + + writable._writev = (chunks, cb) => { + setImmediate(cb, new Error('writev test error')); + }; + + let firstError = false; + writable.on('finish', common.mustCall(function() { + assert.strictEqual(firstError, true); + })); + + writable.on('prefinish', common.mustCall()); + + writable.on('error', common.mustCall((er) => { + assert.strictEqual(er.message, 'writev test error'); + firstError = true; + })); + + writable.cork(); + writable.write('test'); + + setImmediate(function() { + writable.end('test'); + }); +} + +// Regression test for +// https://github.com/nodejs/node/issues/13812 + +{ + const rs = new stream.Readable(); + rs.push('ok'); + rs.push(null); + rs._read = () => {}; + + const ws = new stream.Writable(); + let firstError = false; + + ws.on('finish', common.mustCall(function() { + assert.strictEqual(firstError, true); + })); + ws.on('error', common.mustCall(function() { + firstError = true; + })); + + ws._write = (chunk, encoding, done) => { + setImmediate(done, new Error()); + }; + rs.pipe(ws); +} + +{ + const rs = new stream.Readable(); + rs.push('ok'); + rs.push(null); + rs._read = () => {}; + + const ws = new stream.Writable(); + + ws.on('finish', common.mustNotCall()); + ws.on('error', common.mustCall()); + + ws._write = (chunk, encoding, done) => { + done(new Error()); + }; + rs.pipe(ws); +} diff --git a/test/parallel/test-stream-writable-writev-finish.js b/test/parallel/test-stream-writable-writev-finish.js deleted file mode 100644 index 6f74ca08d2366d..00000000000000 --- a/test/parallel/test-stream-writable-writev-finish.js +++ /dev/null @@ -1,53 +0,0 @@ -'use strict'; - -const common = require('../common'); -const assert = require('assert'); -const stream = require('stream'); - -// ensure consistency between the finish event when using cork() -// and writev and when not using them - -{ - const writable = new stream.Writable(); - - writable._write = (chunks, encoding, cb) => { - cb(new Error('write test error')); - }; - - writable.on('finish', common.mustCall()); - - writable.on('prefinish', common.mustCall()); - - writable.on('error', common.mustCall((er) => { - assert.strictEqual(er.message, 'write test error'); - })); - - writable.end('test'); -} - -{ - const writable = new stream.Writable(); - - writable._write = (chunks, encoding, cb) => { - cb(new Error('write test error')); - }; - - writable._writev = (chunks, cb) => { - cb(new Error('writev test error')); - }; - - writable.on('finish', common.mustCall()); - - writable.on('prefinish', common.mustCall()); - - writable.on('error', common.mustCall((er) => { - assert.strictEqual(er.message, 'writev test error'); - })); - - writable.cork(); - writable.write('test'); - - setImmediate(function() { - writable.end('test'); - }); -}