stream: flush each fused stateless transform

Ensure consecutive stateless stream/iter transforms each receive a
final null flush after upstream flush output has been processed.
Fixes: #63467
Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com>
Assisted-by: openai:gpt-5.5
PR-URL: #63468Fixes: #63467
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>

Author: trivikr

lib/internal/streams/iter/pull.js

@@ -273,6 +273,17 @@ function* processTransformResultSync(result) {
 result);
 }
 
+/**
+ * Append normalized transform result batches to an array (sync).
+ * @param {Array<Uint8Array[]>} target
+ * @param {*} result
+ */
+functionappendTransformResultSync(target,result){
+for(constbatchofprocessTransformResultSync(result)){
+ArrayPrototypePush(target,batch);
+}
+}
+
 /**
  * Process transform result (async).
  * @yields {Uint8Array[]}
@@ -356,6 +367,18 @@ async function* processTransformResultAsync(result) {
 result);
 }
 
+/**
+ * Append normalized transform result batches to an array (async).
+ * @param {Array<Uint8Array[]>} target
+ * @param {*} result
+ * @returns {Promise<void>}
+ */
+asyncfunctionappendTransformResultAsync(target,result){
+forawait(constbatchofprocessTransformResultAsync(result)){
+ArrayPrototypePush(target,batch);
+}
+}
+
 // =============================================================================
 // Sync Pipeline Implementation
 // =============================================================================
@@ -398,18 +421,19 @@ function* applyFusedStatelessSyncTransforms(source, run) {
 yield*processTransformResultSync(current);
 }
 }
-// Flush
-letcurrent=null;
+// Flush each transform after all upstream data, including data emitted by
+// earlier flushes, has been processed by that transform.
+letpending=[];
 for(leti=0;i<run.length;i++){
-constresult=run[i](current);
-if(result===null){
-current=null;
-continue;
+constnext=[];
+for(letj=0;j<pending.length;j++){
+appendTransformResultSync(next,run[i](pending[j]));
 }
-current=result;
+appendTransformResultSync(next,run[i](null));
+pending=next;
 }
-if(current!=null){
-yield*processTransformResultSync(current);
+for(leti=0;i<pending.length;i++){
+yieldpending[i];
 }
 }
 
@@ -522,30 +546,23 @@ async function* applyFusedStatelessAsyncTransforms(source, run, signal) {
 yield*processTransformResultAsync(current);
 }
 }
-// Flush: send null through each transform in order
-letcurrent=null;
+// Flush each transform after all upstream data, including data emitted by
+// earlier flushes, has been processed by that transform.
+letpending=[];
 for(leti=0;i<run.length;i++){
-constresult=run[i](current,{__proto__: null, signal });
-if(result===null){
-current=null;
-continue;
-}
-if(isPromise(result)){
-current=awaitresult;
-}else{
-current=result;
+constnext=[];
+for(letj=0;j<pending.length;j++){
+awaitappendTransformResultAsync(
+next,
+run[i](pending[j],{__proto__: null, signal }));
 }
+awaitappendTransformResultAsync(
+next,
+run[i](null,{__proto__: null, signal }));
+pending=next;
 }
-if(current!==null){
-if(isUint8ArrayBatch(current)){
-if(current.length>0)yieldcurrent;
-}elseif(isUint8Array(current)){
-yield[current];
-}elseif(typeofcurrent==='string'){
-yield[toUint8Array(current)];
-}else{
-yield*processTransformResultAsync(current);
-}
+for(leti=0;i<pending.length;i++){
+yieldpending[i];
 }
 }
 

test/parallel/test-stream-iter-pull-async.js

@@ -233,6 +233,19 @@ async function testPullStatelessTransformFlush() {
 assert.strictEqual(data,'data-TRAILER');
 }
 
+// Consecutive stateless transforms each receive a final flush signal after
+// upstream flush output has been processed.
+asyncfunctiontestPullConsecutiveStatelessTransformFlush(){
+constenc=newTextEncoder();
+constaddAOnFlush=(chunks)=>(chunks===null ?
+[enc.encode('-A')] : chunks);
+constaddBOnFlush=(chunks)=>(chunks===null ?
+[enc.encode('-B')] : chunks);
+
+constdata=awaittext(pull(from('x'),addAOnFlush,addBOnFlush));
+assert.strictEqual(data,'x-A-B');
+}
+
 // Stateless transform flush error propagates
 asyncfunctiontestPullStatelessTransformFlushError(){
 constbadFlush=(chunks)=>{
@@ -357,6 +370,7 @@ async function testTransformOptionsNotShared() {
 testPullStatelessTransformError(),
 testPullStatefulTransformError(),
 testPullStatelessTransformFlush(),
+testPullConsecutiveStatelessTransformFlush(),
 testPullStatelessTransformFlushError(),
 testPullWithSyncSource(),
 testPullStringSource(),

test/parallel/test-stream-iter-pull-sync.js

@@ -127,6 +127,20 @@ function testPullSyncStatelessTransformFlush() {
 assert.strictEqual(data,'data-TRAILER');
 }
 
+// Consecutive stateless transforms each receive a final flush signal after
+// upstream flush output has been processed.
+functiontestPullSyncConsecutiveStatelessTransformFlush(){
+constenc=newTextEncoder();
+constaddAOnFlush=(chunks)=>(chunks===null ?
+[enc.encode('-A')] : chunks);
+constaddBOnFlush=(chunks)=>(chunks===null ?
+[enc.encode('-B')] : chunks);
+
+constdata=newTextDecoder().decode(bytesSync(
+pullSync(fromSync('x'),addAOnFlush,addBOnFlush)));
+assert.strictEqual(data,'x-A-B');
+}
+
 // Stateless transform flush error propagates
 functiontestPullSyncStatelessTransformFlushError(){
 constbadFlush=(chunks)=>{
@@ -173,6 +187,7 @@ Promise.all([
 testPullSyncStatelessTransformError(),
 testPullSyncStatefulTransformError(),
 testPullSyncStatelessTransformFlush(),
+testPullSyncConsecutiveStatelessTransformFlush(),
 testPullSyncStatelessTransformFlushError(),
 testPullSyncInvalidTransform(),
 ]).then(common.mustCall());