stream: pipeline should use req.abort() to destroy response

destroy(err) on http response will propagate the error to the
request causing 'error' to be unexpectedly emitted. Furthermore,
response.destroy() unlike request.abort() does not _dump buffered
data.

Fixes a breaking change introduced in 648088289d.

Prefer res.req.abort() over res.destroy() until this situation is
clarified.

Fixes: https://github.com/nodejs/node/issues/31029
Refs: 648088289d

PR-URL: https://github.com/nodejs/node/pull/31054
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Stephen Belanger <admin@stephenbelanger.com>
Reviewed-By: Rich Trott <rtrott@gmail.com>
Reviewed-By: Ruben Bridgewater <ruben@bridgewater.de>

lib/internal/streams/pipeline.js

@@ -17,7 +17,7 @@ const {
 } = require('internal/errors').codes;
 
 function isRequest(stream) {
-  return stream.setHeader && typeof stream.abort === 'function';
+  return stream && stream.setHeader && typeof stream.abort === 'function';
 }
 
 function destroyer(stream, reading, writing, callback) {
@@ -43,22 +43,13 @@ function destroyer(stream, reading, writing, callback) {
 
     // request.destroy just do .end - .abort is what we want
     if (isRequest(stream)) return stream.abort();
-    if (typeof stream.destroy === 'function') {
+    if (isRequest(stream.req)) return stream.req.abort();
-      if (stream.req && stream._writableState === undefined) {
+    if (typeof stream.destroy === 'function') return stream.destroy(err);
-        // This is a ClientRequest
-        // TODO(mcollina): backward compatible fix to avoid crashing.
-        // Possibly remove in a later semver-major change.
-        stream.req.on('error', noop);
-      }
-      return stream.destroy(err);
-    }
 
     callback(err || new ERR_STREAM_DESTROYED('pipe'));
   };
 }
 
-function noop() {}
-
 function pipe(from, to) {
   return from.pipe(to);
 }

test/parallel/test-stream-pipeline.js

@@ -1,7 +1,14 @@
 'use strict';
 
 const common = require('../common');
-const { Stream, Writable, Readable, Transform, pipeline } = require('stream');
+const {
+  Stream,
+  Writable,
+  Readable,
+  Transform,
+  pipeline,
+  PassThrough
+} = require('stream');
 const assert = require('assert');
 const http = require('http');
 const { promisify } = require('util');
@@ -483,3 +490,29 @@ const { promisify } = require('util');
     { code: 'ERR_INVALID_CALLBACK' }
   );
 }
+
+{
+  const server = http.Server(function(req, res) {
+    res.write('asd');
+  });
+  server.listen(0, function() {
+    http.get({ port: this.address().port }, (res) => {
+      const stream = new PassThrough();
+
+      stream.on('error', common.mustCall());
+
+      pipeline(
+        res,
+        stream,
+        common.mustCall((err) => {
+          assert.ok(err);
+          // TODO(ronag):
+          // assert.strictEqual(err.message, 'oh no');
+          server.close();
+        })
+      );
+
+      stream.destroy(new Error('oh no'));
+    }).on('error', common.mustNotCall());
+  });
+}