From 250bb00621369e1a9f29d94ff1111694499e884e Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Thu, 18 Jul 2019 13:15:42 +0200
Subject: [PATCH 1/5] stream: make finished call the callback if the stream is
 closed

Make stream.finished callback invoked if stream is already
closed/destroyed.
---
 lib/internal/streams/async_iterator.js     |  10 --
 lib/internal/streams/end-of-stream.js      |  40 +++---
 test/parallel/test-http-client-finished.js | 106 ++++++++++++++++
 test/parallel/test-stream-finished.js      | 140 ++++++++++++++++++++-
 4 files changed, 266 insertions(+), 30 deletions(-)

diff --git a/lib/internal/streams/async_iterator.js b/lib/internal/streams/async_iterator.js
index 8755e22de0d00f..39ce3bbedd198e 100644
--- a/lib/internal/streams/async_iterator.js
+++ b/lib/internal/streams/async_iterator.js
@@ -122,16 +122,6 @@ const ReadableStreamAsyncIteratorPrototype = ObjectSetPrototypeOf({
   return() {
     return new Promise((resolve, reject) => {
       const stream = this[kStream];
-
-      // TODO(ronag): Remove this check once finished() handles
-      // already ended and/or destroyed streams.
-      const ended = stream.destroyed || stream.readableEnded ||
-        (stream._readableState && stream._readableState.endEmitted);
-      if (ended) {
-        resolve(createIterResult(undefined, true));
-        return;
-      }
-
       finished(stream, (err) => {
         if (err && err.code !== 'ERR_STREAM_PREMATURE_CLOSE') {
           reject(err);
diff --git a/lib/internal/streams/end-of-stream.js b/lib/internal/streams/end-of-stream.js
index bb01e800becdf3..ca6091fe55fe8a 100644
--- a/lib/internal/streams/end-of-stream.js
+++ b/lib/internal/streams/end-of-stream.js
@@ -28,6 +28,25 @@ function eos(stream, opts, callback) {
 
   callback = once(callback);
 
+  const onerror = (err) => {
+    callback.call(stream, err);
+  };
+
+  let writableFinished = stream.writableFinished ||
+    (stream._writableState && stream._writableState.finished);
+  let readableEnded = stream.readableEnded ||
+    (stream._readableState && stream._readableState.endEmitted);
+
+  if (writableFinished || readableEnded || stream.destroyed ||
+      stream.aborted) {
+    if (opts.error !== false) stream.on('error', onerror);
+    // A destroy(err) call emits error in nextTick.
+    process.nextTick(callback.bind(stream));
+    return () => {
+      stream.removeListener('error', onerror);
+    };
+  }
+
   let readable = opts.readable || (opts.readable !== false && stream.readable);
   let writable = opts.writable || (opts.writable !== false && stream.writable);
 
@@ -35,36 +54,23 @@ function eos(stream, opts, callback) {
     if (!stream.writable) onfinish();
   };
 
-  let writableEnded = stream._writableState && stream._writableState.finished;
   const onfinish = () => {
     writable = false;
-    writableEnded = true;
+    writableFinished = true;
     if (!readable) callback.call(stream);
   };
 
-  let readableEnded = stream.readableEnded ||
-    (stream._readableState && stream._readableState.endEmitted);
   const onend = () => {
     readable = false;
     readableEnded = true;
     if (!writable) callback.call(stream);
   };
 
-  const onerror = (err) => {
-    callback.call(stream, err);
-  };
-
   const onclose = () => {
-    let err;
     if (readable && !readableEnded) {
-      if (!stream._readableState || !stream._readableState.ended)
-        err = new ERR_STREAM_PREMATURE_CLOSE();
-      return callback.call(stream, err);
-    }
-    if (writable && !writableEnded) {
-      if (!stream._writableState || !stream._writableState.ended)
-        err = new ERR_STREAM_PREMATURE_CLOSE();
-      return callback.call(stream, err);
+      callback.call(stream, new ERR_STREAM_PREMATURE_CLOSE());
+    } else if (writable && !writableFinished) {
+      callback.call(stream, new ERR_STREAM_PREMATURE_CLOSE());
     }
   };
 
diff --git a/test/parallel/test-http-client-finished.js b/test/parallel/test-http-client-finished.js
index 2d7e5b95b3ca33..337f7b596d7442 100644
--- a/test/parallel/test-http-client-finished.js
+++ b/test/parallel/test-http-client-finished.js
@@ -25,3 +25,109 @@ const { finished } = require('stream');
     .end();
   }));
 }
+
+{
+  // Test abort before finished.
+
+  const server = http.createServer(function(req, res) {
+  });
+
+  server.listen(0, common.mustCall(function() {
+    const req = http.request({
+      port: this.address().port
+    }, common.mustNotCall());
+    req.abort();
+    finished(req, common.mustCall(() => {
+      server.close();
+    }));
+  }));
+}
+
+{
+  // Test abort after request.
+
+  const server = http.createServer(function(req, res) {
+  });
+
+  server.listen(0, common.mustCall(function() {
+    const req = http.request({
+      port: this.address().port
+    }).end();
+    finished(req, (err) => {
+      common.expectsError({
+        type: Error,
+        code: 'ERR_STREAM_PREMATURE_CLOSE'
+      })(err);
+      finished(req, common.mustCall(() => {
+        server.close();
+      }));
+    });
+    req.abort();
+  }));
+}
+
+{
+  // Test abort before end.
+
+  const server = http.createServer(function(req, res) {
+    res.write('test');
+  });
+
+  server.listen(0, common.mustCall(function() {
+    const req = http.request({
+      port: this.address().port
+    }).on('response', common.mustCall((res) => {
+      req.abort();
+      finished(res, common.mustCall(() => {
+        finished(res, common.mustCall(() => {
+          server.close();
+        }));
+      }));
+    })).end();
+  }));
+}
+
+{
+  // Test destroy before end.
+
+  const server = http.createServer(function(req, res) {
+    res.write('test');
+  });
+
+  server.listen(0, common.mustCall(function() {
+    http.request({
+      port: this.address().port
+    }).on('response', common.mustCall((res) => {
+      // TODO(ronag): Bug? Won't emit 'close' unless read.
+      res.on('data', () => {});
+      res.destroy();
+      finished(res, common.mustCall(() => {
+        finished(res, common.mustCall(() => {
+          server.close();
+        }));
+      }));
+    })).end();
+  }));
+}
+
+{
+  // Test finish after end.
+
+  const server = http.createServer(function(req, res) {
+    res.end('asd');
+  });
+
+  server.listen(0, common.mustCall(function() {
+    http.request({
+      port: this.address().port
+    }).on('response', common.mustCall((res) => {
+      // TODO(ronag): Bug? Won't emit 'close' unless read.
+      res.on('data', () => {});
+      finished(res, common.mustCall(() => {
+        finished(res, common.mustCall(() => {
+          server.close();
+        }));
+      }));
+    })).end();
+  }));
+}
diff --git a/test/parallel/test-stream-finished.js b/test/parallel/test-stream-finished.js
index d6361ea303635d..c5e792b45e2e4b 100644
--- a/test/parallel/test-stream-finished.js
+++ b/test/parallel/test-stream-finished.js
@@ -97,6 +97,18 @@ const { promisify } = require('util');
   }));
 }
 
+{
+  const rs = new Readable();
+
+  finished(rs, common.mustCall((err) => {
+    assert(err, 'premature close error');
+  }));
+
+  rs.push(null);
+  rs.emit('close');
+  rs.resume();
+}
+
 {
   const rs = new Readable();
 
@@ -105,7 +117,9 @@ const { promisify } = require('util');
   }));
 
   rs.push(null);
-  rs.emit('close'); // Should not trigger an error
+  rs.on('end', common.mustCall(() => {
+    rs.emit('close'); // Should not trigger an error
+  }));
   rs.resume();
 }
 
@@ -155,8 +169,9 @@ const { promisify } = require('util');
   rs.resume();
 }
 
-// Test that calling returned function removes listeners
 {
+  // Nothing happens if disposed.
+
   const ws = new Writable({
     write(data, env, cb) {
       cb();
@@ -168,6 +183,8 @@ const { promisify } = require('util');
 }
 
 {
+  // Nothing happens if disposed.
+
   const rs = new Readable();
   const removeListeners = finished(rs, common.mustNotCall());
   removeListeners();
@@ -178,9 +195,126 @@ const { promisify } = require('util');
 }
 
 {
+  // Completed if readable-like is ended before.
+
   const streamLike = new EE();
   streamLike.readableEnded = true;
   streamLike.readable = true;
-  finished(streamLike, common.mustCall);
+  finished(streamLike, common.mustCall());
+}
+
+{
+  // Completed if readable-like is never ended.
+
+  const streamLike = new EE();
+  streamLike.readableEnded = false;
+  streamLike.readable = true;
+  finished(streamLike, common.expectsError({
+    code: 'ERR_STREAM_PREMATURE_CLOSE'
+  }));
+  streamLike.emit('close');
+}
+
+{
+  // Completed if writable-like is destroyed before.
+
+  const streamLike = new EE();
+  streamLike.destroyed = true;
+  streamLike.writable = true;
+  finished(streamLike, common.mustCall());
+}
+
+{
+  // Completed if readable-like is aborted before.
+
+  const streamLike = new EE();
+  streamLike.destroyed = true;
+  streamLike.readable = true;
+  finished(streamLike, common.mustCall());
+}
+
+{
+  // Completed if writable-like is aborted before.
+
+  const streamLike = new EE();
+  streamLike.aborted = true;
+  streamLike.writable = true;
+  finished(streamLike, common.mustCall());
+}
+
+{
+  // Completed if readable-like is aborted before.
+
+  const streamLike = new EE();
+  streamLike.aborted = true;
+  streamLike.readable = true;
+  finished(streamLike, common.mustCall());
+}
+
+{
+  // Completed if streamlike is finished before.
+
+  const streamLike = new EE();
+  streamLike.writableFinished = true;
+  streamLike.writable = true;
+  finished(streamLike, common.mustCall());
+}
+
+{
+  // Premature close if stream is not finished.
+
+  const streamLike = new EE();
+  streamLike.writableFinished = false;
+  streamLike.writable = true;
+  finished(streamLike, common.expectsError({
+    code: 'ERR_STREAM_PREMATURE_CLOSE'
+  }));
+  streamLike.emit('close');
+}
+
+{
+  // Premature close if stream never emitted 'finish'
+  // even if writableFinished says something else.
+
+  const streamLike = new EE();
+  streamLike.writable = true;
+  finished(streamLike, common.expectsError({
+    code: 'ERR_STREAM_PREMATURE_CLOSE'
+  }));
+  streamLike.writableFinished = true;
+  streamLike.emit('close');
+}
+
+
+{
+  // Premature close if stream never emitted 'end'
+  // even if readableEnded says something else.
+
+  const streamLike = new EE();
+  streamLike.readable = true;
+  finished(streamLike, common.expectsError({
+    code: 'ERR_STREAM_PREMATURE_CLOSE'
+  }));
+  streamLike.readableEnded = true;
   streamLike.emit('close');
 }
+
+{
+  // Completes if already finished.
+
+  const w = new Writable();
+  finished(w, common.mustCall(() => {
+    finished(w, common.mustCall());
+  }));
+  w.destroy();
+}
+
+{
+  // Completes if already ended.
+
+  const r = new Readable();
+  finished(r, common.mustCall(() => {
+    finished(r, common.mustCall());
+  }));
+  r.destroy();
+}

From 4eebd396c4a06dd42476b89039d12b8de6a9f486 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Thu, 10 Oct 2019 22:27:54 +0200
Subject: [PATCH 2/5] fixup: isReadable & isWritable

---
 lib/internal/streams/end-of-stream.js | 18 ++++++++++++++++--
 1 file changed, 16 insertions(+), 2 deletions(-)

diff --git a/lib/internal/streams/end-of-stream.js b/lib/internal/streams/end-of-stream.js
index ca6091fe55fe8a..cc9789ddf0bb98 100644
--- a/lib/internal/streams/end-of-stream.js
+++ b/lib/internal/streams/end-of-stream.js
@@ -13,6 +13,18 @@ function isRequest(stream) {
   return stream.setHeader && typeof stream.abort === 'function';
 }
 
+function isReadable(stream) {
+  return typeof stream.readable === 'boolean' ||
+    typeof stream.readableEnded === 'boolean' ||
+    !!stream._readableState;
+}
+
+function isWritable(stream) {
+  return typeof stream.writable === 'boolean' ||
+    typeof stream.writableEnded === 'boolean' ||
+    !!stream._writableState;
+}
+
 function eos(stream, opts, callback) {
   if (arguments.length === 2) {
     callback = opts;
@@ -47,8 +59,10 @@ function eos(stream, opts, callback) {
     };
   }
 
-  let readable = opts.readable || (opts.readable !== false && stream.readable);
-  let writable = opts.writable || (opts.writable !== false && stream.writable);
+  const readable = opts.readable ||
+    (opts.readable !== false && isReadable(stream));
+  const writable = opts.writable ||
+    (opts.writable !== false && isWritable(stream));
 
   const onlegacyfinish = () => {
     if (!stream.writable) onfinish();

From ba74fd84933ad3dbc7397522231d6dbf81468fc8 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Mon, 23 Sep 2019 09:36:04 +0200
Subject: [PATCH 3/5] stream: remove ambiguous code

PR-URL: https://github.com/nodejs/node/pull/29664
Reviewed-By: Minwoo Jung <minwoo@nodesource.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
---
 lib/internal/streams/end-of-stream.js | 6 ++----
 1 file changed, 2 insertions(+), 4 deletions(-)

diff --git a/lib/internal/streams/end-of-stream.js b/lib/internal/streams/end-of-stream.js
index cc9789ddf0bb98..585a52ae2f2382 100644
--- a/lib/internal/streams/end-of-stream.js
+++ b/lib/internal/streams/end-of-stream.js
@@ -69,15 +69,13 @@ function eos(stream, opts, callback) {
   };
 
   const onfinish = () => {
-    writable = false;
     writableFinished = true;
-    if (!readable) callback.call(stream);
+    if (!readable || readableEnded) callback.call(stream);
   };
 
   const onend = () => {
-    readable = false;
     readableEnded = true;
-    if (!writable) callback.call(stream);
+    if (!writable || writableFinished) callback.call(stream);
   };
 
   const onclose = () => {

From fa1239e028e3cea8c4146dadb6e1d068cb407a85 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Thu, 26 Sep 2019 20:48:37 +0200
Subject: [PATCH 4/5] stream: eof & pipeline compat

Some broken streams might emit 'close' before 'end' or 'finish'.
However, if they have actually been ended or finished, this
should not result in a premature close error.

Fixes: https://github.com/nodejs/node/issues/29699
---
 doc/api/stream.md                          | 10 +++++++
 lib/internal/streams/end-of-stream.js      | 22 ++++++++++++---
 lib/internal/streams/pipeline.js           | 23 ++++++++++++++++
 test/parallel/test-http-client-finished.js |  2 +-
 test/parallel/test-stream-finished.js      | 17 ++++++------
 test/parallel/test-stream-pipeline.js      | 31 ++++++++++++++++++++++
 6 files changed, 91 insertions(+), 14 deletions(-)

diff --git a/doc/api/stream.md b/doc/api/stream.md
index cf03163218bfbc..b2f121f62604e0 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -1525,6 +1525,11 @@ Especially useful in error handling scenarios where a stream is destroyed
 prematurely (like an aborted HTTP request), and will not emit `'end'`
 or `'finish'`.
 
+`stream.finished()` will error with `ERR_STREAM_PREMATURE_CLOSE` if:
+* `Writable` emits `'close'` before `'finish'` and
+[`writable.writableFinished`][].
+* `Readable` emits `'close'` before `'end'` and [`readable.readableEnded`][].
+
 The `finished` API is promisify-able as well;
 
 ```js
@@ -1647,6 +1652,10 @@ run().catch(console.error);
 * `Readable` streams which have emitted `'end'` or `'close'`.
 * `Writable` streams which have emitted `'finish'` or `'close'`.
 
+If any `Writable` or `Readable` stream emits `'close'` without being able to
+fully flush or drain, `stream.pipeline()` will error with
+`ERR_STREAM_PREMATURE_CLOSE`.
+
 `stream.pipeline()` leaves dangling event listeners on the streams
 after the `callback` has been invoked. In the case of reuse of streams after
 failure, this can cause event listener leaks and swallowed errors.
@@ -2865,6 +2874,7 @@ contain multi-byte characters.
 [`process.stdout`]: process.html#process_process_stdout
 [`readable._read()`]: #stream_readable_read_size_1
 [`readable.push('')`]: #stream_readable_push
+[`readable.readableEnded`]: #stream_readable_readableended
 [`readable.setEncoding()`]: #stream_readable_setencoding_encoding
 [`stream.Readable.from()`]: #stream_stream_readable_from_iterable_options
 [`stream.cork()`]: #stream_writable_cork
diff --git a/lib/internal/streams/end-of-stream.js b/lib/internal/streams/end-of-stream.js
index 585a52ae2f2382..dd42eb9f5b3f46 100644
--- a/lib/internal/streams/end-of-stream.js
+++ b/lib/internal/streams/end-of-stream.js
@@ -79,10 +79,24 @@ function eos(stream, opts, callback) {
   };
 
   const onclose = () => {
-    if (readable && !readableEnded) {
-      callback.call(stream, new ERR_STREAM_PREMATURE_CLOSE());
-    } else if (writable && !writableFinished) {
-      callback.call(stream, new ERR_STREAM_PREMATURE_CLOSE());
+    if (readable) {
+      const ended = (stream._readableState &&
+        stream._readableState.endEmitted) || stream.readableEnded;
+      if (!ended && !readableEnded) {
+        callback.call(stream, new ERR_STREAM_PREMATURE_CLOSE());
+      } else if (!readableEnded) {
+        // Compat. Stream has ended but haven't emitted 'end'.
+        callback.call(stream);
+      }
+    } else if (writable) {
+      const finished = (stream._writableState &&
+        stream._writableState.finished) || stream.writableFinished;
+      if (!finished && !writableFinished) {
+        callback.call(stream, new ERR_STREAM_PREMATURE_CLOSE());
+      } else if (!writableFinished) {
+        // Compat. Stream has finished but haven't emitted 'finish'.
+        callback.call(stream);
+      }
     }
   };
 
diff --git a/lib/internal/streams/pipeline.js b/lib/internal/streams/pipeline.js
index e0834171bfb8fc..9a3521f523b54e 100644
--- a/lib/internal/streams/pipeline.js
+++ b/lib/internal/streams/pipeline.js
@@ -46,6 +46,29 @@ function destroyer(stream, reading, writing, callback) {
 
   if (eos === undefined) eos = require('internal/streams/end-of-stream');
   eos(stream, { readable: reading, writable: writing }, (err) => {
+    if (
+      err &&
+      err.code === 'ERR_STREAM_PREMATURE_CLOSE' &&
+      reading &&
+      (stream._readableState && stream._readableState.ended)
+    ) {
+      // Some readable streams will emit 'close' before 'end'. However, since
+      // this is on the readable side 'end' should still be emitted if the
+      // stream has been ended and no error emitted. This should be allowed in
+      // favor of backwards compatibility. Since the stream is piped to a
+      // destination this should not result in any observable difference.
+      // We don't need to check if this is a writable premature close since
+      // eos will only fail with premature close on the reading side for
+      // duplex streams.
+      stream
+        .on('end', () => {
+          closed = true;
+          callback();
+        })
+        .on('error', callback);
+      return;
+    }
+
     if (err) return callback(err);
     closed = true;
     callback();
diff --git a/test/parallel/test-http-client-finished.js b/test/parallel/test-http-client-finished.js
index 337f7b596d7442..8eddf8660b4476 100644
--- a/test/parallel/test-http-client-finished.js
+++ b/test/parallel/test-http-client-finished.js
@@ -55,7 +55,7 @@ const { finished } = require('stream');
     }).end();
     finished(req, (err) => {
       common.expectsError({
-        type: Error,
+        name: 'Error',
         code: 'ERR_STREAM_PREMATURE_CLOSE'
       })(err);
       finished(req, common.mustCall(() => {
diff --git a/test/parallel/test-stream-finished.js b/test/parallel/test-stream-finished.js
index c5e792b45e2e4b..caca55601868a2 100644
--- a/test/parallel/test-stream-finished.js
+++ b/test/parallel/test-stream-finished.js
@@ -273,27 +273,26 @@ const { promisify } = require('util');
 }
 
 {
-  // Premature close if stream never emitted 'finish'
-  // even if writableFinished says something else.
+  // No error if stream never emitted 'end'
+  // even if readableEnded says something else.
 
   const streamLike = new EE();
   streamLike.writable = true;
-  finished(streamLike, common.expectsError({
-    code: 'ERR_STREAM_PREMATURE_CLOSE'
+  finished(streamLike, common.mustCall((err) => {
+    assert.strictEqual(err, undefined);
   }));
   streamLike.writableFinished = true;
   streamLike.emit('close');
 }
 
-
 {
-  // Premature close if stream never emitted 'end'
-  // even if readableEnded says something else.
+  // No error if stream never emitted 'finished'
+  // even if writableFinished says something else.
 
   const streamLike = new EE();
   streamLike.readable = true;
-  finished(streamLike, common.expectsError({
-    code: 'ERR_STREAM_PREMATURE_CLOSE'
+  finished(streamLike, common.mustCall((err) => {
+    assert.strictEqual(err, undefined);
   }));
   streamLike.readableEnded = true;
   streamLike.emit('close');
diff --git a/test/parallel/test-stream-pipeline.js b/test/parallel/test-stream-pipeline.js
index 19fc246e2bf3cd..99876738415c69 100644
--- a/test/parallel/test-stream-pipeline.js
+++ b/test/parallel/test-stream-pipeline.js
@@ -912,4 +912,35 @@ const { promisify } = require('util');
   }, common.mustCall((err) => {
     assert.strictEqual(err.message, 'kaboom');
   }));
+  // Make sure 'close' before 'end' finishes without error
+  // if readable has received eof.
+  // Ref: https://github.com/nodejs/node/issues/29699
+  const r = new Readable();
+  const w = new Writable({
+    write(chunk, encoding, cb) {
+      cb();
+    }
+  });
+  pipeline(r, w, (err) => {
+    assert.strictEqual(err, undefined);
+  });
+  r.push(null);
+  r.destroy();
+}
+
+{
+  // Make sure 'close' before 'end' finishes without error
+  // if readable has received eof.
+  // Ref: https://github.com/nodejs/node/issues/29699
+  const r = new Readable();
+  const w = new Writable({
+    write(chunk, encoding, cb) {
+      cb();
+    }
+  });
+  pipeline(r, w, (err) => {
+    assert.strictEqual(err, undefined);
+  });
+  r.push(null);
+  r.emit('close');
 }

From f730a54a981897b47e8e1089b26bfcc71d7dd442 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Sat, 18 Jan 2020 18:38:45 +0100
Subject: [PATCH 5/5] fixup

---
 lib/_stream_readable.js               |  2 ++
 lib/_stream_writable.js               |  2 ++
 lib/internal/streams/destroy.js       |  9 ++++++
 lib/internal/streams/end-of-stream.js | 23 ++++++++++-----
 test/parallel/test-stream-finished.js | 42 ++++-----------------------
 5 files changed, 34 insertions(+), 44 deletions(-)

diff --git a/lib/_stream_readable.js b/lib/_stream_readable.js
index 6fa3540057be33..73d18b19bb53dc 100644
--- a/lib/_stream_readable.js
+++ b/lib/_stream_readable.js
@@ -147,6 +147,8 @@ function ReadableState(options, stream, isDuplex) {
   // Indicates whether the stream has errored.
   this.errored = false;
 
+  this.closed = false;
+
   // Crypto is kind of old and crusty.  Historically, its default string
   // encoding is 'binary' so we have to make this configurable.
   // Everything else in the universe uses 'utf8', though.
diff --git a/lib/_stream_writable.js b/lib/_stream_writable.js
index 3f029883705a0a..6d65abebcb030c 100644
--- a/lib/_stream_writable.js
+++ b/lib/_stream_writable.js
@@ -175,6 +175,8 @@ function WritableState(options, stream, isDuplex) {
   // is disabled we need a way to tell whether the stream has failed.
   this.errored = false;
 
+  this.closed = false;
+
   // Count buffered requests
   this.bufferedRequestCount = 0;
 
diff --git a/lib/internal/streams/destroy.js b/lib/internal/streams/destroy.js
index b80fb56a6bf324..3115202ca6bdee 100644
--- a/lib/internal/streams/destroy.js
+++ b/lib/internal/streams/destroy.js
@@ -73,6 +73,13 @@ function emitCloseNT(self) {
   const r = self._readableState;
   const w = self._writableState;
 
+  if (w) {
+    w.closed = true;
+  }
+  if (r) {
+    r.closed = true;
+  }
+
   if ((w && w.emitClose) || (r && r.emitClose)) {
     self.emit('close');
   }
@@ -103,6 +110,7 @@ function undestroy() {
   if (r) {
     r.destroyed = false;
     r.errored = false;
+    r.closed = false;
     r.reading = false;
     r.ended = false;
     r.endEmitted = false;
@@ -112,6 +120,7 @@ function undestroy() {
   if (w) {
     w.destroyed = false;
     w.errored = false;
+    w.closed = false;
     w.ended = false;
     w.ending = false;
     w.finalCalled = false;
diff --git a/lib/internal/streams/end-of-stream.js b/lib/internal/streams/end-of-stream.js
index dd42eb9f5b3f46..0388533dac0167 100644
--- a/lib/internal/streams/end-of-stream.js
+++ b/lib/internal/streams/end-of-stream.js
@@ -44,16 +44,25 @@ function eos(stream, opts, callback) {
     callback.call(stream, err);
   };
 
-  let writableFinished = stream.writableFinished ||
-    (stream._writableState && stream._writableState.finished);
-  let readableEnded = stream.readableEnded ||
-    (stream._readableState && stream._readableState.endEmitted);
+  const w = stream._writableState;
+  const r = stream._readableState;
 
-  if (writableFinished || readableEnded || stream.destroyed ||
-      stream.aborted) {
+  let writableFinished = stream.writableFinished || (w && w.finished);
+  let readableEnded = stream.readableEnded || (r && r.endEmitted);
+
+  const errorEmitted = (w && w.errorEmitted) || (r && r.errorEmitted);
+  const closed = (w && w.closed) || (r && r.closed);
+
+  if (writableFinished || readableEnded || errorEmitted || closed) {
+    // TODO(ronag): rethrow if errorEmitted?
+    // TODO(ronag): premature close if closed but not
+    // errored, finished or ended?
+
+    // Swallow any error past this point.
     if (opts.error !== false) stream.on('error', onerror);
-    // A destroy(err) call emits error in nextTick.
+
     process.nextTick(callback.bind(stream));
+
     return () => {
       stream.removeListener('error', onerror);
     };
diff --git a/test/parallel/test-stream-finished.js b/test/parallel/test-stream-finished.js
index caca55601868a2..b6db7a867606a4 100644
--- a/test/parallel/test-stream-finished.js
+++ b/test/parallel/test-stream-finished.js
@@ -197,10 +197,14 @@ const { promisify } = require('util');
 {
   // Completed if readable-like is ended before.
 
+  let ticked = false;
   const streamLike = new EE();
   streamLike.readableEnded = true;
   streamLike.readable = true;
-  finished(streamLike, common.mustCall());
+  finished(streamLike, common.mustCall(() => {
+    assert.strictEqual(ticked, true);
+  }));
+  ticked = true;
 }
 
 {
@@ -215,42 +219,6 @@ const { promisify } = require('util');
   streamLike.emit('close');
 }
 
-{
-  // Completed if writable-like is destroyed before.
-
-  const streamLike = new EE();
-  streamLike.destroyed = true;
-  streamLike.writable = true;
-  finished(streamLike, common.mustCall());
-}
-
-{
-  // Completed if readable-like is aborted before.
-
-  const streamLike = new EE();
-  streamLike.destroyed = true;
-  streamLike.readable = true;
-  finished(streamLike, common.mustCall());
-}
-
-{
-  // Completed if writable-like is aborted before.
-
-  const streamLike = new EE();
-  streamLike.aborted = true;
-  streamLike.writable = true;
-  finished(streamLike, common.mustCall());
-}
-
-{
-  // Completed if readable-like is aborted before.
-
-  const streamLike = new EE();
-  streamLike.aborted = true;
-  streamLike.readable = true;
-  finished(streamLike, common.mustCall());
-}
-
 {
   // Completed if streamlike is finished before.