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 https://github.com/nodejs/node/commit/648088289d619bfb149fe90316ce0127083c4c99.

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

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

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>

1 files changed, 34 insertions, 1 deletions

diff --git a/test/parallel/test-stream-pipeline.js b/test/parallel/test-stream-pipeline.js
index 4a41f053bd0..f6ee97ba43d 100644
--- a/test/parallel/test-stream-pipeline.js
+++ b/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());
+  });
+}