@@ -16,6 +16,10 @@ const {
   DataViewPrototypeGetByteOffset,
   FunctionPrototypeCall,
   PromisePrototypeThen,
+  PromiseResolve,
+  PromiseWithResolvers,
+  SafePromiseRace,
+  Symbol,
   SymbolAsyncIterator,
   SymbolIterator,
   TypedArrayPrototypeGetBuffer,
@@ -29,8 +33,10 @@ const { markPromiseAsHandled } = internalBinding('util');
 const {
   codes: {
     ERR_INVALID_ARG_TYPE,
+    ERR_INVALID_RETURN_VALUE,
   },
 } = require('internal/errors');
+const { lazyDOMException } = require('internal/util');
 
 const {
   isAnyArrayBuffer,
@@ -54,6 +60,81 @@ const {
 // Bounds peak memory when arrays flow through transforms, which must
 // allocate output for the entire batch at once.
 const FROM_BATCH_SIZE = 128;
+const kNormalizationCancelled = Symbol('kNormalizationCancelled');
+
+function createNormalizationContext() {
+  return {
+    __proto__: null,
+    cancelled: false,
+    reason: undefined,
+    resolve: null,
+    suppressCleanup: false,
+  };
+}
+
+function cancelNormalization(context, reason, suppressCleanup = false) {
+  if (context.cancelled) return;
+  context.cancelled = true;
+  context.reason = reason;
+  context.suppressCleanup = suppressCleanup;
+  context.resolve?.(kNormalizationCancelled);
+}
+
+function throwIfNormalizationCancelled(context) {
+  if (context?.cancelled) throw context.reason;
+}
+
+async function waitForNormalization(value, context) {
+  if (context === undefined) return value;
+  const { promise, resolve } = PromiseWithResolvers();
+  if (context.cancelled) {
+    resolve(kNormalizationCancelled);
+  } else {
+    context.resolve = resolve;
+  }
+  try {
+    const result = await SafePromiseRace([
+      PromiseResolve(value),
+      promise,
+    ]);
+    throwIfNormalizationCancelled(context);
+    return result;
+  } finally {
+    if (context.resolve === resolve) context.resolve = null;
+  }
+}
+
+function createNormalizationIterator(createIterator) {
+  const context = createNormalizationContext();
+  const iterator = createIterator(context);
+  return {
+    __proto__: null,
+    next(value) {
+      return FunctionPrototypeCall(iterator.next, iterator, value);
+    },
+    return(value) {
+      cancelNormalization(
+        context, lazyDOMException('Aborted', 'AbortError'));
+      return FunctionPrototypeCall(iterator.return, iterator, value);
+    },
+    throw(error) {
+      cancelNormalization(context, error, true);
+      return FunctionPrototypeCall(iterator.throw, iterator, error);
+    },
+    [SymbolAsyncIterator]() {
+      return this;
+    },
+  };
+}
+
+function createNormalizationSource(createIterator) {
+  return {
+    __proto__: null,
+    [SymbolAsyncIterator]() {
+      return createNormalizationIterator(createIterator);
+    },
+  };
+}
 
 // =============================================================================
 // Type Guards and Detection
@@ -256,6 +337,101 @@ function* normalizeSyncSource(source) {
   }
 }
 
+function yieldNormalizationAbortable(source, context) {
+  if (context === undefined) return source;
+  return {
+    __proto__: null,
+    [SymbolAsyncIterator]() {
+      const iteratorMethod = source[SymbolAsyncIterator];
+      const iterator = FunctionPrototypeCall(iteratorMethod, source);
+      const nextMethod = iterator.next;
+      let completed = false;
+      let closed = false;
+      let reading = false;
+
+      async function closeSource(suppressError) {
+        if (closed) return;
+        closed = true;
+        completed = true;
+
+        if (suppressError) {
+          try {
+            const returnMethod = iterator.return;
+            if (typeof returnMethod === 'function') {
+              const cleanup = PromisePrototypeThen(
+                PromiseResolve(),
+                () => FunctionPrototypeCall(returnMethod, iterator));
+              markPromiseAsHandled(cleanup);
+            }
+          } catch {
+            // Cancellation has precedence over source cleanup errors.
+          }
+          return;
+        }
+
+        const returnMethod = iterator.return;
+        if (typeof returnMethod === 'function') {
+          const result = await FunctionPrototypeCall(returnMethod, iterator);
+          if ((typeof result !== 'object' && typeof result !== 'function') ||
+              result === null) {
+            throw new ERR_INVALID_RETURN_VALUE(
+              'an object', 'iterator.return()', result);
+          }
+        }
+      }
+
+      return {
+        __proto__: null,
+        async next() {
+          if (completed) {
+            return { __proto__: null, done: true, value: undefined };
+          }
+          throwIfNormalizationCancelled(context);
+          reading = true;
+
+          try {
+            const next = FunctionPrototypeCall(nextMethod, iterator);
+            const result = await waitForNormalization(next, context);
+            if ((typeof result !== 'object' && typeof result !== 'function') ||
+                result === null) {
+              throw new ERR_INVALID_RETURN_VALUE(
+                'an object', 'iterator.next()', result);
+            }
+            if (result.done) {
+              reading = false;
+              throwIfNormalizationCancelled(context);
+              completed = true;
+              closed = true;
+              return { __proto__: null, done: true, value: result.value };
+            }
+            const value = result.value;
+            reading = false;
+            throwIfNormalizationCancelled(context);
+            return { __proto__: null, done: false, value };
+          } catch (error) {
+            if (context.cancelled) await closeSource(true);
+            reading = false;
+            throw error;
+          }
+        },
+        async return(value) {
+          await closeSource(
+            context.suppressCleanup || (context.cancelled && reading));
+          return { __proto__: null, done: true, value };
+        },
+        async throw(error) {
+          await closeSource(
+            context.suppressCleanup || (context.cancelled && reading));
+          throw error;
+        },
+        [SymbolAsyncIterator]() {
+          return this;
+        },
+      };
+    },
+  };
+}
+
 // =============================================================================
 // Async Normalization (for from and async contexts)
 // =============================================================================
@@ -266,11 +442,15 @@ function* normalizeSyncSource(source) {
  * and protocol conversions.
  * @yields {Uint8Array}
  */
-async function* normalizeAsyncValue(value, allowNestedAsyncStreamables = true) {
+async function* normalizeAsyncValue(
+  value, allowNestedAsyncStreamables = true, context) {
+  throwIfNormalizationCancelled(context);
+
   // Handle promises first
   if (isPromise(value)) {
-    const resolved = await value;
-    yield* normalizeAsyncValue(resolved, allowNestedAsyncStreamables);
+    const resolved = await waitForNormalization(value, context);
+    yield* normalizeAsyncValue(
+      resolved, allowNestedAsyncStreamables, context);
     return;
   }
 
@@ -293,41 +473,47 @@ async function* normalizeAsyncValue(value, allowNestedAsyncStreamables = true) {
   if (hasProtocol(value, toAsyncStreamable)) {
     const result = FunctionPrototypeCall(value[toAsyncStreamable], value);
     if (isPromise(result)) {
-      yield* normalizeAsyncValue(await result, allowNestedAsyncStreamables);
+      yield* normalizeAsyncValue(
+        await waitForNormalization(result, context),
+        allowNestedAsyncStreamables,
+        context);
     } else {
-      yield* normalizeAsyncValue(result, allowNestedAsyncStreamables);
+      yield* normalizeAsyncValue(
+        result, allowNestedAsyncStreamables, context);
     }
     return;
   }
 
   // Handle ToStreamable protocol
   if (hasProtocol(value, toStreamable)) {
     const result = FunctionPrototypeCall(value[toStreamable], value);
-    yield* normalizeAsyncValue(result, allowNestedAsyncStreamables);
+    yield* normalizeAsyncValue(result, allowNestedAsyncStreamables, context);
     return;
   }
 
   // Handle arrays (which are also iterable, but check first for efficiency)
   if (ArrayIsArray(value)) {
     for (let i = 0; i < value.length; i++) {
-      yield* normalizeAsyncValue(value[i], allowNestedAsyncStreamables);
+      yield* normalizeAsyncValue(
+        value[i], allowNestedAsyncStreamables, context);
     }
     return;
   }
 
   // Handle async iterables (check before sync iterables since some objects
   // have both)
   if (isAsyncIterable(value)) {
-    for await (const item of value) {
-      yield* normalizeAsyncValue(item, allowNestedAsyncStreamables);
+    const iterable = yieldNormalizationAbortable(value, context);
+    for await (const item of iterable) {
+      yield* normalizeAsyncValue(item, allowNestedAsyncStreamables, context);
     }
     return;
   }
 
   // Handle sync iterables
   if (isSyncIterable(value)) {
     for (const item of value) {
-      yield* normalizeAsyncValue(item, allowNestedAsyncStreamables);
+      yield* normalizeAsyncValue(item, allowNestedAsyncStreamables, context);
     }
     return;
   }
@@ -346,10 +532,13 @@ async function* normalizeAsyncValue(value, allowNestedAsyncStreamables = true) {
  * @param {AsyncIterable|Iterable} source
  * @yields {Uint8Array[]}
  */
-async function* normalizeAsyncSource(source) {
+async function* normalizeAsyncSource(source, context) {
+  throwIfNormalizationCancelled(context);
+
   // Prefer async iteration if available
   if (isAsyncIterable(source)) {
-    for await (const value of source) {
+    const iterable = yieldNormalizationAbortable(source, context);
+    for await (const value of iterable) {
       // Fast path 1: value is already a Uint8Array[] batch
       if (isUint8ArrayBatch(value)) {
         if (value.length > 0) {
@@ -364,7 +553,7 @@ async function* normalizeAsyncSource(source) {
       }
       // Slow path: normalize the value
       let batch = [];
-      for await (const chunk of normalizeAsyncValue(value)) {
+      for await (const chunk of normalizeAsyncValue(value, true, context)) {
         ArrayPrototypePush(batch, chunk);
         if (batch.length === FROM_BATCH_SIZE) {
           yield batch;
@@ -383,6 +572,7 @@ async function* normalizeAsyncSource(source) {
     let batch = [];
 
     for (const value of source) {
+      throwIfNormalizationCancelled(context);
       // Fast path 1: value is already a Uint8Array[] batch
       if (isUint8ArrayBatch(value)) {
         // Flush any accumulated batch first
@@ -408,7 +598,7 @@ async function* normalizeAsyncSource(source) {
         batch = [];
       }
       let asyncBatch = [];
-      for await (const chunk of normalizeAsyncValue(value, false)) {
+      for await (const chunk of normalizeAsyncValue(value, false, context)) {
         ArrayPrototypePush(asyncBatch, chunk);
         if (asyncBatch.length === FROM_BATCH_SIZE) {
           yield asyncBatch;
@@ -434,6 +624,12 @@ async function* normalizeAsyncSource(source) {
   );
 }
 
+async function* normalizeAsyncStreamableResult(result, context) {
+  const resolved = await waitForNormalization(result, context);
+  const source = resolved?.[kValidatedSource] ? resolved : from(resolved);
+  yield* yieldNormalizationAbortable(source, context);
+}
+
 // =============================================================================
 // Public API: from() and fromSync()
 // =============================================================================
@@ -610,19 +806,8 @@ function from(input) {
     if (result?.[kValidatedSource]) {
       return result;
     }
-    return {
-      __proto__: null,
-      async *[SymbolAsyncIterator]() {
-        // The result may be a Promise. Check validated on both the Promise
-        // itself (if tagged) and the resolved value.
-        const resolved = await result;
-        if (resolved?.[kValidatedSource]) {
-          yield* resolved[SymbolAsyncIterator]();
-          return;
-        }
-        yield* from(resolved)[SymbolAsyncIterator]();
-      },
-    };
+    return createNormalizationSource(
+      (context) => normalizeAsyncStreamableResult(result, context));
   }
 
   // Check toStreamable protocol (takes precedence over iteration protocols)
@@ -640,7 +825,8 @@ function from(input) {
     );
   }
 
-  return normalizeAsyncSource(input);
+  return createNormalizationIterator(
+    (context) => normalizeAsyncSource(input, context));
 }
 
 // =============================================================================