Await the final value of an untyped sync pipeline source

for await over a sync iterator awaits the value of the result that ends
it too, so a sync iterator that returns a rejected promise rejects the
pump. The source's generator skipped that value and reported success.
It now awaits a final value that is an object (a promise included)
before it stops, without closing the finished iterator.
This commit is contained in:
Chris Tate
2026-10-24 17:34:40 -05:00
parent 530bd9534f
commit 6e4bfb3c91
2 changed files with 70 additions and 34 deletions
@@ -3954,6 +3954,38 @@ function lowerDynIterableSource(
const sync = (): IrExpr => varRef(syncId, BOOL, loc);
const step = (): IrExpr => varRef(stepId, DYN, loc);
const value = (): IrExpr => varRef(valueId, DYN, loc);
// for await over a sync iterator awaits each value, the final one
// included (CreateAsyncFromSyncIterator): an object (a promise or a
// thenable) passes on its result
const awaitsValue = (): IrExpr => ({
kind: "logical",
op: "&&",
left: sync(),
right: {
kind: "logical",
op: "||",
left: { kind: "dynTest", test: "promise", value: value(), type: BOOL, loc },
right: {
kind: "logical",
op: "&&",
left: {
kind: "dynTest",
test: "nullish",
negated: true,
value: value(),
type: BOOL,
loc,
},
right: { kind: "dynTest", test: "object", value: value(), type: BOOL, loc },
type: BOOL,
loc,
},
type: BOOL,
loc,
},
type: BOOL,
loc,
});
const get = (obj: IrExpr, name: string): IrExpr => ({
kind: "dynKeyGet",
value: obj,
@@ -4053,46 +4085,28 @@ function lowerDynIterableSource(
type: BOOL,
loc,
},
then: [{ kind: "break", loc }],
// the final value is awaited too: a rejection rejects the
// pump, and the finished iterator is not closed
then: [
{ kind: "assign", localId: valueId, value: get(step(), "value"), loc },
{
kind: "if",
cond: awaitsValue(),
then: [{ kind: "exprStmt", expr: awaitDyn(value()), loc }],
else_: null,
loc,
},
{ kind: "break", loc },
],
else_: null,
loc,
},
setActive(true),
{ kind: "assign", localId: valueId, value: get(step(), "value"), loc },
// for await over a sync iterator awaits each value
// (CreateAsyncFromSyncIterator): a promise or a thenable
// passes on its result, and a rejection closes the iterator
// a rejection here closes the iterator
{
kind: "if",
cond: {
kind: "logical",
op: "&&",
left: sync(),
right: {
kind: "logical",
op: "||",
left: { kind: "dynTest", test: "promise", value: value(), type: BOOL, loc },
right: {
kind: "logical",
op: "&&",
left: {
kind: "dynTest",
test: "nullish",
negated: true,
value: value(),
type: BOOL,
loc,
},
right: { kind: "dynTest", test: "object", value: value(), type: BOOL, loc },
type: BOOL,
loc,
},
type: BOOL,
loc,
},
type: BOOL,
loc,
},
cond: awaitsValue(),
then: [{ kind: "assign", localId: valueId, value: awaitDyn(value()), loc }],
else_: null,
loc,
@@ -1,7 +1,7 @@
// pipeline() over untyped iterable sources: a destination that fails closes
// the source, backpressure waits for drain, an abort signal stops the pump,
// async generator and Map sources reach object-mode destinations, and a
// sync source's promise values are awaited.
// sync source's promise values, its final value included, are awaited.
import { pipeline, Writable, PassThrough } from "node:stream";
import { pipeline as pipelineP } from "node:stream/promises";
@@ -113,3 +113,25 @@ try {
} catch (err) {
console.log("promised rejected", err.message);
}
// a sync source's final value is awaited too: a rejected one rejects the
// pipeline, and a resolved one is not written
function* finalRejects() {
yield "f1";
return Promise.reject(new Error("final value rejected"));
}
function* finalResolves() {
yield "r1";
return Promise.resolve("not written");
}
for (const [label, source] of [["finalRejects", finalRejects], ["finalResolves", finalResolves]]) {
try {
await pipelineP(
pick(source()),
new Writable({ objectMode: true, write(c, _e, cb) { console.log(label, "write", c); cb(); } }),
);
console.log(label, "resolved");
} catch (err) {
console.log(label, "rejected", err.message);
}
}