stream: ensure from() observes returned rejecting promise correctly

Signed-off-by: James M Snell <jasnell@gmail.com>
Assisted-by: Opencode
PR-URL: #65658
Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>

Author: jasnell

lib/internal/streams/iter/from.js

@@ -15,6 +15,7 @@ const {
   DataViewPrototypeGetByteLength,
   DataViewPrototypeGetByteOffset,
   FunctionPrototypeCall,
+  PromisePrototypeThen,
   SymbolAsyncIterator,
   SymbolIterator,
   TypedArrayPrototypeGetBuffer,
@@ -23,6 +24,8 @@ const {
   Uint8Array,
 } = primordials;
 
+const { markPromiseAsHandled } = internalBinding('util');
+
 const {
   codes: {
     ERR_INVALID_ARG_TYPE,
@@ -594,7 +597,11 @@ function from(input) {
   // Check toAsyncStreamable protocol (takes precedence over toStreamable and
   // iteration protocols)
   if (typeof input[toAsyncStreamable] === 'function') {
-    const result = input[toAsyncStreamable]();
+    let result = input[toAsyncStreamable]();
+    if (isPromise(result)) {
+      result = PromisePrototypeThen(result, undefined, undefined);
+      markPromiseAsHandled(result);
+    }
     // Synchronous validated source (e.g. Readable batched iterator)
     if (result?.[kValidatedSource]) {
       return result;

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

@@ -235,6 +235,20 @@ async function testFromTopLevelProtocolOverIterator() {
   assert.strictEqual(result, 'from-protocol');
 }
 
+async function testFromHandlesProtocolRejectionUntilIteration() {
+  const reason = new Error('protocol failed');
+  const iterable = from({
+    [Symbol.for('Stream.toAsyncStreamable')]: common.mustCall(
+      () => Promise.reject(reason)),
+  });
+
+  await new Promise(setImmediate);
+  await assert.rejects(
+    iterable[Symbol.asyncIterator]().next(),
+    (error) => error === reason,
+  );
+}
+
 // DataView input should be converted to Uint8Array (zero-copy)
 async function testFromDataView() {
   const buf = new ArrayBuffer(5);
@@ -279,5 +293,6 @@ Promise.all([
   testFromTopLevelToStreamable(),
   testFromTopLevelAsyncPrecedence(),
   testFromTopLevelProtocolOverIterator(),
+  testFromHandlesProtocolRejectionUntilIteration(),
   testFromDataView(),
 ]).then(common.mustCall());