stream: don't emit finish on error

After 'error' only 'close' should be emitted.

PR-URL: https://github.com/nodejs/node/pull/28979
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>

doc/api/stream.md

@@ -2425,7 +2425,8 @@ and `stream.Readable` classes, respectively. The `'finish'` event is emitted
 after [`stream.end()`][stream-end] is called and all chunks have been processed
 by [`stream._transform()`][stream-_transform]. The `'end'` event is emitted
 after all data has been output, which occurs after the callback in
-[`transform._flush()`][stream-_flush] has been called.
+[`transform._flush()`][stream-_flush] has been called. In the case of an error,
+neither `'finish'` nor `'end'` should be emitted.
 
 #### transform.\_flush(callback)
 

lib/_stream_writable.js

@@ -437,19 +437,12 @@ function onwriteError(stream, state, sync, er, cb) {
     // 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);
-    errorOrDestroy(stream, er);
   } else {
     // The caller expect this to happen before if
     // it is async
     cb(er);
-    errorOrDestroy(stream, er);
-    // This can emit finish, but finish must
-    // always follow error
-    finishMaybe(stream, state);
   }
+  errorOrDestroy(stream, er);
 }
 
 function onwrite(stream, er) {
@@ -618,6 +611,7 @@ Object.defineProperty(Writable.prototype, 'writableLength', {
 function needFinish(state) {
   return (state.ending &&
           state.length === 0 &&
+          !state.errorEmitted &&
           state.bufferedRequest === null &&
           !state.finished &&
           !state.writing);

test/parallel/test-net-connect-buffer.js

@@ -51,34 +51,6 @@ tcp.listen(0, common.mustCall(function() {
   assert.strictEqual(socket.connecting, true);
   assert.strictEqual(socket.readyState, 'opening');
 
-  // Make sure that anything besides a buffer or a string throws.
-  common.expectsError(() => socket.write(null),
-                      {
-                        code: 'ERR_STREAM_NULL_VALUES',
-                        type: TypeError,
-                        message: 'May not write null values to stream'
-                      });
-  [
-    true,
-    false,
-    undefined,
-    1,
-    1.0,
-    +Infinity,
-    -Infinity,
-    [],
-    {}
-  ].forEach((value) => {
-    // We need to check the callback since 'error' will only
-    // be emitted once per instance.
-    socket.write(value, common.expectsError({
-      code: 'ERR_INVALID_ARG_TYPE',
-      type: TypeError,
-      message: 'The "chunk" argument must be one of type string or Buffer. ' +
-               `Received type ${typeof value}`
-    }));
-  });
-
   // Write a string that contains a multi-byte character sequence to test that
   // `bytesWritten` is incremented with the # of bytes, not # of characters.
   const a = "L'État, c'est ";

test/parallel/test-net-socket-write-error.js

@@ -13,7 +13,7 @@ function connectToServer() {
                           type: TypeError
                         });
 
-    client.end();
+    client.destroy();
   })
-  .on('end', () => server.close());
+  .on('close', () => server.close());
 }

test/parallel/test-net-write-arguments.js

@@ -0,0 +1,33 @@
+'use strict';
+const common = require('../common');
+const net = require('net');
+
+const socket = net.Stream({ highWaterMark: 0 });
+
+// Make sure that anything besides a buffer or a string throws.
+common.expectsError(() => socket.write(null),
+                    {
+                      code: 'ERR_STREAM_NULL_VALUES',
+                      type: TypeError,
+                      message: 'May not write null values to stream'
+                    });
+[
+  true,
+  false,
+  undefined,
+  1,
+  1.0,
+  +Infinity,
+  -Infinity,
+  [],
+  {}
+].forEach((value) => {
+  // We need to check the callback since 'error' will only
+  // be emitted once per instance.
+  socket.write(value, common.expectsError({
+    code: 'ERR_INVALID_ARG_TYPE',
+    type: TypeError,
+    message: 'The "chunk" argument must be one of type string or Buffer. ' +
+              `Received type ${typeof value}`
+  }));
+});

test/parallel/test-stream-writable-write-writev-finish.js

@@ -14,16 +14,10 @@ const stream = require('stream');
     cb(new Error('write test error'));
   };
 
-  let firstError = false;
+  writable.on('finish', common.mustNotCall());
-  writable.on('finish', common.mustCall(function() {
+  writable.on('prefinish', common.mustNotCall());
-    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');
@@ -36,16 +30,10 @@ const stream = require('stream');
     setImmediate(cb, new Error('write test error'));
   };
 
-  let firstError = false;
+  writable.on('finish', common.mustNotCall());
-  writable.on('finish', common.mustCall(function() {
+  writable.on('prefinish', common.mustNotCall());
-    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');
@@ -62,16 +50,10 @@ const stream = require('stream');
     cb(new Error('writev test error'));
   };
 
-  let firstError = false;
+  writable.on('finish', common.mustNotCall());
-  writable.on('finish', common.mustCall(function() {
+  writable.on('prefinish', common.mustNotCall());
-    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();
@@ -93,16 +75,10 @@ const stream = require('stream');
     setImmediate(cb, new Error('writev test error'));
   };
 
-  let firstError = false;
+  writable.on('finish', common.mustNotCall());
-  writable.on('finish', common.mustCall(function() {
+  writable.on('prefinish', common.mustNotCall());
-    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();
@@ -123,14 +99,9 @@ const stream = require('stream');
   rs._read = () => {};
 
   const ws = new stream.Writable();
-  let firstError = false;
 
-  ws.on('finish', common.mustCall(function() {
+  ws.on('finish', common.mustNotCall());
-    assert.strictEqual(firstError, true);
+  ws.on('error', common.mustCall());
-  }));
-  ws.on('error', common.mustCall(function() {
-    firstError = true;
-  }));
 
   ws._write = (chunk, encoding, done) => {
     setImmediate(done, new Error());