stream: fixup error handling in stream/iter adapters

jasnell · jasnell · commit bea41b144a4b · 2026-06-01T09:22:12.000-07:00
Signed-off-by: James M Snell &lt;jasnell@gmail.com&gt;
diff --git a/lib/internal/streams/iter/classic.js b/lib/internal/streams/iter/classic.js
@@ -779,8 +779,10 @@ function toWritable(writer) {
           queueMicrotask(cb);
           return;
         }
-      } catch {
-        // Sync path threw -- fall through to async.
+        // WriteSync returned false: not accepted, fall through to async.
+      } catch (err) {
+        cb(err);
+        return;
       }
     }
     try {
@@ -803,8 +805,10 @@ function toWritable(writer) {
           queueMicrotask(cb);
           return;
         }
-      } catch {
-        // Sync path threw -- fall through to async.
+        // WritevSync returned false: not accepted, fall through to async.
+      } catch (err) {
+        cb(err);
+        return;
       }
     }
     try {
@@ -826,8 +830,10 @@ function toWritable(writer) {
           queueMicrotask(cb);
           return;
         }
-      } catch {
-        // Sync path threw -- fall through to async.
+        // Result < 0: can't end synchronously, fall through to async.
+      } catch (err) {
+        cb(err);
+        return;
       }
     }
     try {
diff --git a/lib/internal/streams/iter/consumers.js b/lib/internal/streams/iter/consumers.js
@@ -480,6 +480,7 @@ function merge(...args) {
         );
       }
 
+      let primaryError;
       try {
         while (activeCount > 0 || ready.length > 0) {
           signal?.throwIfAborted();
@@ -500,17 +501,42 @@ function merge(...args) {
             });
           }
         }
+      } catch (err) {
+        primaryError = err;
       } finally {
-        // Clean up: return all iterators
+        // Clean up: return all iterators. Cleanup errors are not
+        // swallowed - a broken iterator.return() (e.g., failing to
+        // release a resource) should be visible to the caller.
+        let cleanupError;
         await SafePromiseAllReturnVoid(iterators, async (iterator) => {
           if (iterator.return) {
             try {
               await iterator.return();
-            } catch {
-              // Ignore return errors
+            } catch (err) {
+              // Keep the first cleanup error encountered.
+              cleanupError ??= err;
             }
           }
         });
+        // Throws in finally are intentional: in an async generator,
+        // cleanup must run on consumer break (generator return), and
+        // code after the try/finally block would not execute.
+        if (cleanupError !== undefined) {
+          if (primaryError !== undefined) {
+            // Both a primary error and a cleanup error occurred.
+            // Wrap in SuppressedError so neither is lost:
+            // .error = primaryError, .suppressed = cleanupError.
+            // eslint-disable-next-line no-restricted-syntax, no-unsafe-finally
+            throw new SuppressedError(cleanupError, primaryError);
+          }
+          // No primary error - the cleanup error is the only error.
+          // eslint-disable-next-line no-unsafe-finally
+          throw cleanupError;
+        }
+        if (primaryError !== undefined) {
+          // eslint-disable-next-line no-unsafe-finally
+          throw primaryError;
+        }
       }
     },
   };
diff --git a/test/parallel/test-stream-iter-writable-from.js b/test/parallel/test-stream-iter-writable-from.js
@@ -537,18 +537,15 @@ function testHighWaterMarkIsMaxSafeInt() {
 }
 
 // =============================================================================
-// writeSync throws -- falls back to async
+// writeSync throws -- error propagates, does NOT fall back to async
 // =============================================================================
 
-async function testWriteSyncThrowsFallback() {
-  let asyncCalled = false;
-
+async function testWriteSyncThrowsPropagation() {
   const writer = {
     writeSync() {
       throw new Error('sync broken');
     },
-    write(chunk) {
-      asyncCalled = true;
+    write() {
       return Promise.resolve();
     },
     end() { return Promise.resolve(0); },
@@ -557,11 +554,12 @@ async function testWriteSyncThrowsFallback() {
 
   const writable = toWritable(writer);
 
-  await new Promise((resolve) => {
-    writable.write('test', resolve);
-  });
-
-  assert.ok(asyncCalled, 'async write should be called as fallback');
+  await assert.rejects(new Promise((resolve, reject) => {
+    writable.write('test', (err) => {
+      if (err) reject(err);
+      else resolve();
+    });
+  }), { message: 'sync broken' });
 }
 
 // =============================================================================
@@ -625,7 +623,7 @@ Promise.all([
   testWritevDelegation(),
   testWriteSyncFirst(),
   testWriteSyncFallback(),
-  testWriteSyncThrowsFallback(),
+  testWriteSyncThrowsPropagation(),
   testEndSyncFirst(),
   testEndSyncFallback(),
   testFinalDelegatesToEnd(),

Original file line number	Diff line number	Diff line change
`@@ -779,8 +779,10 @@ function toWritable(writer) {`
`779`	`779`	`queueMicrotask(cb);`
`780`	`780`	`return;`
`781`	`781`	`}`
`782`		`- } catch {`
`783`		`- // Sync path threw -- fall through to async.`
	`782`	`+ // WriteSync returned false: not accepted, fall through to async.`
	`783`	`+ } catch (err) {`
	`784`	`+ cb(err);`
	`785`	`+ return;`
`784`	`786`	`}`
`785`	`787`	`}`
`786`	`788`	`try {`
`@@ -803,8 +805,10 @@ function toWritable(writer) {`
`803`	`805`	`queueMicrotask(cb);`
`804`	`806`	`return;`
`805`	`807`	`}`
`806`		`- } catch {`
`807`		`- // Sync path threw -- fall through to async.`
	`808`	`+ // WritevSync returned false: not accepted, fall through to async.`
	`809`	`+ } catch (err) {`
	`810`	`+ cb(err);`
	`811`	`+ return;`
`808`	`812`	`}`
`809`	`813`	`}`
`810`	`814`	`try {`
`@@ -826,8 +830,10 @@ function toWritable(writer) {`
`826`	`830`	`queueMicrotask(cb);`
`827`	`831`	`return;`
`828`	`832`	`}`
`829`		`- } catch {`
`830`		`- // Sync path threw -- fall through to async.`
	`833`	`+ // Result < 0: can't end synchronously, fall through to async.`
	`834`	`+ } catch (err) {`
	`835`	`+ cb(err);`
	`836`	`+ return;`
`831`	`837`	`}`
`832`	`838`	`}`
`833`	`839`	`try {`