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

PR-URL: #32197
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>

Author: ronag

lib/internal/streams/pipeline.js

@@ -25,10 +25,32 @@ let EE;
 let PassThrough;
 let createReadableStreamAsyncIterator;
 
+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) {
   const _destroy = once((err) => {
-    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);
@@ -71,10 +93,6 @@ function popCallback(streams) {
   return streams.pop();
 }
 
-function isRequest(stream) {
-  return stream.setHeader && typeof stream.abort === 'function';
-}
-
 function isPromise(obj) {
   return !!(obj && typeof obj.then === 'function');
 }

test/parallel/test-stream-pipeline.js

@@ -1011,3 +1011,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());
+  });
+}