[v13.x backport] pipeline destroy compat

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,22 @@ 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 && reading && !writing && stream.writable) {
+      return callback();
+    }
+
+    if (err || !final || !stream.readable) {
       destroyImpl.destroyer(stream, err);
     }
+
     callback(err);
   });
 

test/parallel/test-stream-pipeline.js

@@ -995,3 +995,40 @@ 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());
+  });
+}
+
+{
+  // 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);
+}