From 66fe2d90ffa85e9c7c93426fade005145a073721 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Wed, 11 Mar 2020 13:42:54 +0100
Subject: [PATCH] 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
PR-URL: https://github.com/nodejs/node/pull/32197
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
---
 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());
+  });
+}