stream: fixup error handling in stream/iter adapters

Signed-off-by: James M Snell <jasnell@gmail.com>

Author: jasnell

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 {

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;
+        }
       }
     },
   };

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(),