From 354c8974bba9c7b440797870b0782715bd27e53a Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Sun, 18 Aug 2019 23:38:35 +0200
Subject: [PATCH] stream: fix multiple destroy calls

Previously destroy could be called multiple times causing inconsistent
and hard to predict behavior. Furthermore, since the stream _destroy
implementation can only be called once, the behavior of applying destroy
multiple times becomes unclear.

This changes so that only the first destroy() call is executed and any
subsequent calls are noops.
---
 doc/api/stream.md                             | 11 +++++++
 lib/_tls_wrap.js                              |  2 +-
 lib/internal/fs/streams.js                    |  2 +-
 lib/internal/streams/destroy.js               | 33 ++++++++++---------
 test/parallel/test-file-write-stream.js       |  3 +-
 test/parallel/test-file-write-stream2.js      | 13 +++-----
 ...est-http2-server-stream-session-destroy.js |  8 ++---
 test/parallel/test-net-socket-destroy-send.js |  6 +---
 test/parallel/test-stream-catch-rejections.js |  5 ++-
 test/parallel/test-stream-pipeline.js         |  4 +--
 test/parallel/test-stream-readable-destroy.js | 12 +++----
 test/parallel/test-stream-writable-destroy.js | 22 +++++--------
 .../parallel/test-stream-writable-writable.js |  1 -
 .../test-stream-writable-write-error.js       |  1 -
 .../test-tls-wrap-econnreset-localaddress.js  |  5 +++
 .../parallel/test-tls-wrap-econnreset-pipe.js |  5 +++
 .../test-tls-wrap-econnreset-socket.js        |  5 +++
 test/parallel/test-tls-wrap-econnreset.js     |  5 +++
 test/parallel/test-zlib-write-after-close.js  |  7 +---
 19 files changed, 77 insertions(+), 73 deletions(-)

diff --git a/doc/api/stream.md b/doc/api/stream.md
index cf3ecd461c13ed..6a32316d4dbb83 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -385,6 +385,10 @@ This is a destructive and immediate way to destroy a stream. Previous calls to
 `write()` may not have drained, and may trigger an `ERR_STREAM_DESTROYED` error.
 Use `end()` instead of destroy if data should flush before close, or wait for
 the `'drain'` event before destroying the stream.
+
+Once `destroy()` has been called any further calls will be a noop and no
+further errors except from `_destroy` may be emitted as `'error'`.
+
 Implementors should not override this method,
 but instead implement [`writable._destroy()`][writable-_destroy].
 
@@ -953,6 +957,10 @@ Destroy the stream. Optionally emit an `'error'` event, and emit a `'close'`
 event (unless `emitClose` is set to `false`). After this call, the readable
 stream will release any internal resources and subsequent calls to `push()`
 will be ignored.
+
+Once `destroy()` has been called any further calls will be a noop and no
+further errors except from `_destroy` may be emitted as `'error'`.
+
 Implementors should not override this method, but instead implement
 [`readable._destroy()`][readable-_destroy].
 
@@ -1484,6 +1492,9 @@ Implementors should not override this method, but instead implement
 The default implementation of `_destroy()` for `Transform` also emit `'close'`
 unless `emitClose` is set in false.
 
+Once `destroy()` has been called any further calls will be a noop and no
+further errors except from `_destroy` may be emitted as `'error'`.
+
 ### `stream.finished(stream[, options], callback)`
 <!-- YAML
 added: v10.0.0
diff --git a/lib/_tls_wrap.js b/lib/_tls_wrap.js
index aeb1cfe7af9682..cb3ab6e13685fe 100644
--- a/lib/_tls_wrap.js
+++ b/lib/_tls_wrap.js
@@ -1611,7 +1611,7 @@ exports.connect = function connect(...args) {
     tlssock._start();
 
   tlssock.on('secure', onConnectSecure);
-  tlssock.once('end', onConnectEnd);
+  tlssock.prependListener('end', onConnectEnd);
 
   return tlssock;
 };
diff --git a/lib/internal/fs/streams.js b/lib/internal/fs/streams.js
index 5b37dca74bed63..a76a8f6895c887 100644
--- a/lib/internal/fs/streams.js
+++ b/lib/internal/fs/streams.js
@@ -458,7 +458,7 @@ WriteStream.prototype._writev = function(data, cb) {
 
     if (er) {
       if (this.autoClose) {
-        this.destroy();
+        this.destroy(er);
       }
       return cb(er);
     }
diff --git a/lib/internal/streams/destroy.js b/lib/internal/streams/destroy.js
index 170ddb3d14f0b7..3a953afd445649 100644
--- a/lib/internal/streams/destroy.js
+++ b/lib/internal/streams/destroy.js
@@ -7,6 +7,15 @@ function destroy(err, cb) {
   const r = this._readableState;
   const w = this._writableState;
 
+  if ((w && w.destroyed) || (r && r.destroyed)) {
+    if (typeof cb === 'function') {
+      // TODO(ronag): Invoke with `'close'`/`'error'`.
+      cb();
+    }
+
+    return this;
+  }
+
   if (err) {
     if (w) {
       w.errored = true;
@@ -16,16 +25,6 @@ function destroy(err, cb) {
     }
   }
 
-  if ((w && w.destroyed) || (r && r.destroyed)) {
-    if (cb) {
-      cb(err);
-    } else if (err) {
-      process.nextTick(emitErrorNT, this, err);
-    }
-
-    return this;
-  }
-
   // We set destroyed to true before firing error callbacks in order
   // to make it re-entrance safe in case destroy() is called within callbacks
 
@@ -53,13 +52,11 @@ function destroy(err, cb) {
       r.closed = true;
     }
 
-    if (cb) {
-      // Invoke callback before scheduling emitClose so that callback
-      // can schedule before.
+    if (typeof cb === 'function') {
       cb(err);
-      // Don't emit 'error' if passed a callback.
-      process.nextTick(emitCloseNT, this);
-    } else if (err) {
+    }
+
+    if (err) {
       process.nextTick(emitErrorCloseNT, this, err);
     } else {
       process.nextTick(emitCloseNT, this);
@@ -138,6 +135,10 @@ function errorOrDestroy(stream, err, sync) {
   const r = stream._readableState;
   const w = stream._writableState;
 
+  if ((w && w.destroyed) || (r && r.destroyed)) {
+    return this;
+  }
+
   if ((r && r.autoDestroy) || (w && w.autoDestroy))
     stream.destroy(err);
   else if (err) {
diff --git a/test/parallel/test-file-write-stream.js b/test/parallel/test-file-write-stream.js
index 1055fac698fa92..6cab71a4353ff2 100644
--- a/test/parallel/test-file-write-stream.js
+++ b/test/parallel/test-file-write-stream.js
@@ -63,8 +63,7 @@ file
 
     callbacks.close++;
     console.error('write after end should not be allowed');
-    file.write('should not work anymore');
-    file.on('error', common.expectsError({
+    file.write('should not work anymore', common.expectsError({
       code: 'ERR_STREAM_WRITE_AFTER_END',
       name: 'Error',
       message: 'write after end'
diff --git a/test/parallel/test-file-write-stream2.js b/test/parallel/test-file-write-stream2.js
index 2654c8bca20712..487b5d44054373 100644
--- a/test/parallel/test-file-write-stream2.js
+++ b/test/parallel/test-file-write-stream2.js
@@ -33,7 +33,7 @@ const filepath = path.join(tmpdir.path, 'write.txt');
 
 const EXPECTED = '012345678910';
 
-const cb_expected = 'write open drain write drain close error ';
+const cb_expected = 'write open drain write drain close ';
 let cb_occurred = '';
 
 let countDrains = 0;
@@ -92,16 +92,11 @@ file.on('drain', function() {
 file.on('close', function() {
   cb_occurred += 'close ';
   assert.strictEqual(file.bytesWritten, EXPECTED.length * 2);
-  file.write('should not work anymore');
+  file.write('should not work anymore', (err) => {
+    assert.ok(err.message.includes('write after end'));
+  });
 });
 
-
-file.on('error', function(err) {
-  cb_occurred += 'error ';
-  assert.ok(err.message.includes('write after end'));
-});
-
-
 for (let i = 0; i < 11; i++) {
   const ret = file.write(String(i));
   console.error(`${i} ${ret}`);
diff --git a/test/parallel/test-http2-server-stream-session-destroy.js b/test/parallel/test-http2-server-stream-session-destroy.js
index b25dfbb347d9bf..a03569cdee7825 100644
--- a/test/parallel/test-http2-server-stream-session-destroy.js
+++ b/test/parallel/test-http2-server-stream-session-destroy.js
@@ -34,11 +34,9 @@ server.on('stream', common.mustCall((stream) => {
       name: 'Error'
     }
   );
-  stream.on('error', common.expectsError({
-    name: 'Error',
-    code: 'ERR_STREAM_WRITE_AFTER_END',
-    message: 'write after end'
-  }));
+  // When session is detroyed all streams are destroyed and no further
+  // error should be emitted.
+  stream.on('error', common.mustNotCall());
   assert.strictEqual(stream.write('data', common.expectsError({
     name: 'Error',
     code: 'ERR_STREAM_WRITE_AFTER_END',
diff --git a/test/parallel/test-net-socket-destroy-send.js b/test/parallel/test-net-socket-destroy-send.js
index 54dadd94861ddf..db792ad6d3b921 100644
--- a/test/parallel/test-net-socket-destroy-send.js
+++ b/test/parallel/test-net-socket-destroy-send.js
@@ -12,11 +12,7 @@ server.listen(0, common.mustCall(function() {
   conn.on('connect', common.mustCall(function() {
     // Test destroy returns this, even on multiple calls when it short-circuits.
     assert.strictEqual(conn, conn.destroy().destroy());
-    conn.on('error', common.expectsError({
-      code: 'ERR_STREAM_DESTROYED',
-      message: 'Cannot call write after a stream was destroyed',
-      name: 'Error'
-    }));
+    conn.on('error', common.mustNotCall());
 
     conn.write(Buffer.from('kaboom'), common.expectsError({
       code: 'ERR_STREAM_DESTROYED',
diff --git a/test/parallel/test-stream-catch-rejections.js b/test/parallel/test-stream-catch-rejections.js
index fb5f1fccc18bd2..848c2ada130e64 100644
--- a/test/parallel/test-stream-catch-rejections.js
+++ b/test/parallel/test-stream-catch-rejections.js
@@ -8,11 +8,10 @@ const assert = require('assert');
   const r = new stream.Readable({
     captureRejections: true,
     read() {
-      this.push('hello');
-      this.push('world');
-      this.push(null);
     }
   });
+  r.push('hello');
+  r.push('world');
 
   const err = new Error('kaboom');
 
diff --git a/test/parallel/test-stream-pipeline.js b/test/parallel/test-stream-pipeline.js
index 19fc246e2bf3cd..796f9740f8203b 100644
--- a/test/parallel/test-stream-pipeline.js
+++ b/test/parallel/test-stream-pipeline.js
@@ -505,9 +505,7 @@ const { promisify } = require('util');
         res,
         stream,
         common.mustCall((err) => {
-          assert.ok(err);
-          // TODO(ronag):
-          // assert.strictEqual(err.message, 'oh no');
+          assert.strictEqual(err.message, 'oh no');
           server.close();
         })
       );
diff --git a/test/parallel/test-stream-readable-destroy.js b/test/parallel/test-stream-readable-destroy.js
index d446dd3611781e..6caf88a1f151e8 100644
--- a/test/parallel/test-stream-readable-destroy.js
+++ b/test/parallel/test-stream-readable-destroy.js
@@ -183,12 +183,12 @@ const assert = require('assert');
 
   let ticked = false;
   read.on('close', common.mustCall(() => {
-    assert.strictEqual(read._readableState.errorEmitted, false);
+    assert.strictEqual(read._readableState.errorEmitted, true);
     assert.strictEqual(ticked, true);
   }));
-  // 'error' should not be emitted since a callback is passed to
-  // destroy(err, callback);
-  read.on('error', common.mustNotCall());
+  read.on('error', common.mustCall((err) => {
+    assert.strictEqual(err, expected);
+  }));
 
   assert.strictEqual(read._readableState.errored, false);
   assert.strictEqual(read._readableState.errorEmitted, false);
@@ -217,7 +217,7 @@ const assert = require('assert');
   }));
   readable.on('error', common.mustCall((err) => {
     assert.strictEqual(ticked, true);
-    assert.strictEqual(err.message, 'kaboom 2');
+    assert.strictEqual(err.message, 'kaboom 1');
     assert.strictEqual(readable._readableState.errorEmitted, true);
   }));
 
@@ -230,7 +230,7 @@ const assert = require('assert');
   // the `_destroy()` callback is called.
   readable.destroy(new Error('kaboom 2'));
   assert.strictEqual(readable._readableState.errorEmitted, false);
-  assert.strictEqual(readable._readableState.errored, true);
+  assert.strictEqual(readable._readableState.errored, false);
 
   ticked = true;
 }
diff --git a/test/parallel/test-stream-writable-destroy.js b/test/parallel/test-stream-writable-destroy.js
index 2a9a1965adbe37..706847a8582d0c 100644
--- a/test/parallel/test-stream-writable-destroy.js
+++ b/test/parallel/test-stream-writable-destroy.js
@@ -187,12 +187,14 @@ const assert = require('assert');
 
   let ticked = false;
   writable.on('close', common.mustCall(() => {
+    writable.on('error', common.mustNotCall());
+    writable.destroy(new Error('hello'));
     assert.strictEqual(ticked, true);
     assert.strictEqual(writable._writableState.errorEmitted, true);
   }));
   writable.on('error', common.mustCall((err) => {
     assert.strictEqual(ticked, true);
-    assert.strictEqual(err.message, 'kaboom 2');
+    assert.strictEqual(err.message, 'kaboom 1');
     assert.strictEqual(writable._writableState.errorEmitted, true);
   }));
 
@@ -205,7 +207,7 @@ const assert = require('assert');
   // the `_destroy()` callback is called.
   writable.destroy(new Error('kaboom 2'));
   assert.strictEqual(writable._writableState.errorEmitted, false);
-  assert.strictEqual(writable._writableState.errored, true);
+  assert.strictEqual(writable._writableState.errored, false);
 
   ticked = true;
 }
@@ -246,8 +248,8 @@ const assert = require('assert');
 
   const expected = new Error('kaboom');
 
-  write.destroy(expected, common.mustCall(function(err) {
-    assert.strictEqual(err, expected);
+  write.destroy(expected, common.mustCall((err) => {
+    assert.strictEqual(err, undefined);
   }));
 }
 
@@ -271,11 +273,7 @@ const assert = require('assert');
   const write = new Writable();
 
   write.destroy();
-  write.on('error', common.expectsError({
-    name: 'Error',
-    code: 'ERR_STREAM_DESTROYED',
-    message: 'Cannot call write after a stream was destroyed'
-  }));
+  write.on('error', common.mustNotCall());
   write.write('asd', common.expectsError({
     name: 'Error',
     code: 'ERR_STREAM_DESTROYED',
@@ -288,11 +286,7 @@ const assert = require('assert');
     write(chunk, enc, cb) { cb(); }
   });
 
-  write.on('error', common.expectsError({
-    name: 'Error',
-    code: 'ERR_STREAM_DESTROYED',
-    message: 'Cannot call write after a stream was destroyed'
-  }));
+  write.on('error', common.mustNotCall());
 
   write.cork();
   write.write('asd', common.mustCall());
diff --git a/test/parallel/test-stream-writable-writable.js b/test/parallel/test-stream-writable-writable.js
index 98c1eab7719a22..ef5454dc52ef35 100644
--- a/test/parallel/test-stream-writable-writable.js
+++ b/test/parallel/test-stream-writable-writable.js
@@ -23,7 +23,6 @@ const { Writable } = require('stream');
   w.write('asd');
   assert.strictEqual(w.writable, false);
   w.on('error', common.mustCall());
-  w.destroy();
 }
 
 {
diff --git a/test/parallel/test-stream-writable-write-error.js b/test/parallel/test-stream-writable-write-error.js
index e23b24a19df487..c87ddd5dabbe38 100644
--- a/test/parallel/test-stream-writable-write-error.js
+++ b/test/parallel/test-stream-writable-write-error.js
@@ -35,7 +35,6 @@ function test(autoDestroy) {
       _write() {}
     });
     w.destroy();
-    expectError(w, 'asd', 'ERR_STREAM_DESTROYED');
   }
 
   {
diff --git a/test/parallel/test-tls-wrap-econnreset-localaddress.js b/test/parallel/test-tls-wrap-econnreset-localaddress.js
index 9df145ac374ab7..30d3a8873fa8f0 100644
--- a/test/parallel/test-tls-wrap-econnreset-localaddress.js
+++ b/test/parallel/test-tls-wrap-econnreset-localaddress.js
@@ -13,6 +13,7 @@ const server = net.createServer((c) => {
 }).listen(common.mustCall(() => {
   const port = server.address().port;
 
+  let errored = false;
   tls.connect({
     port: port,
     localAddress: common.localhostIPv4
@@ -24,5 +25,9 @@ const server = net.createServer((c) => {
       assert.strictEqual(e.port, port);
       assert.strictEqual(e.localAddress, common.localhostIPv4);
       server.close();
+      errored = true;
+    }))
+    .on('close', common.mustCall(() => {
+      assert.strictEqual(errored, true);
     }));
 }));
diff --git a/test/parallel/test-tls-wrap-econnreset-pipe.js b/test/parallel/test-tls-wrap-econnreset-pipe.js
index a5cba7e47bfd06..f294f23f1d00c9 100644
--- a/test/parallel/test-tls-wrap-econnreset-pipe.js
+++ b/test/parallel/test-tls-wrap-econnreset-pipe.js
@@ -31,6 +31,7 @@ if (process.argv[2] !== 'child') {
 const server = net.createServer((c) => {
   c.end();
 }).listen(common.PIPE, common.mustCall(() => {
+  let errored = false;
   tls.connect({ path: common.PIPE })
     .once('error', common.mustCall((e) => {
       assert.strictEqual(e.code, 'ECONNRESET');
@@ -39,5 +40,9 @@ const server = net.createServer((c) => {
       assert.strictEqual(e.host, undefined);
       assert.strictEqual(e.localAddress, undefined);
       server.close();
+      errored = true;
+    }))
+    .on('close', common.mustCall(() => {
+      assert.strictEqual(errored, true);
     }));
 }));
diff --git a/test/parallel/test-tls-wrap-econnreset-socket.js b/test/parallel/test-tls-wrap-econnreset-socket.js
index 672da9876fd005..ec305b785e0545 100644
--- a/test/parallel/test-tls-wrap-econnreset-socket.js
+++ b/test/parallel/test-tls-wrap-econnreset-socket.js
@@ -15,6 +15,7 @@ const server = net.createServer((c) => {
 
   const socket = new net.Socket();
 
+  let errored = false;
   tls.connect({ socket })
     .once('error', common.mustCall((e) => {
       assert.strictEqual(e.code, 'ECONNRESET');
@@ -22,7 +23,11 @@ const server = net.createServer((c) => {
       assert.strictEqual(e.host, undefined);
       assert.strictEqual(e.port, undefined);
       assert.strictEqual(e.localAddress, undefined);
+      errored = true;
       server.close();
+    }))
+    .on('close', common.mustCall(() => {
+      assert.strictEqual(errored, true);
     }));
 
   socket.connect(port);
diff --git a/test/parallel/test-tls-wrap-econnreset.js b/test/parallel/test-tls-wrap-econnreset.js
index 5c6db86b75e06a..6ed268e766f704 100644
--- a/test/parallel/test-tls-wrap-econnreset.js
+++ b/test/parallel/test-tls-wrap-econnreset.js
@@ -13,6 +13,7 @@ const server = net.createServer((c) => {
 }).listen(common.mustCall(() => {
   const port = server.address().port;
 
+  let errored = false;
   tls.connect(port, common.localhostIPv4)
     .once('error', common.mustCall((e) => {
       assert.strictEqual(e.code, 'ECONNRESET');
@@ -21,5 +22,9 @@ const server = net.createServer((c) => {
       assert.strictEqual(e.port, port);
       assert.strictEqual(e.localAddress, undefined);
       server.close();
+      errored = true;
+    }))
+    .on('close', common.mustCall(() => {
+      assert.strictEqual(errored, true);
     }));
 }));
diff --git a/test/parallel/test-zlib-write-after-close.js b/test/parallel/test-zlib-write-after-close.js
index d67abee0ac3d58..eb8ff4353963d4 100644
--- a/test/parallel/test-zlib-write-after-close.js
+++ b/test/parallel/test-zlib-write-after-close.js
@@ -26,12 +26,7 @@ const zlib = require('zlib');
 zlib.gzip('hello', common.mustCall(function(err, out) {
   const unzip = zlib.createGunzip();
   unzip.close(common.mustCall());
-  unzip.write(out, common.expectsError({
-    code: 'ERR_STREAM_DESTROYED',
-    name: 'Error',
-    message: 'Cannot call write after a stream was destroyed'
-  }));
-  unzip.on('error', common.expectsError({
+  unzip.write('asd', common.expectsError({
     code: 'ERR_STREAM_DESTROYED',
     name: 'Error',
     message: 'Cannot call write after a stream was destroyed'