Commit 313ecaa
stream: fix broken pipeline error propagation
If the destination was an async function any
error thrown from that function would be swallowed.
Backport-PR-URL: #31975
PR-URL: #31835
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Denys Otrishko <shishugi@gmail.com>1 parent 8ad64b8 commit 313ecaa
3 files changed
Lines changed: 34 additions & 13 deletions
File tree
- lib/internal/streams
- test/parallel
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -163,9 +163,10 @@ function pipeline(...streams) { | |||
| 163 | 163 | } | |
| 164 | 164 | ||
| 165 | 165 | let error; | |
| 166 | + let value; | ||
| 166 | 167 | const destroys = []; | |
| 167 | 168 | ||
| 168 | - function finish(err, val, final) { | ||
| 169 | + function finish(err, final) { | ||
| 169 | 170 | if (!error && err) { | |
| 170 | 171 | error = err; | |
| 171 | 172 | } | |
@@ -177,13 +178,13 @@ function pipeline(...streams) { | |||
| 177 | 178 | } | |
| 178 | 179 | ||
| 179 | 180 | if (final) { | |
| 180 | - callback(error, val); | ||
| 181 | + callback(error, value); | ||
| 181 | 182 | } | |
| 182 | 183 | } | |
| 183 | 184 | ||
| 184 | 185 | function wrap(stream, reading, writing, final) { | |
| 185 | 186 | destroys.push(destroyer(stream, reading, writing, (err) => { | |
| 186 | - finish(err, null, final); | ||
| 187 | + finish(err, final); | ||
| 187 | 188 | })); | |
| 188 | 189 | } | |
| 189 | 190 | ||
@@ -229,11 +230,10 @@ function pipeline(...streams) { | |||
| 229 | 230 | if (isPromise(ret)) { | |
| 230 | 231 | ret | |
| 231 | 232 | .then((val) => { | |
| 233 | + value = val; | ||
| 232 | 234 | pt.end(val); | |
| 233 | - finish(null, val, true); | ||
| 234 | - }) | ||
| 235 | - .catch((err) => { | ||
| 236 | - finish(err, null, true); | ||
| 235 | + }, (err) => { | ||
| 236 | + pt.destroy(err); | ||
| 237 | 237 | }); | |
| 238 | 238 | } else if (isIterable(ret, true)) { | |
| 239 | 239 | pump(ret, pt, finish); | |
@@ -243,7 +243,7 @@ function pipeline(...streams) { | |||
| 243 | 243 | } | |
| 244 | 244 | ||
| 245 | 245 | ret = pt; | |
| 246 | - wrap(ret, true, false, true); | ||
| 246 | + wrap(ret, false, true, true); | ||
| 247 | 247 | } | |
| 248 | 248 | } else if (isStream(stream)) { | |
| 249 | 249 | if (isReadable(ret)) { | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -0,0 +1,25 @@ | |||
| 1 | + 'use strict'; | ||
| 2 | + | ||
| 3 | + const common = require('../common'); | ||
| 4 | + const { | ||
| 5 | + pipeline, | ||
| 6 | + PassThrough | ||
| 7 | + } = require('stream'); | ||
| 8 | + const assert = require('assert'); | ||
| 9 | + | ||
| 10 | + process.on('uncaughtException', common.mustCall((err) => { | ||
| 11 | + assert.strictEqual(err.message, 'error'); | ||
| 12 | + })); | ||
| 13 | + | ||
| 14 | + // Ensure that pipeline that ends with Promise | ||
| 15 | + // still propagates error to uncaughtException. | ||
| 16 | + const s = new PassThrough(); | ||
| 17 | + s.end('data'); | ||
| 18 | + pipeline(s, async function(source) { | ||
| 19 | + for await (const chunk of source) { | ||
| 20 | + chunk; | ||
| 21 | + } | ||
| 22 | + }, common.mustCall((err) => { | ||
| 23 | + assert.ifError(err); | ||
| 24 | + throw new Error('error'); | ||
| 25 | + })); | ||
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -613,11 +613,9 @@ const { promisify } = require('util'); | |||
| 613 | 613 | yield 'hello'; | |
| 614 | 614 | yield 'world'; | |
| 615 | 615 | }, async function*(source) { | |
| 616 | - const ret = []; | ||
| 617 | 616 | for await (const chunk of source) { | |
| 618 | - ret.push(chunk.toUpperCase()); | ||
| 617 | + yield chunk.toUpperCase(); | ||
| 619 | 618 | } | |
| 620 | - yield ret; | ||
| 621 | 619 | }, async function(source) { | |
| 622 | 620 | let ret = ''; | |
| 623 | 621 | for await (const chunk of source) { | |
@@ -754,7 +752,6 @@ const { promisify } = require('util'); | |||
| 754 | 752 | }, common.mustCall((err) => { | |
| 755 | 753 | assert.strictEqual(err, undefined); | |
| 756 | 754 | assert.strictEqual(ret, 'asd'); | |
| 757 | - assert.strictEqual(s.destroyed, true); | ||
| 758 | 755 | })); | |
| 759 | 756 | } | |
| 760 | 757 | ||
@@ -775,7 +772,6 @@ const { promisify } = require('util'); | |||
| 775 | 772 | }, common.mustCall((err) => { | |
| 776 | 773 | assert.strictEqual(err, undefined); | |
| 777 | 774 | assert.strictEqual(ret, 'asd'); | |
| 778 | - assert.strictEqual(s.destroyed, true); | ||
| 779 | 775 | })); | |
| 780 | 776 | } | |
| 781 | 777 | ||
0 commit comments