diff --git a/doc/api/stream_iter.md b/doc/api/stream_iter.md
index f2fbba2286e0..55c36914755a 100644
--- a/doc/api/stream_iter.md
+++ b/doc/api/stream_iter.md
@@ -1374,7 +1374,8 @@ added:
 
 * `input` {AsyncIterable|Iterable|BroadcastChannel}
 * `options` {Object} Same as `broadcast()`.
-* Returns: {Object} `{ writer, broadcast }`
+* Returns: {BroadcastChannel|Object} A `broadcastProtocol` input returns its
+  {BroadcastChannel} directly. Other inputs return `{ writer, broadcast }`.
 
 Create a {BroadcastChannel} from an existing source. The source is consumed
 automatically and pushed to all subscribers.
@@ -1875,7 +1876,7 @@ class MessageBus {
 }
 
 const bus = new MessageBus();
-const { broadcast } = Broadcast.from(bus);
+const broadcast = Broadcast.from(bus);
 const consumer = broadcast.push();
 bus.send('hello');
 bus.close();
@@ -1911,7 +1912,7 @@ class MessageBus {
 }
 
 const bus = new MessageBus();
-const { broadcast } = Broadcast.from(bus);
+const broadcast = Broadcast.from(bus);
 const consumer = broadcast.push();
 bus.send('hello');
 bus.close();
diff --git a/lib/internal/streams/iter/broadcast.js b/lib/internal/streams/iter/broadcast.js
index ddce0d12d42c..3b37a5b5f2ab 100644
--- a/lib/internal/streams/iter/broadcast.js
+++ b/lib/internal/streams/iter/broadcast.js
@@ -10,6 +10,7 @@ const {
   ArrayIsArray,
   ArrayPrototypePush,
   ArrayPrototypeShift,
+  FunctionPrototypeCall,
   PromisePrototypeThen,
   PromiseReject,
   PromiseResolve,
@@ -24,6 +25,10 @@ const {
 } = primordials;
 
 const { lazyDOMException } = require('internal/util');
+const {
+  AbortController,
+  abortSignal,
+} = require('internal/abort_controller');
 
 const {
   codes: {
@@ -56,13 +61,14 @@ const {
   kResolvedPromise,
   convertChunks,
   createBatchEntry,
+  getProtocolMethod,
   getWriterSignal,
   getMinCursor,
-  hasProtocol,
   onSignalAbort,
   parsePullArgs,
   toWriterUint8Array,
   validateBatchEntry,
+  yieldAbortable,
 } = require('internal/streams/iter/utils');
 const {
   converters,
@@ -79,8 +85,10 @@ const kAbort = Symbol('kAbort');
 const kCanWrite = Symbol('kCanWrite');
 const kOnBufferDrained = Symbol('kOnBufferDrained');
 const kOnEndDrained = Symbol('kOnEndDrained');
+const kOnCancel = Symbol('kOnCancel');
 const kPendingWriteRemoved = Symbol('kPendingWriteRemoved');
 const kNoBroadcastError = Symbol('kNoBroadcastError');
+const kSetFactorySignal = Symbol('kSetFactorySignal');
 
 function raceEndWithSignal(promise, signal) {
   if (!signal) return promise;
@@ -104,7 +112,7 @@ class BroadcastImpl {
   #buffer = new RingBuffer();
   #bufferStart = 0;
   #consumers = new SafeSet();
-  #waiters = [];  // Consumers with pending resolve (subset of #consumers)
+  #waiters = new SafeSet();  // Consumers with pending resolve
   #ended = false;
   #error;
   #errored = false;
@@ -113,6 +121,7 @@ class BroadcastImpl {
   #writer = null;
   #cachedMinCursor = 0;
   #cachedMinCursorConsumers = 0;
+  #abortHandler;
   /** Cumulative byte size of buffered entries */
   #bufferedBytes = 0;
 
@@ -120,12 +129,18 @@ class BroadcastImpl {
     this.#options = options;
     this[kOnBufferDrained] = null;
     this[kOnEndDrained] = null;
+    this[kOnCancel] = null;
   }
 
   setWriter(writer) {
     this.#writer = writer;
   }
 
+  [kSetFactorySignal](signal) {
+    this.#abortHandler = () => this.cancel(signal.reason);
+    onSignalAbort(signal, this.#abortHandler);
+  }
+
   get backpressurePolicy() {
     return this.#options.backpressure;
   }
@@ -196,6 +211,7 @@ class BroadcastImpl {
 
     function detach() {
       state.detached = true;
+      self.#waiters.delete(state);
       if (state.resolve) {
         state.resolve({ __proto__: null, done: true, value: undefined });
       }
@@ -255,7 +271,7 @@ class BroadcastImpl {
             const { promise, resolve, reject } = PromiseWithResolvers();
             state.resolve = resolve;
             state.reject = reject;
-            ArrayPrototypePush(self.#waiters, state);
+            self.#waiters.add(state);
             return promise;
           },
 
@@ -306,7 +322,12 @@ class BroadcastImpl {
       consumer.detached = true;
     }
     this.#consumers.clear();
+    this.#waiters.clear();
     this.#cachedMinCursorConsumers = 0;
+    this.#cleanupFactorySignal();
+    const onCancel = this[kOnCancel];
+    this[kOnCancel] = null;
+    onCancel?.(reason);
   }
 
   [SymbolDispose]() {
@@ -391,6 +412,7 @@ class BroadcastImpl {
         }
       }
     }
+    this.#waiters.clear();
     this.#notifyEndDrained();
   }
 
@@ -412,7 +434,9 @@ class BroadcastImpl {
       consumer.detached = true;
     }
     this.#consumers.clear();
+    this.#waiters.clear();
     this.#cachedMinCursorConsumers = 0;
+    this.#cleanupFactorySignal();
   }
 
   /**
@@ -432,10 +456,18 @@ class BroadcastImpl {
 
   #notifyEndDrained() {
     if (this.#ended && this.#consumers.size === 0) {
+      this.#cleanupFactorySignal();
       this[kOnEndDrained]?.();
     }
   }
 
+  #cleanupFactorySignal() {
+    if (this.#abortHandler !== undefined) {
+      this.#options.signal.removeEventListener('abort', this.#abortHandler);
+      this.#abortHandler = undefined;
+    }
+  }
+
   #recomputeMinCursor() {
     const { minCursor, minCursorConsumers } = getMinCursor(
       this.#consumers, this.#bufferStart + this.#buffer.length);
@@ -444,6 +476,8 @@ class BroadcastImpl {
   }
 
   #tryTrimBuffer() {
+    // Retain buffered data for consumers that attach while none are active.
+    if (this.#consumers.size === 0) return;
     if (this.#cachedMinCursorConsumers === 0) {
       this.#recomputeMinCursor();
     }
@@ -477,12 +511,11 @@ class BroadcastImpl {
 
   #notifyConsumers() {
     const waiters = this.#waiters;
-    if (waiters.length === 0) return;
+    if (waiters.size === 0) return;
     // Swap out the waiters list so consumers that re-wait during
     // resolve don't get processed twice in this cycle.
-    this.#waiters = [];
-    for (let i = 0; i < waiters.length; i++) {
-      const consumer = waiters[i];
+    this.#waiters = new SafeSet();
+    for (const consumer of waiters) {
       if (consumer.resolve) {
         const bufferIndex = consumer.cursor - this.#bufferStart;
         if (bufferIndex < this.#buffer.length) {
@@ -501,11 +534,11 @@ class BroadcastImpl {
           if (consumer.detached && this.#deleteConsumer(consumer)) {
             this.#tryTrimBuffer();
           } else if (this.#promotePending(consumer)) {
-            ArrayPrototypePush(this.#waiters, consumer);
+            this.#waiters.add(consumer);
           }
         } else {
           // Still waiting -- put back
-          ArrayPrototypePush(this.#waiters, consumer);
+          this.#waiters.add(consumer);
         }
       }
     }
@@ -841,10 +874,6 @@ function wireBroadcastWriteSignal(entry, signal, resolve, reject, self) {
   signal.addEventListener('abort', onAbort, { __proto__: null, once: true });
 }
 
-function onBroadcastCancel(broadcastImpl, signal) {
-  onSignalAbort(signal, () => broadcastImpl.cancel(signal.reason));
-}
-
 // =============================================================================
 // Public API
 // =============================================================================
@@ -878,26 +907,23 @@ function broadcast(options = { __proto__: null }) {
   broadcastImpl.setWriter(writer);
 
   if (signal) {
-    onBroadcastCancel(broadcastImpl, signal);
+    broadcastImpl[kSetFactorySignal](signal);
   }
 
   return { __proto__: null, writer, broadcast: broadcastImpl };
 }
 
-function isBroadcastable(value) {
-  return hasProtocol(value, broadcastProtocol);
-}
-
 const Broadcast = {
   __proto__: null,
   from(input, options) {
-    if (isBroadcastable(input)) {
-      const bc = input[broadcastProtocol](options);
+    const protocol = getProtocolMethod(input, broadcastProtocol);
+    if (protocol !== undefined) {
+      const bc = FunctionPrototypeCall(protocol, input, options);
       if (bc === null || typeof bc !== 'object') {
         throw new ERR_INVALID_RETURN_VALUE(
           'an object', '[Symbol.for(\'Stream.broadcastProtocol\')]', bc);
       }
-      return { __proto__: null, writer: { __proto__: null }, broadcast: bc };
+      return bc;
     }
 
     const source = from(input);
@@ -913,13 +939,23 @@ const Broadcast = {
     });
     const result = broadcast(options);
     const { signal } = options;
+    const controller = new AbortController();
+    if (signal?.aborted) {
+      abortSignal(controller.signal, signal.reason);
+    }
+    const onCancel = (reason) => {
+      if (!controller.signal.aborted) {
+        abortSignal(controller.signal, reason);
+      }
+    };
+    result.broadcast[kOnCancel] = onCancel;
 
     const pump = async () => {
       const w = result.writer;
       try {
         if (isAsyncIterable(source)) {
-          for await (const chunks of source) {
-            signal?.throwIfAborted();
+          for await (const chunks of yieldAbortable(source, controller.signal)) {
+            controller.signal.throwIfAborted();
             if (ArrayIsArray(chunks)) {
               if (!w.writevSync(chunks)) {
                 await w.writev(chunks, signal ? { signal } : undefined);
@@ -930,7 +966,7 @@ const Broadcast = {
           }
         } else if (isSyncIterable(source)) {
           for (const chunks of source) {
-            signal?.throwIfAborted();
+            controller.signal.throwIfAborted();
             if (ArrayIsArray(chunks)) {
               if (!w.writevSync(chunks)) {
                 await w.writev(chunks, signal ? { signal } : undefined);
@@ -944,7 +980,13 @@ const Broadcast = {
           await w.end(signal ? { signal } : undefined);
         }
       } catch (error) {
-        w.fail(error);
+        if (!controller.signal.aborted) {
+          w.fail(error);
+        }
+      } finally {
+        if (result.broadcast[kOnCancel] === onCancel) {
+          result.broadcast[kOnCancel] = null;
+        }
       }
     };
     PromisePrototypeThen(pump(), undefined, () => {});
diff --git a/lib/internal/streams/iter/consumers.js b/lib/internal/streams/iter/consumers.js
index 75ad4026ad7c..9e9a8fde4ec0 100644
--- a/lib/internal/streams/iter/consumers.js
+++ b/lib/internal/streams/iter/consumers.js
@@ -15,6 +15,7 @@ const {
   ArrayPrototypePush,
   ArrayPrototypeShift,
   ArrayPrototypeSlice,
+  FunctionPrototypeCall,
   Promise,
   PromisePrototypeThen,
   SafePromiseAllReturnVoid,
@@ -53,6 +54,7 @@ const {
 const {
   concatBytes,
   createBatchEntry,
+  getProtocolMethod,
   validateBatchEntry,
   yieldAbortable,
 } = require('internal/streams/iter/utils');
@@ -391,14 +393,9 @@ function ondrain(drainable) {
     return null;
   }
 
-  if (
-    !(drainableProtocol in drainable) ||
-    typeof drainable[drainableProtocol] !== 'function'
-  ) {
-    return null;
-  }
-
-  return drainable[drainableProtocol]();
+  const protocol = getProtocolMethod(drainable, drainableProtocol);
+  return protocol === undefined ?
+    null : FunctionPrototypeCall(protocol, drainable);
 }
 
 // =============================================================================
diff --git a/lib/internal/streams/iter/duplex.js b/lib/internal/streams/iter/duplex.js
index 674ef81a53c9..95a117d81419 100644
--- a/lib/internal/streams/iter/duplex.js
+++ b/lib/internal/streams/iter/duplex.js
@@ -45,8 +45,10 @@ function duplex(options = { __proto__: null }) {
     backpressure: b?.backpressure ?? backpressure,
   });
 
-  const channelA = createDuplexChannel(aWriter, aReadable);
-  const channelB = createDuplexChannel(bWriter, bReadable);
+  let cleanupSignal;
+  const onClose = () => cleanupSignal?.();
+  const channelA = createDuplexChannel(aWriter, aReadable, onClose);
+  const channelB = createDuplexChannel(bWriter, bReadable, onClose);
 
   // Signal handler: fail both writers with the abort reason so consumers
   // see the error. This is an error-path shutdown, not a clean close.
@@ -55,6 +57,11 @@ function duplex(options = { __proto__: null }) {
       const reason = signal.reason;
       aWriter.fail(reason);
       bWriter.fail(reason);
+      cleanupSignal?.();
+    };
+    cleanupSignal = () => {
+      signal.removeEventListener('abort', abortBoth);
+      cleanupSignal = undefined;
     };
     if (signal.aborted) {
       abortBoth();
@@ -67,7 +74,7 @@ function duplex(options = { __proto__: null }) {
   return [channelA, channelB];
 }
 
-function createDuplexChannel(writer, readable) {
+function createDuplexChannel(writer, readable, onClose) {
   // A push readable has one shared consumer state. Keeping an iterator from
   // creation lets close() terminate that state even if no caller has iterated.
   const closeIterator = readable[SymbolAsyncIterator]();
@@ -78,7 +85,7 @@ function createDuplexChannel(writer, readable) {
     get writer() { return writer; },
     get readable() { return readable; },
     close() {
-      closePromise ??= closeDuplexChannel(writer, closeIterator);
+      closePromise ??= closeDuplexChannel(writer, closeIterator, onClose);
       return closePromise;
     },
     [SymbolAsyncDispose]() {
@@ -87,19 +94,23 @@ function createDuplexChannel(writer, readable) {
   };
 }
 
-async function closeDuplexChannel(writer, closeIterator) {
-  const result = writer.endSync();
-  const endPromise = result < 0 ? writer.end() : undefined;
-  const returnPromise = closeIterator.return();
+async function closeDuplexChannel(writer, closeIterator, onClose) {
+  try {
+    const result = writer.endSync();
+    const endPromise = result < 0 ? writer.end() : undefined;
+    const returnPromise = closeIterator.return();
 
-  if (endPromise !== undefined) {
-    try {
-      await SafePromiseAllReturnVoid([endPromise, returnPromise]);
-    } catch (error) {
-      if (!isConsumerReturnError(error)) throw error;
+    if (endPromise !== undefined) {
+      try {
+        await SafePromiseAllReturnVoid([endPromise, returnPromise]);
+      } catch (error) {
+        if (!isConsumerReturnError(error)) throw error;
+      }
+    } else {
+      await returnPromise;
     }
-  } else {
-    await returnPromise;
+  } finally {
+    onClose();
   }
 }
 
diff --git a/lib/internal/streams/iter/from.js b/lib/internal/streams/iter/from.js
index 7fe03511861f..cac8fd3565b6 100644
--- a/lib/internal/streams/iter/from.js
+++ b/lib/internal/streams/iter/from.js
@@ -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,
@@ -46,7 +52,7 @@ const {
 } = require('internal/streams/iter/types');
 
 const {
-  hasProtocol,
+  getProtocolMethod,
   toUint8Array,
 } = require('internal/streams/iter/utils');
 
@@ -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
@@ -144,8 +225,9 @@ function* normalizeSyncValue(value) {
   }
 
   // Handle ToStreamable protocol
-  if (hasProtocol(value, toStreamable)) {
-    const result = FunctionPrototypeCall(value[toStreamable], value);
+  const streamableMethod = getProtocolMethod(value, toStreamable);
+  if (streamableMethod !== undefined) {
+    const result = FunctionPrototypeCall(streamableMethod, value);
     yield* normalizeSyncValue(result);
     return;
   }
@@ -256,6 +338,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 +443,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;
   }
 
@@ -280,8 +461,12 @@ async function* normalizeAsyncValue(value, allowNestedAsyncStreamables = true) {
     return;
   }
 
-  if (!allowNestedAsyncStreamables &&
-      (isAsyncIterable(value) || hasProtocol(value, toAsyncStreamable))) {
+  const hasDisallowedAsyncIterator =
+    !allowNestedAsyncStreamables && isAsyncIterable(value);
+  const asyncStreamableMethod = hasDisallowedAsyncIterator ?
+    undefined : getProtocolMethod(value, toAsyncStreamable);
+  if (hasDisallowedAsyncIterator ||
+      (!allowNestedAsyncStreamables && asyncStreamableMethod !== undefined)) {
     throw new ERR_INVALID_ARG_TYPE(
       'value',
       ['string', 'ArrayBuffer', 'ArrayBufferView', 'Iterable', 'toStreamable'],
@@ -290,27 +475,33 @@ async function* normalizeAsyncValue(value, allowNestedAsyncStreamables = true) {
   }
 
   // Handle ToAsyncStreamable protocol (check before ToStreamable)
-  if (hasProtocol(value, toAsyncStreamable)) {
-    const result = FunctionPrototypeCall(value[toAsyncStreamable], value);
+  if (asyncStreamableMethod !== undefined) {
+    const result = FunctionPrototypeCall(asyncStreamableMethod, 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);
+  const streamableMethod = getProtocolMethod(value, toStreamable);
+  if (streamableMethod !== undefined) {
+    const result = FunctionPrototypeCall(streamableMethod, value);
+    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;
   }
@@ -318,8 +509,9 @@ async function* normalizeAsyncValue(value, allowNestedAsyncStreamables = true) {
   // 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;
   }
@@ -327,7 +519,7 @@ async function* normalizeAsyncValue(value, allowNestedAsyncStreamables = true) {
   // Handle sync iterables
   if (isSyncIterable(value)) {
     for (const item of value) {
-      yield* normalizeAsyncValue(item, allowNestedAsyncStreamables);
+      yield* normalizeAsyncValue(item, allowNestedAsyncStreamables, context);
     }
     return;
   }
@@ -346,10 +538,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 +559,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 +578,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 +604,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 +630,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()
 // =============================================================================
@@ -459,6 +661,13 @@ function fromSync(input) {
     };
   }
 
+  // Check toStreamable protocol (takes precedence over iteration protocols).
+  // toAsyncStreamable is ignored entirely in fromSync.
+  const streamableMethod = getProtocolMethod(input, toStreamable);
+  if (streamableMethod !== undefined) {
+    return fromSync(FunctionPrototypeCall(streamableMethod, input));
+  }
+
   // Fast path: Uint8Array[] - yield in bounded sub-batches.
   // Yielding the entire array as one batch forces downstream transforms
   // to process all data at once, causing peak memory proportional to total
@@ -494,12 +703,6 @@ function fromSync(input) {
     }
   }
 
-  // Check toStreamable protocol (takes precedence over iteration protocols).
-  // toAsyncStreamable is ignored entirely in fromSync.
-  if (typeof input[toStreamable] === 'function') {
-    return fromSync(input[toStreamable]());
-  }
-
   const isIterable = isSyncIterable(input);
 
   // Reject explicit async-only inputs
@@ -548,11 +751,6 @@ function from(input) {
     throw new ERR_INVALID_ARG_TYPE('input', 'a non-null value', input);
   }
 
-  // Fast path: validated source already yields valid Uint8Array[] batches
-  if (input[kValidatedSource]) {
-    return input;
-  }
-
   // Check for primitives first (ByteInput)
   if (isPrimitiveChunk(input)) {
     const chunk = primitiveToUint8Array(input);
@@ -564,6 +762,34 @@ function from(input) {
     };
   }
 
+  // Check toAsyncStreamable protocol (takes precedence over toStreamable and
+  // iteration protocols)
+  const asyncStreamableMethod = getProtocolMethod(input, toAsyncStreamable);
+  if (asyncStreamableMethod !== undefined) {
+    let result = FunctionPrototypeCall(asyncStreamableMethod, input);
+    if (isPromise(result)) {
+      result = PromisePrototypeThen(result, undefined, undefined);
+      markPromiseAsHandled(result);
+    }
+    // Synchronous validated source (e.g. Readable batched iterator)
+    if (result?.[kValidatedSource]) {
+      return result;
+    }
+    return createNormalizationSource(
+      (context) => normalizeAsyncStreamableResult(result, context));
+  }
+
+  // Check toStreamable protocol (takes precedence over iteration protocols)
+  const streamableMethod = getProtocolMethod(input, toStreamable);
+  if (streamableMethod !== undefined) {
+    return from(FunctionPrototypeCall(streamableMethod, input));
+  }
+
+  // Fast path: validated source already yields valid Uint8Array[] batches
+  if (input[kValidatedSource]) {
+    return input;
+  }
+
   // Fast path: Uint8Array[] - yield in bounded sub-batches.
   // Yielding the entire array as one batch forces downstream transforms
   // to process all data at once, causing peak memory proportional to total
@@ -598,38 +824,6 @@ function from(input) {
     }
   }
 
-  // Check toAsyncStreamable protocol (takes precedence over toStreamable and
-  // iteration protocols)
-  if (typeof input[toAsyncStreamable] === 'function') {
-    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;
-    }
-    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]();
-      },
-    };
-  }
-
-  // Check toStreamable protocol (takes precedence over iteration protocols)
-  if (typeof input[toStreamable] === 'function') {
-    return from(input[toStreamable]());
-  }
-
   // Must be a Streamable (sync or async iterable)
   if (!isSyncIterable(input) && !isAsyncIterable(input)) {
     throw new ERR_INVALID_ARG_TYPE(
@@ -640,7 +834,8 @@ function from(input) {
     );
   }
 
-  return normalizeAsyncSource(input);
+  return createNormalizationIterator(
+    (context) => normalizeAsyncSource(input, context));
 }
 
 // =============================================================================
diff --git a/lib/internal/streams/iter/pull.js b/lib/internal/streams/iter/pull.js
index 01a9504dc871..9b4d7fb884d1 100644
--- a/lib/internal/streams/iter/pull.js
+++ b/lib/internal/streams/iter/pull.js
@@ -49,9 +49,9 @@ const {
 
 const {
   createBatchEntry,
-  isTransform,
   isTransformObject,
   parsePullArgs,
+  snapshotTransform,
   toUint8Array,
   validateBatchEntry,
   validateByteView,
@@ -93,7 +93,8 @@ function parsePipeToArgs(args, requiredMethod) {
 
   // Check if last arg is options
   const last = args[args.length - 1];
-  if (!isTransform(last) && !hasMethod(last, requiredMethod)) {
+  if (snapshotTransform(last) === undefined &&
+      !hasMethod(last, requiredMethod)) {
     options = last;
     writerIndex = args.length - 2;
   }
@@ -110,11 +111,13 @@ function parsePipeToArgs(args, requiredMethod) {
 
   const transforms = ArrayPrototypeSlice(args, 0, writerIndex);
   for (let i = 0; i < transforms.length; i++) {
-    if (!isTransform(transforms[i])) {
+    const transform = snapshotTransform(transforms[i]);
+    if (transform === undefined) {
       throw new ERR_INVALID_ARG_TYPE(
         `transforms[${i}]`, ['Function', 'Object with transform()'],
         transforms[i]);
     }
+    transforms[i] = transform;
   }
 
   return {
@@ -540,7 +543,7 @@ function* createSyncPipeline(source, transforms) {
         statelessRun = [];
       }
       current = applyStatefulSyncTransform(
-        current, transform.transform, transform);
+        current, transform.transform, transform.receiver);
     } else {
       ArrayPrototypePush(statelessRun, transform);
     }
@@ -728,51 +731,51 @@ async function* createAsyncPipeline(source, transforms, signal) {
     signal.addEventListener('abort', abortHandler, { __proto__: null, once: true });
   }
 
-  // Apply transforms - fuse consecutive stateless transforms into a single
-  // generator layer to avoid unnecessary async generator ticks.
-  //
-  // INVARIANT: Each transform invocation MUST receive its own fresh options
-  // object ({ __proto__: null, signal }). Transforms may mutate the options
-  // object, so sharing a single object across invocations would allow one
-  // transform to corrupt the options seen by another. The signal is shared
-  // across calls (mutations to it are acceptable), but the containing options
-  // object must be unique per call. This is enforced inside
-  // applyFusedStatelessAsyncTransforms and applyStatefulAsyncTransform, which
-  // accept the signal directly and create the options object per invocation.
-  // DO NOT pass a pre-built options object.
-  let current = normalized;
-  const transformSignal = controller.signal;
-  let statelessRun = [];
-
-  for (let i = 0; i < transforms.length; i++) {
-    const transform = transforms[i];
-    if (isTransformObject(transform)) {
-      // Flush any accumulated stateless run before the stateful transform
-      if (statelessRun.length > 0) {
-        current = applyFusedStatelessAsyncTransforms(current, statelessRun,
-                                                     transformSignal);
-        statelessRun = [];
-      }
-      const opts = { __proto__: null, signal: transformSignal };
-      if (transform[kValidatedTransform]) {
-        current = applyValidatedStatefulAsyncTransform(
-          current, transform.transform, transform, opts);
+  let completed = false;
+  try {
+    // Apply transforms - fuse consecutive stateless transforms into a single
+    // generator layer to avoid unnecessary async generator ticks.
+    //
+    // INVARIANT: Each transform invocation MUST receive its own fresh options
+    // object ({ __proto__: null, signal }). Transforms may mutate the options
+    // object, so sharing a single object across invocations would allow one
+    // transform to corrupt the options seen by another. The signal is shared
+    // across calls (mutations to it are acceptable), but the containing options
+    // object must be unique per call. This is enforced inside
+    // applyFusedStatelessAsyncTransforms and applyStatefulAsyncTransform, which
+    // accept the signal directly and create the options object per invocation.
+    // DO NOT pass a pre-built options object.
+    let current = normalized;
+    const transformSignal = controller.signal;
+    let statelessRun = [];
+
+    for (let i = 0; i < transforms.length; i++) {
+      const transform = transforms[i];
+      if (isTransformObject(transform)) {
+        // Flush any accumulated stateless run before the stateful transform
+        if (statelessRun.length > 0) {
+          current = applyFusedStatelessAsyncTransforms(current, statelessRun,
+                                                       transformSignal);
+          statelessRun = [];
+        }
+        const opts = { __proto__: null, signal: transformSignal };
+        if (transform[kValidatedTransform]) {
+          current = applyValidatedStatefulAsyncTransform(
+            current, transform.transform, transform.receiver, opts);
+        } else {
+          current = applyStatefulAsyncTransform(
+            current, transform.transform, transform.receiver, opts);
+        }
       } else {
-        current = applyStatefulAsyncTransform(
-          current, transform.transform, transform, opts);
+        ArrayPrototypePush(statelessRun, transform);
       }
-    } else {
-      ArrayPrototypePush(statelessRun, transform);
     }
-  }
-  // Flush remaining stateless run
-  if (statelessRun.length > 0) {
-    current = applyFusedStatelessAsyncTransforms(current, statelessRun,
-                                                 transformSignal);
-  }
+    // Flush remaining stateless run
+    if (statelessRun.length > 0) {
+      current = applyFusedStatelessAsyncTransforms(current, statelessRun,
+                                                   transformSignal);
+    }
 
-  let completed = false;
-  try {
     for await (const batch of current) {
       controller.signal.throwIfAborted();
       yield batch;
@@ -813,11 +816,13 @@ async function* createAsyncPipeline(source, transforms, signal) {
 function pullSync(source, ...transforms) {
   const normalized = fromSync(source);
   for (let i = 0; i < transforms.length; i++) {
-    if (!isTransform(transforms[i])) {
+    const transform = snapshotTransform(transforms[i]);
+    if (transform === undefined) {
       throw new ERR_INVALID_ARG_TYPE(
         `transforms[${i}]`, ['Function', 'Object with transform()'],
         transforms[i]);
     }
+    transforms[i] = transform;
   }
   return {
     __proto__: null,
diff --git a/lib/internal/streams/iter/share.js b/lib/internal/streams/iter/share.js
index 6154509a5b64..4e0ac708db17 100644
--- a/lib/internal/streams/iter/share.js
+++ b/lib/internal/streams/iter/share.js
@@ -39,8 +39,8 @@ const {
 const {
   kMultiConsumerDefaultBudget,
   createBatchEntry,
+  getProtocolMethod,
   getMinCursor,
-  hasProtocol,
   onSignalAbort,
   parsePullArgs,
   validateBatchEntry,
@@ -56,6 +56,7 @@ const {
 const {
   codes: {
     ERR_INVALID_ARG_TYPE,
+    ERR_INVALID_ARG_VALUE,
     ERR_INVALID_RETURN_VALUE,
     ERR_OUT_OF_RANGE,
   },
@@ -70,6 +71,7 @@ const {
 
 const kNoShareError = Symbol('kNoShareError');
 const kShareCancelled = Symbol('kShareCancelled');
+const kSetFactorySignal = Symbol('kSetFactorySignal');
 
 class ShareImpl {
   #source;
@@ -88,6 +90,7 @@ class ShareImpl {
   #cancelError = kNoShareError;
   #cachedMinCursor = 0;
   #cachedMinCursorConsumers = 0;
+  #abortHandler;
   /** Cumulative byte size of buffered entries */
   #bufferedBytes = 0;
 
@@ -103,6 +106,11 @@ class ShareImpl {
     return this.#consumers.size;
   }
 
+  [kSetFactorySignal](signal) {
+    this.#abortHandler = () => this.cancel(signal.reason);
+    onSignalAbort(signal, this.#abortHandler);
+  }
+
   pull(...args) {
     const parsed = parsePullArgs(args);
     const { transforms } = parsed;
@@ -200,7 +208,16 @@ class ShareImpl {
             }
 
             // Need to pull from source - check buffer limit
-            const shouldBuffer = await self.#waitForBufferSpace();
+            let shouldBuffer;
+            try {
+              shouldBuffer = await self.#waitForBufferSpace();
+            } catch (error) {
+              state.detached = true;
+              if (self.#deleteConsumer(state)) {
+                self.#tryTrimBuffer();
+              }
+              throw error;
+            }
             if (shouldBuffer === null) {
               state.detached = true;
               state.error = self.#cancelError;
@@ -293,6 +310,7 @@ class ShareImpl {
     this.#consumers.clear();
     this.#buffer.clear();
     this.#bufferedBytes = 0;
+    this.#cleanupFactorySignal();
 
     for (let i = 0; i < this.#pullWaiters.length; i++) {
       this.#pullWaiters[i]();
@@ -414,6 +432,7 @@ class ShareImpl {
         this.#sourceError = error;
         this.#sourceExhausted = true;
       } finally {
+        if (this.#sourceExhausted) this.#cleanupFactorySignal();
         this.#pulling = false;
         for (let i = 0; i < this.#pullWaiters.length; i++) {
           this.#pullWaiters[i]();
@@ -460,6 +479,13 @@ class ShareImpl {
     this.#cachedMinCursorConsumers = minCursorConsumers;
   }
 
+  #cleanupFactorySignal() {
+    if (this.#abortHandler !== undefined) {
+      this.#options.signal.removeEventListener('abort', this.#abortHandler);
+      this.#abortHandler = undefined;
+    }
+  }
+
   #deleteConsumerFromMin(consumer) {
     if (consumer.cursor === this.#cachedMinCursor) {
       this.#cachedMinCursorConsumers--;
@@ -579,11 +605,6 @@ class SyncShareImpl {
                   throw new ERR_OUT_OF_RANGE(
                     'buffered bytes', `< ${self.#options.budget}`,
                     self.#bufferedBytes);
-                case 'unbounded':
-                  throw new ERR_OUT_OF_RANGE(
-                    'buffered bytes', `< ${self.#options.budget} ` +
-                    '(unbounded not available in sync context)',
-                    self.#bufferedBytes);
                 case 'drop-oldest':
                   while (self.#bufferedBytes >= self.#options.budget &&
                          self.#buffer.length > 0) {
@@ -600,9 +621,13 @@ class SyncShareImpl {
                   self.#recomputeMinCursor();
                   break;
                 case 'drop-newest':
-                  state.detached = true;
-                  self.#deleteConsumer(state);
-                  return { __proto__: null, done: true, value: undefined };
+                  while (self.#bufferedBytes >= self.#options.budget &&
+                         !self.#sourceExhausted &&
+                         !self.#cancelled &&
+                         self.#sourceError === kNoShareError) {
+                    self.#pullFromSource(true);
+                  }
+                  break;
               }
             }
 
@@ -685,7 +710,7 @@ class SyncShareImpl {
     this.cancel();
   }
 
-  #pullFromSource() {
+  #pullFromSource(discard = false) {
     if (this.#sourceExhausted || this.#cancelled) return;
 
     try {
@@ -695,7 +720,7 @@ class SyncShareImpl {
 
       if (result.done) {
         this.#sourceExhausted = true;
-      } else {
+      } else if (!discard) {
         const entry = createBatchEntry(result.value);
         this.#buffer.push(entry);
         this.#bufferedBytes += entry.byteLength;
@@ -755,10 +780,6 @@ class SyncShareImpl {
   }
 }
 
-function onShareCancel(shareImpl, signal) {
-  onSignalAbort(signal, () => shareImpl.cancel(signal.reason));
-}
-
 // =============================================================================
 // Public API
 // =============================================================================
@@ -787,7 +808,7 @@ function share(source, options = { __proto__: null }) {
   const shareImpl = new ShareImpl(normalized, opts);
 
   if (signal) {
-    onShareCancel(shareImpl, signal);
+    shareImpl[kSetFactorySignal](signal);
   }
 
   return shareImpl;
@@ -805,6 +826,11 @@ function shareSync(source, options = { __proto__: null }) {
     backpressure = 'strict',
   } = options;
   validateInteger(budget, 'options.budget', 16384);
+  if (backpressure === 'unbounded') {
+    throw new ERR_INVALID_ARG_VALUE(
+      'options.backpressure', backpressure,
+      'unbounded is not supported by shareSync()');
+  }
 
   const opts = {
     __proto__: null,
@@ -815,19 +841,12 @@ function shareSync(source, options = { __proto__: null }) {
   return new SyncShareImpl(normalized, opts);
 }
 
-function isShareable(value) {
-  return hasProtocol(value, shareProtocol);
-}
-
-function isSyncShareable(value) {
-  return hasProtocol(value, shareSyncProtocol);
-}
-
 const Share = {
   __proto__: null,
   from(input, options) {
-    if (isShareable(input)) {
-      const result = input[shareProtocol](options);
+    const protocol = getProtocolMethod(input, shareProtocol);
+    if (protocol !== undefined) {
+      const result = FunctionPrototypeCall(protocol, input, options);
       if (result === null || typeof result !== 'object') {
         throw new ERR_INVALID_RETURN_VALUE(
           'an object', '[Symbol.for(\'Stream.shareProtocol\')]', result);
@@ -845,8 +864,9 @@ const Share = {
 const SyncShare = {
   __proto__: null,
   fromSync(input, options) {
-    if (isSyncShareable(input)) {
-      const result = input[shareSyncProtocol](options);
+    const protocol = getProtocolMethod(input, shareSyncProtocol);
+    if (protocol !== undefined) {
+      const result = FunctionPrototypeCall(protocol, input, options);
       if (result === null || typeof result !== 'object') {
         throw new ERR_INVALID_RETURN_VALUE(
           'an object', '[Symbol.for(\'Stream.shareSyncProtocol\')]', result);
diff --git a/lib/internal/streams/iter/utils.js b/lib/internal/streams/iter/utils.js
index 95d2d914eb1a..a6069c9ed3b7 100644
--- a/lib/internal/streams/iter/utils.js
+++ b/lib/internal/streams/iter/utils.js
@@ -9,6 +9,7 @@ const {
   PromiseWithResolvers,
   SafePromisePrototypeFinally,
   SafePromiseRace,
+  SafeWeakSet,
   SymbolAsyncIterator,
   TypedArrayPrototypeGetBuffer,
   TypedArrayPrototypeGetByteLength,
@@ -37,6 +38,9 @@ const {
 const {
   converters,
 } = require('internal/streams/iter/webidl');
+const {
+  kValidatedTransform,
+} = require('internal/streams/iter/types');
 
 // Cached resolved promise to avoid allocating a new one on every sync fast-path.
 const kResolvedPromise = PromiseResolve();
@@ -341,13 +345,40 @@ function toWriterUint8Array(chunk) {
  * @param {symbol} symbol
  * @returns {boolean}
  */
+function getProtocolMethod(value, symbol) {
+  if (value === null || typeof value !== 'object' || !(symbol in value)) {
+    return undefined;
+  }
+  const method = value[symbol];
+  return typeof method === 'function' ? method : undefined;
+}
+
 function hasProtocol(value, symbol) {
-  return (
-    value !== null &&
-    typeof value === 'object' &&
-    symbol in value &&
-    typeof value[symbol] === 'function'
-  );
+  return getProtocolMethod(value, symbol) !== undefined;
+}
+
+const transformRecords = new SafeWeakSet();
+
+/**
+ * Read and retain a stateful transform's callable exactly once.
+ * @param {unknown} value
+ * @returns {Function|object|undefined}
+ */
+function snapshotTransform(value) {
+  if (typeof value === 'function') return value;
+  if (transformRecords.has(value)) return value;
+
+  const transform = value?.transform;
+  if (typeof transform !== 'function') return undefined;
+
+  const record = {
+    __proto__: null,
+    transform,
+    receiver: value,
+    [kValidatedTransform]: value[kValidatedTransform],
+  };
+  transformRecords.add(record);
+  return record;
 }
 
 /**
@@ -356,7 +387,7 @@ function hasProtocol(value, symbol) {
  * @returns {boolean}
  */
 function isTransformObject(value) {
-  return typeof value?.transform === 'function';
+  return transformRecords.has(value) || typeof value?.transform === 'function';
 }
 
 /**
@@ -382,20 +413,24 @@ function parsePullArgs(args) {
   let transforms;
   let options;
   const last = args[args.length - 1];
-  if (!isTransform(last)) {
+  const lastTransform = snapshotTransform(last);
+  if (lastTransform === undefined) {
     transforms = ArrayPrototypeSlice(args, 0, -1);
     options = last;
   } else {
-    transforms = args;
+    transforms = ArrayPrototypeSlice(args);
+    transforms[transforms.length - 1] = lastTransform;
     options = undefined;
   }
 
   for (let i = 0; i < transforms.length; i++) {
-    if (!isTransform(transforms[i])) {
+    const transform = snapshotTransform(transforms[i]);
+    if (transform === undefined) {
       throw new ERR_INVALID_ARG_TYPE(
         `transforms[${i}]`, ['Function', 'Object with transform()'],
         transforms[i]);
     }
+    transforms[i] = transform;
   }
 
   return { __proto__: null, transforms, options };
@@ -421,6 +456,7 @@ module.exports = {
   concatBytes,
   convertChunks,
   createBatchEntry,
+  getProtocolMethod,
   getWriterSignal,
   getMinCursor,
   hasProtocol,
@@ -428,6 +464,7 @@ module.exports = {
   isTransformObject,
   onSignalAbort,
   parsePullArgs,
+  snapshotTransform,
   toUint8Array,
   toWriterUint8Array,
   validateBackpressure,
diff --git a/test/parallel/test-stream-iter-broadcast-basic.js b/test/parallel/test-stream-iter-broadcast-basic.js
index 3dbd3ce97512..4135aea6d97d 100644
--- a/test/parallel/test-stream-iter-broadcast-basic.js
+++ b/test/parallel/test-stream-iter-broadcast-basic.js
@@ -350,6 +350,18 @@ async function testLateJoinerSeesBufferedData() {
   assert.strictEqual(result, 'before-join');
 }
 
+async function testLateJoinerAfterDetachSeesBufferedData() {
+  const { writer, broadcast: bc } = broadcast({ budget: 16384 });
+  const first = bc.push()[Symbol.asyncIterator]();
+
+  writer.writeSync('before-detach');
+  await first.return();
+
+  const second = bc.push();
+  writer.endSync();
+  assert.strictEqual(await text(second), 'before-detach');
+}
+
 async function testOverlappingNextKeepsEarlierRead() {
   const { writer, broadcast: bc } = broadcast();
   const it = bc.push()[Symbol.asyncIterator]();
@@ -403,5 +415,6 @@ Promise.all([
   testFailDetachesConsumers(),
   testWriterFailIdempotent(),
   testLateJoinerSeesBufferedData(),
+  testLateJoinerAfterDetachSeesBufferedData(),
   testOverlappingNextKeepsEarlierRead(),
 ]).then(common.mustCall());
diff --git a/test/parallel/test-stream-iter-broadcast-from.js b/test/parallel/test-stream-iter-broadcast-from.js
index 928d4d472f06..cd76f3647b2b 100644
--- a/test/parallel/test-stream-iter-broadcast-from.js
+++ b/test/parallel/test-stream-iter-broadcast-from.js
@@ -4,6 +4,7 @@
 const common = require('../common');
 const assert = require('assert');
 const { broadcast, Broadcast, from, text } = require('stream/iter');
+const { setImmediate } = require('timers/promises');
 
 // =============================================================================
 // Broadcast.from
@@ -117,34 +118,46 @@ async function testAlreadyAbortedSignal() {
 // =============================================================================
 
 async function testBroadcastFromCancelWhileBlocked() {
-  // Create a slow async source that blocks between yields
-  let sourceFinished = false;
-  async function* slowSource() {
-    const enc = new TextEncoder();
-    yield [enc.encode('chunk1')];
-    // Simulate a long delay without keeping the cancelled source alive.
-    await new Promise((resolve) => setTimeout(resolve, 10000).unref());
-    yield [enc.encode('chunk2')];
-    sourceFinished = true;
-  }
-
-  const { broadcast: bc } = Broadcast.from(slowSource());
-  const consumer = bc.push();
+  let resolveNext;
+  let sourceReturned = false;
+  const source = {
+    [Symbol.asyncIterator]() {
+      return {
+        next() {
+          const { promise, resolve } = Promise.withResolvers();
+          resolveNext = resolve;
+          return promise;
+        },
+        return() {
+          sourceReturned = true;
+          return Promise.resolve({ __proto__: null, done: true });
+        },
+      };
+    },
+  };
 
-  // Read the first chunk
-  const iter = consumer[Symbol.asyncIterator]();
-  const first = await iter.next();
-  assert.strictEqual(first.done, false);
+  const { writer, broadcast: bc } = Broadcast.from(source);
+  const iter = bc.push()[Symbol.asyncIterator]();
+  const pendingRead = iter.next();
+  await setImmediate();
 
-  // Cancel while the source is blocked waiting to yield the next chunk
+  let writesAfterCancel = 0;
+  writer.writevSync = () => { writesAfterCancel++; return true; };
   bc.cancel();
-
-  // The iteration should complete (not hang)
-  const next = await iter.next();
-  assert.strictEqual(next.done, true);
-
-  // Source should NOT have finished (we cancelled before chunk2)
-  assert.strictEqual(sourceFinished, false);
+  assert.deepStrictEqual(await pendingRead, {
+    __proto__: null,
+    done: true,
+    value: undefined,
+  });
+
+  resolveNext({
+    __proto__: null,
+    done: false,
+    value: [new TextEncoder().encode('late')],
+  });
+  await setImmediate();
+  assert.strictEqual(writesAfterCancel, 0);
+  assert.strictEqual(sourceReturned, true);
 }
 
 // =============================================================================
@@ -168,6 +181,15 @@ async function testBroadcastFromSourceError() {
 // Protocol validation
 // =============================================================================
 
+function testBroadcastProtocolReturnsBroadcast() {
+  const { broadcast: expected } = broadcast();
+  const obj = {
+    [Symbol.for('Stream.broadcastProtocol')]() { return expected; },
+  };
+  assert.strictEqual(Broadcast.from(obj), expected);
+  expected.cancel();
+}
+
 function testBroadcastProtocolReturnsNull() {
   const obj = {
     [Symbol.for('Stream.broadcastProtocol')]() { return null; },
@@ -210,6 +232,7 @@ Promise.all([
   testAlreadyAbortedSignal(),
   testBroadcastFromCancelWhileBlocked(),
   testBroadcastFromSourceError(),
+  testBroadcastProtocolReturnsBroadcast(),
   testBroadcastProtocolReturnsNull(),
   testBroadcastProtocolReturnsString(),
   testBroadcastProtocolReturnsUndefined(),
diff --git a/test/parallel/test-stream-iter-broadcast-waiter-cleanup.js b/test/parallel/test-stream-iter-broadcast-waiter-cleanup.js
new file mode 100644
index 000000000000..4e0f6dd4d47a
--- /dev/null
+++ b/test/parallel/test-stream-iter-broadcast-waiter-cleanup.js
@@ -0,0 +1,34 @@
+// Flags: --experimental-stream-iter --expose-gc
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const { broadcast } = require('stream/iter');
+
+async function detachConsumers(shared, count) {
+  for (let i = 0; i < count; i++) {
+    const iterator = shared.push()[Symbol.asyncIterator]();
+    const pending = iterator.next();
+    await iterator.return();
+    await pending;
+  }
+}
+
+async function testDetachedWaitersAreReleased() {
+  const { broadcast: shared } = broadcast();
+
+  await detachConsumers(shared, 100);
+  global.gc();
+  const before = process.memoryUsage().heapUsed;
+
+  await detachConsumers(shared, 50_000);
+  global.gc();
+  const retained = process.memoryUsage().heapUsed - before;
+
+  assert.ok(retained < 8 * 1024 * 1024,
+            `Detached Broadcast waiters retained ${retained} bytes`);
+  assert.strictEqual(shared.consumerCount, 0);
+  shared.cancel();
+}
+
+testDetachedWaitersAreReleased().then(common.mustCall());
diff --git a/test/parallel/test-stream-iter-factory-signal.js b/test/parallel/test-stream-iter-factory-signal.js
new file mode 100644
index 000000000000..7d0ac92b1254
--- /dev/null
+++ b/test/parallel/test-stream-iter-factory-signal.js
@@ -0,0 +1,84 @@
+// Flags: --experimental-stream-iter
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const { getEventListeners } = require('events');
+const {
+  broadcast,
+  duplex,
+  push,
+  share,
+  text,
+} = require('stream/iter');
+
+function abortListenerCount(signal) {
+  return getEventListeners(signal, 'abort').length;
+}
+
+async function testPushSignalLifetime() {
+  const controller = new AbortController();
+  const { writer, readable } = push({ signal: controller.signal });
+  const iterator = readable[Symbol.asyncIterator]();
+
+  assert.strictEqual(abortListenerCount(controller.signal), 1);
+  assert.strictEqual(writer.writeSync('x'), true);
+  const ending = writer.end();
+  assert.strictEqual(abortListenerCount(controller.signal), 1);
+
+  assert.strictEqual((await iterator.next()).done, false);
+  assert.strictEqual((await iterator.next()).done, true);
+  assert.strictEqual(await ending, 1);
+  assert.strictEqual(abortListenerCount(controller.signal), 0);
+}
+
+async function testBroadcastSignalLifetime() {
+  const controller = new AbortController();
+  const { writer, broadcast: shared } = broadcast({
+    signal: controller.signal,
+  });
+  const iterator = shared.push()[Symbol.asyncIterator]();
+
+  assert.strictEqual(abortListenerCount(controller.signal), 1);
+  assert.strictEqual(writer.writeSync('x'), true);
+  const ending = writer.end();
+  assert.strictEqual(abortListenerCount(controller.signal), 1);
+
+  assert.strictEqual((await iterator.next()).done, false);
+  assert.strictEqual((await iterator.next()).done, true);
+  assert.strictEqual(await ending, 1);
+  assert.strictEqual(abortListenerCount(controller.signal), 0);
+}
+
+async function testShareSignalLifetime() {
+  const controller = new AbortController();
+  const shared = share('x', { signal: controller.signal });
+  const iterator = shared.pull()[Symbol.asyncIterator]();
+
+  assert.strictEqual(abortListenerCount(controller.signal), 1);
+  assert.strictEqual((await iterator.next()).done, false);
+  assert.strictEqual(abortListenerCount(controller.signal), 1);
+  assert.strictEqual((await iterator.next()).done, true);
+  assert.strictEqual(abortListenerCount(controller.signal), 0);
+}
+
+async function testDuplexSignalLifetime() {
+  const controller = new AbortController();
+  const [channelA, channelB] = duplex({ signal: controller.signal });
+
+  assert.strictEqual(abortListenerCount(controller.signal), 1);
+  await channelA.writer.write('x');
+  const closing = channelA.close();
+  assert.strictEqual(abortListenerCount(controller.signal), 1);
+
+  assert.strictEqual(await text(channelB.readable), 'x');
+  await closing;
+  assert.strictEqual(abortListenerCount(controller.signal), 0);
+}
+
+Promise.all([
+  testPushSignalLifetime(),
+  testBroadcastSignalLifetime(),
+  testShareSignalLifetime(),
+  testDuplexSignalLifetime(),
+]).then(common.mustCall());
diff --git a/test/parallel/test-stream-iter-from-async.js b/test/parallel/test-stream-iter-from-async.js
index a4726d633ef0..29ecdf425767 100644
--- a/test/parallel/test-stream-iter-from-async.js
+++ b/test/parallel/test-stream-iter-from-async.js
@@ -3,7 +3,8 @@
 
 const common = require('../common');
 const assert = require('assert');
-const { from, text, Stream } = require('stream/iter');
+const { bytes, from, text, Stream } = require('stream/iter');
+const { setImmediate } = require('timers/promises');
 
 async function testFromString() {
   const readable = from('hello-async');
@@ -31,6 +32,55 @@ async function testFromAsyncGenerator() {
   assert.deepStrictEqual(batches[1][0], new Uint8Array([30, 40]));
 }
 
+async function testFromAsyncIteratorResultShapes() {
+  const wrappers = [
+    (result) => result,
+    (result) => ({
+      then(resolve) {
+        resolve(result);
+      },
+    }),
+  ];
+
+  for (const wrap of wrappers) {
+    let done = false;
+    const source = {
+      [Symbol.asyncIterator]() {
+        return {
+          next() {
+            if (done) return wrap({ done: true });
+            done = true;
+            return wrap({ done: false, value: 'data' });
+          },
+        };
+      },
+    };
+
+    assert.strictEqual(await text(from(source)), 'data');
+  }
+}
+
+async function testFromSourceErrorDoesNotWaitForReturn() {
+  const reason = new Error('source failed');
+  const source = {
+    [Symbol.asyncIterator]() {
+      return {
+        next() {
+          return Promise.reject(reason);
+        },
+        return() {
+          return new Promise(() => {});
+        },
+      };
+    },
+  };
+
+  await assert.rejects(
+    from(source).next(),
+    (error) => error === reason,
+  );
+}
+
 async function testFromBoundsNestedAsyncIterable() {
   let nestedClosed = false;
   async function* nested() {
@@ -266,13 +316,126 @@ async function testFromHandlesProtocolRejectionUntilIteration() {
       () => Promise.reject(reason)),
   });
 
-  await new Promise(setImmediate);
+  await setImmediate();
   await assert.rejects(
     iterable[Symbol.asyncIterator]().next(),
     (error) => error === reason,
   );
 }
 
+async function testFromReturnCancelsPendingPromises() {
+  const toAsyncStreamable = Symbol.for('Stream.toAsyncStreamable');
+  const createSources = [
+    (promise) => from([promise]),
+    (promise) => from({
+      [Symbol.asyncIterator]() {
+        let done = false;
+        return {
+          next() {
+            if (done) return { done: true };
+            done = true;
+            return { done: false, value: promise };
+          },
+        };
+      },
+    }),
+    (promise) => from({
+      [toAsyncStreamable]() {
+        return promise;
+      },
+    }),
+  ];
+
+  for (const createSource of createSources) {
+    const deferred = Promise.withResolvers();
+    const iterator = createSource(deferred.promise)[Symbol.asyncIterator]();
+    const read = iterator.next();
+    await setImmediate();
+
+    const rejected = assert.rejects(read, { name: 'AbortError' });
+    const closed = iterator.return();
+    const [, result] = await Promise.all([rejected, closed]);
+    assert.strictEqual(result.done, true);
+    deferred.resolve('late value');
+  }
+}
+
+function createPendingNestedSource(
+  returnResult = () => ({ done: true })) {
+  const started = Promise.withResolvers();
+  const pending = Promise.withResolvers();
+  let returned = false;
+  const nested = {
+    [Symbol.asyncIterator]() {
+      return {
+        next() {
+          started.resolve();
+          return pending.promise;
+        },
+        return() {
+          returned = true;
+          return returnResult();
+        },
+      };
+    },
+  };
+
+  async function* source() {
+    yield nested;
+  }
+
+  return {
+    source: source(),
+    started: started.promise,
+    resolve: pending.resolve,
+    wasReturned() {
+      return returned;
+    },
+  };
+}
+
+async function testFromReturnClosesPendingNestedIterator() {
+  const fixture = createPendingNestedSource();
+  const iterator = from(fixture.source)[Symbol.asyncIterator]();
+  const read = iterator.next();
+  await fixture.started;
+
+  const rejected = assert.rejects(read, { name: 'AbortError' });
+  const closed = iterator.return();
+  await Promise.all([rejected, closed]);
+  assert.strictEqual(fixture.wasReturned(), true);
+  fixture.resolve({ done: true });
+}
+
+async function testConsumerAbortClosesPendingNestedIterator() {
+  const fixture = createPendingNestedSource();
+  const controller = new AbortController();
+  const reason = new Error('consumer cancelled');
+  const consumed = bytes(fixture.source, { signal: controller.signal });
+  await fixture.started;
+
+  const rejected = assert.rejects(consumed, (error) => error === reason);
+  controller.abort(reason);
+  await rejected;
+  await setImmediate();
+  assert.strictEqual(fixture.wasReturned(), true);
+  fixture.resolve({ done: true });
+}
+
+async function testFromCancellationHandlesCleanupRejection() {
+  const fixture = createPendingNestedSource(
+    () => Promise.reject(new Error('cleanup failed')));
+  const iterator = from(fixture.source)[Symbol.asyncIterator]();
+  const read = iterator.next();
+  await fixture.started;
+
+  const rejected = assert.rejects(read, { name: 'AbortError' });
+  await Promise.all([rejected, iterator.return()]);
+  await setImmediate();
+  assert.strictEqual(fixture.wasReturned(), true);
+  fixture.resolve({ done: true });
+}
+
 // DataView input should be converted to Uint8Array (zero-copy)
 async function testFromDataView() {
   const buf = new ArrayBuffer(5);
@@ -298,6 +461,8 @@ function testFromUndefinedThrows() {
 Promise.all([
   testFromString(),
   testFromAsyncGenerator(),
+  testFromAsyncIteratorResultShapes(),
+  testFromSourceErrorDoesNotWaitForReturn(),
   testFromBoundsNestedAsyncIterable(),
   testFromSyncIterableAsAsync(),
   testFromSyncIterableAwaitsPromiseValues(),
@@ -319,5 +484,9 @@ Promise.all([
   testFromTopLevelAsyncPrecedence(),
   testFromTopLevelProtocolOverIterator(),
   testFromHandlesProtocolRejectionUntilIteration(),
+  testFromReturnCancelsPendingPromises(),
+  testFromReturnClosesPendingNestedIterator(),
+  testConsumerAbortClosesPendingNestedIterator(),
+  testFromCancellationHandlesCleanupRejection(),
   testFromDataView(),
 ]).then(common.mustCall());
diff --git a/test/parallel/test-stream-iter-property-access.js b/test/parallel/test-stream-iter-property-access.js
new file mode 100644
index 000000000000..7c4b8ef3a7e7
--- /dev/null
+++ b/test/parallel/test-stream-iter-property-access.js
@@ -0,0 +1,178 @@
+// Flags: --experimental-stream-iter
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const { Readable } = require('stream');
+const {
+  Broadcast,
+  Share,
+  SyncShare,
+  broadcast,
+  broadcastProtocol,
+  bytes,
+  bytesSync,
+  drainableProtocol,
+  from,
+  fromSync,
+  ondrain,
+  pull,
+  pullSync,
+  share,
+  shareProtocol,
+  shareSync,
+  shareSyncProtocol,
+  text,
+  textSync,
+  toAsyncStreamable,
+  toStreamable,
+} = require('stream/iter');
+
+function protocolFixture(symbol, result, input = {}) {
+  let accesses = 0;
+  const method = common.mustCall(function() {
+    assert.strictEqual(this, input);
+    return result;
+  });
+  Object.defineProperty(input, symbol, {
+    get() {
+      accesses++;
+      if (accesses > 1) throw new Error('protocol method read twice');
+      return method;
+    },
+  });
+  return { input, get accesses() { return accesses; } };
+}
+
+function statefulTransformFixture() {
+  let accesses = 0;
+  const transform = {};
+  const method = common.mustCall(function(source) {
+    assert.strictEqual(this, transform);
+    return source;
+  });
+  Object.defineProperty(transform, 'transform', {
+    get() {
+      accesses++;
+      if (accesses > 1) throw new Error('transform method read twice');
+      return method;
+    },
+  });
+  return { transform, get accesses() { return accesses; } };
+}
+
+async function testFromSnapshotsProtocolMethods() {
+  for (const symbol of [toAsyncStreamable, toStreamable]) {
+    const fixture = protocolFixture(symbol, 'abc');
+    assert.deepStrictEqual(await bytes(from(fixture.input)),
+                           new Uint8Array([97, 98, 99]));
+    assert.strictEqual(fixture.accesses, 1);
+  }
+
+  const fixture = protocolFixture(toAsyncStreamable, 'nested');
+  async function* source() {
+    yield fixture.input;
+  }
+  assert.deepStrictEqual(await bytes(from(source())),
+                         new Uint8Array([110, 101, 115, 116, 101, 100]));
+  assert.strictEqual(fixture.accesses, 1);
+}
+
+async function testFromSyncSnapshotsProtocolMethods() {
+  const fixture = protocolFixture(toStreamable, 'abc');
+  assert.deepStrictEqual(bytesSync(fromSync(fixture.input)),
+                         new Uint8Array([97, 98, 99]));
+  assert.strictEqual(fixture.accesses, 1);
+}
+
+async function testArrayFastPathsHonorProtocols() {
+  const asyncInputs = [
+    protocolFixture(toAsyncStreamable, 'empty-async', []),
+    protocolFixture(toStreamable, 'batch-async', [new Uint8Array([0])]),
+  ];
+  for (const fixture of asyncInputs) {
+    assert.match(await text(from(fixture.input)), /-async$/);
+    assert.strictEqual(fixture.accesses, 1);
+  }
+
+  const syncInputs = [
+    protocolFixture(toStreamable, 'empty-sync', []),
+    protocolFixture(toStreamable, 'batch-sync', [new Uint8Array([0])]),
+  ];
+  for (const fixture of syncInputs) {
+    assert.match(textSync(fromSync(fixture.input)), /-sync$/);
+    assert.strictEqual(fixture.accesses, 1);
+  }
+}
+
+async function testValidatedSourceHonorsProtocol() {
+  const readable = Readable.from(['ignored']);
+  const validated = readable[toAsyncStreamable]();
+  assert.strictEqual(from(validated), validated);
+
+  const fixture = protocolFixture(
+    toAsyncStreamable, 'validated-protocol', validated);
+  assert.strictEqual(await text(from(fixture.input)), 'validated-protocol');
+  assert.strictEqual(fixture.accesses, 1);
+  readable.destroy();
+}
+
+async function testPullSnapshotsStatefulTransform() {
+  const fixture = statefulTransformFixture();
+  const controller = new AbortController();
+  const readable = pull('abc', fixture.transform, {
+    signal: controller.signal,
+  });
+
+  assert.deepStrictEqual(await bytes(readable),
+                         new Uint8Array([97, 98, 99]));
+  assert.strictEqual(fixture.accesses, 1);
+}
+
+async function testPullSyncSnapshotsStatefulTransform() {
+  const fixture = statefulTransformFixture();
+  assert.deepStrictEqual(bytesSync(pullSync('abc', fixture.transform)),
+                         new Uint8Array([97, 98, 99]));
+  assert.strictEqual(fixture.accesses, 1);
+}
+
+async function testMultiConsumerProtocolsSnapshotMethods() {
+  const broadcastTarget = broadcast().broadcast;
+  const broadcastFixture = protocolFixture(
+    broadcastProtocol, broadcastTarget);
+  assert.strictEqual(Broadcast.from(broadcastFixture.input), broadcastTarget);
+  assert.strictEqual(broadcastFixture.accesses, 1);
+
+  const shareTarget = share('abc');
+  const shareFixture = protocolFixture(shareProtocol, shareTarget);
+  assert.strictEqual(Share.from(shareFixture.input), shareTarget);
+  assert.strictEqual(shareFixture.accesses, 1);
+
+  const syncShareTarget = shareSync('abc');
+  const syncShareFixture = protocolFixture(
+    shareSyncProtocol, syncShareTarget);
+  assert.strictEqual(SyncShare.fromSync(syncShareFixture.input),
+                     syncShareTarget);
+  assert.strictEqual(syncShareFixture.accesses, 1);
+
+  broadcastTarget.cancel();
+  shareTarget.cancel();
+  syncShareTarget.cancel();
+}
+
+async function testDrainableProtocolSnapshotMethod() {
+  const fixture = protocolFixture(drainableProtocol, true);
+  assert.strictEqual(await ondrain(fixture.input), true);
+  assert.strictEqual(fixture.accesses, 1);
+}
+
+Promise.all([
+  testFromSnapshotsProtocolMethods(),
+  testFromSyncSnapshotsProtocolMethods(),
+  testArrayFastPathsHonorProtocols(),
+  testValidatedSourceHonorsProtocol(),
+  testPullSnapshotsStatefulTransform(),
+  testPullSyncSnapshotsStatefulTransform(),
+  testMultiConsumerProtocolsSnapshotMethods(),
+  testDrainableProtocolSnapshotMethod(),
+]).then(common.mustCall());
diff --git a/test/parallel/test-stream-iter-share-from.js b/test/parallel/test-stream-iter-share-from.js
index 806e30876302..69b782ce5608 100644
--- a/test/parallel/test-stream-iter-share-from.js
+++ b/test/parallel/test-stream-iter-share-from.js
@@ -221,23 +221,29 @@ async function testShareDropNewest() {
 // =============================================================================
 
 async function testShareStrictBackpressure() {
-  async function* source() {
-    for (let i = 0; i < 10; i++) {
-      yield [new Uint8Array(16384)];
+  for (const transformed of [false, true]) {
+    async function* source() {
+      for (let i = 0; i < 10; i++) {
+        yield [new Uint8Array(16384)];
+      }
     }
+    const shared = share(source(), {
+      budget: 32768,
+      backpressure: 'strict',
+    });
+    const consumer = transformed ?
+      shared.pull((chunks) => chunks) : shared.pull();
+    const fast = consumer[Symbol.asyncIterator]();
+    // This consumer prevents the buffer from being trimmed.
+    shared.pull();
+
+    await fast.next();
+    await fast.next();
+    await assert.rejects(fast.next(), { code: 'ERR_OUT_OF_RANGE' });
+    assert.strictEqual(shared.consumerCount, 1);
+    assert.strictEqual((await fast.next()).done, true);
+    shared.cancel();
   }
-  const shared = share(source(), { budget: 32768, backpressure: 'strict' });
-  const fast = shared.pull();
-  // Create a second consumer that never reads — this prevents buffer trimming
-  shared.pull();
-
-  // The fast consumer's pulls will eventually cause the buffer to exceed
-  // the budget (since the slow consumer prevents trimming),
-  // triggering an ERR_OUT_OF_RANGE error.
-  await assert.rejects(async () => {
-    // eslint-disable-next-line no-unused-vars
-    for await (const _ of fast) { /* consume */ }
-  }, { code: 'ERR_OUT_OF_RANGE' });
 }
 
 Promise.all([
diff --git a/test/parallel/test-stream-iter-share-sync.js b/test/parallel/test-stream-iter-share-sync.js
index 20c98d134206..627508df0a39 100644
--- a/test/parallel/test-stream-iter-share-sync.js
+++ b/test/parallel/test-stream-iter-share-sync.js
@@ -139,6 +139,39 @@ function testShareSyncSourceError() {
   }, { message: 'sync share boom' });
 }
 
+function testShareSyncRejectsUnbounded() {
+  assert.throws(
+    () => shareSync(fromSync('data'), { backpressure: 'unbounded' }),
+    { code: 'ERR_INVALID_ARG_VALUE' },
+  );
+}
+
+function testShareSyncDropNewest() {
+  let pulls = 0;
+  function* source() {
+    for (let i = 0; i < 3; i++) {
+      pulls++;
+      const chunk = new Uint8Array(16384);
+      chunk[0] = i;
+      yield [chunk];
+    }
+  }
+
+  const shared = shareSync(source(), {
+    budget: 16384,
+    backpressure: 'drop-newest',
+  });
+  const fast = shared.pull()[Symbol.iterator]();
+  const slow = shared.pull()[Symbol.iterator]();
+
+  assert.strictEqual(fast.next().value[0][0], 0);
+  assert.strictEqual(fast.next().done, true);
+  assert.strictEqual(pulls, 3);
+
+  assert.strictEqual(slow.next().value[0][0], 0);
+  assert.strictEqual(slow.next().done, true);
+}
+
 // shareSync() accepts string source directly (normalized via fromSync())
 function testShareSyncStringSource() {
   const shared = shareSync('hello-sync-share');
@@ -154,5 +187,7 @@ Promise.all([
   testShareSyncCancelWithReason(),
   testShareSyncCancelWithFalsyReason(),
   testShareSyncSourceError(),
+  testShareSyncRejectsUnbounded(),
+  testShareSyncDropNewest(),
   testShareSyncStringSource(),
 ]).then(common.mustCall());