diff --git a/packages/compiler/src/frontend/lowering/lower-stream.ts b/packages/compiler/src/frontend/lowering/lower-stream.ts index b519ce3780..d3883427aa 100644 --- a/packages/compiler/src/frontend/lowering/lower-stream.ts +++ b/packages/compiler/src/frontend/lowering/lower-stream.ts @@ -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, diff --git a/tests/corpus/7504-pipeline-untyped-source-errors.mjs b/tests/corpus/7504-pipeline-untyped-source-errors.mjs index 3ee984547b..5d6faded18 100644 --- a/tests/corpus/7504-pipeline-untyped-source-errors.mjs +++ b/tests/corpus/7504-pipeline-untyped-source-errors.mjs @@ -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); + } +}