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: #32184

Backport-PR-URL: #32212
PR-URL: #32197
Reviewed-By: Anna Henningsen <[email protected]>
Reviewed-By: Matteo Collina <[email protected]>

Author: ronag

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);
   });
 

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());
+  });
+}