From 5193f4121b8b626ae2ee5f03fc36bea6e54a75be Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Wed, 11 Mar 2020 13:42:54 +0100
Subject: [PATCH 1/2] stream: avoid destroying http1 objects

http1 objects are coupled with their corresponding
res/req and cannot be treated independently as
normal streams. Add a special exception for this
in the pipeline cleanup.

Fixes: https://github.com/nodejs/node/issues/32184

Backport-PR-URL: https://github.com/nodejs/node/pull/32212
---
 lib/internal/streams/pipeline.js      | 28 +++++++++++++++++++++++----
 test/parallel/test-stream-pipeline.js | 21 ++++++++++++++++++++
 2 files changed, 45 insertions(+), 4 deletions(-)

diff --git a/lib/internal/streams/pipeline.js b/lib/internal/streams/pipeline.js
index edbfed3e516c05..7499e37c7b7afb 100644
--- a/lib/internal/streams/pipeline.js
+++ b/lib/internal/streams/pipeline.js
@@ -25,8 +25,20 @@ let EE;
 let PassThrough;
 let createReadableStreamAsyncIterator;
 
-function isRequest(stream) {
-  return stream && stream.setHeader && typeof stream.abort === 'function';
+function isIncoming(stream) {
+  return (
+    stream.socket &&
+    typeof stream.complete === 'boolean' &&
+    ArrayIsArray(stream.rawTrailers) &&
+    ArrayIsArray(stream.rawHeaders)
+  );
+}
+
+function isOutgoing(stream) {
+  return (
+    stream.socket &&
+    typeof stream.setHeader === 'function'
+  );
 }
 
 function destroyer(stream, reading, writing, final, callback) {
@@ -37,10 +49,18 @@ function destroyer(stream, reading, writing, final, callback) {
   eos(stream, { readable: reading, writable: writing }, (err) => {
     if (destroyed) return;
     destroyed = true;
-    const readable = stream.readable || isRequest(stream);
-    if (err || !final || !readable) {
+
+    if (!err && (isIncoming(stream) || isOutgoing(stream))) {
+      // http/1 request objects have a coupling to their response and should
+      // not be prematurely destroyed. Assume they will handle their own
+      // lifecycle.
+      return callback();
+    }
+
+    if (err || !final || !stream.readable) {
       destroyImpl.destroyer(stream, err);
     }
+
     callback(err);
   });
 
diff --git a/test/parallel/test-stream-pipeline.js b/test/parallel/test-stream-pipeline.js
index 9939a16494499c..17532854f974fd 100644
--- a/test/parallel/test-stream-pipeline.js
+++ b/test/parallel/test-stream-pipeline.js
@@ -995,3 +995,24 @@ const { promisify } = require('util');
     assert.strictEqual(res, '');
   }));
 }
+
+{
+  const server = http.createServer((req, res) => {
+    req.socket.on('error', common.mustNotCall());
+    pipeline(req, new PassThrough(), (err) => {
+      assert.ifError(err);
+      res.end();
+      server.close();
+    });
+  });
+
+  server.listen(0, () => {
+    const req = http.request({
+      method: 'PUT',
+      port: server.address().port
+    });
+    req.end('asd123');
+    req.on('response', common.mustCall());
+    req.on('error', common.mustNotCall());
+  });
+}

From ad0098d0c66f6515a92b684463846c50ad97f8fc Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Wed, 11 Mar 2020 13:56:24 +0100
Subject: [PATCH 2/2] stream: avoid destroying writable source

User might still want to be able to use the writable side
of src. This is in the case where e.g. the Duplex input
is not directly connected to its output. Such a case could
happen when the Duplex is reading from a socket and then echos
the data back on the same socket.

Fixes: https://github.com/nodejs/node/commit/4d93e105bfad79ff6c6f01e4b7c2fdd70caeb43b#commitcomment-37751035

Backport-PR-URL: https://github.com/nodejs/node/pull/32212
---
 lib/internal/streams/pipeline.js      |  4 ++++
 test/parallel/test-stream-pipeline.js | 16 ++++++++++++++++
 2 files changed, 20 insertions(+)

diff --git a/lib/internal/streams/pipeline.js b/lib/internal/streams/pipeline.js
index 7499e37c7b7afb..493a464d07b295 100644
--- a/lib/internal/streams/pipeline.js
+++ b/lib/internal/streams/pipeline.js
@@ -57,6 +57,10 @@ function destroyer(stream, reading, writing, final, callback) {
       return callback();
     }
 
+    if (!err && reading && !writing && stream.writable) {
+      return callback();
+    }
+
     if (err || !final || !stream.readable) {
       destroyImpl.destroyer(stream, err);
     }
diff --git a/test/parallel/test-stream-pipeline.js b/test/parallel/test-stream-pipeline.js
index 17532854f974fd..5175b7e07c53c2 100644
--- a/test/parallel/test-stream-pipeline.js
+++ b/test/parallel/test-stream-pipeline.js
@@ -1016,3 +1016,19 @@ const { promisify } = require('util');
     req.on('error', common.mustNotCall());
   });
 }
+
+{
+  // Might still want to be able to use the writable side
+  // of src. This is in the case where e.g. the Duplex input
+  // is not directly connected to its output. Such a case could
+  // happen when the Duplex is reading from a socket and then echos
+  // the data back on the same socket.
+  const src = new PassThrough();
+  assert.strictEqual(src.writable, true);
+  const dst = new PassThrough();
+  pipeline(src, dst, common.mustCall((err) => {
+    assert.strictEqual(src.writable, true);
+    assert.strictEqual(src.destroyed, false);
+  }));
+  src.push(null);
+}