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 6480882.

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

Fixes: #31029
Refs: 6480882

PR-URL: #31054
Reviewed-By: Luigi Pinca <[email protected]>
Reviewed-By: Matteo Collina <[email protected]>
Reviewed-By: Stephen Belanger <[email protected]>
Reviewed-By: Rich Trott <[email protected]>
Reviewed-By: Ruben Bridgewater <[email protected]>

Author: ronag

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 (stream.req && stream._writableState === undefined) {
-        // 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);
-    }
+    if (isRequest(stream.req)) return stream.req.abort();
+    if (typeof stream.destroy === 'function') 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,31 @@ 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();
+
+      // NOTE: 2 because Node 12 streams can emit 'error'
+      // multiple times.
+      stream.on('error', common.mustCall(2));
+
+      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());
+  });
+}