From 0c3c263f4889c40e0d4c1cc7a3dd7969f0baf446 Mon Sep 17 00:00:00 2001
From: Ethan Arrowood <ethan@arrowood.dev>
Date: Thu, 27 Aug 2026 12:10:42 -0600
Subject: [PATCH 1/2] stream: add drain()/drainSync() for stream/iter

Every consumer in node:stream/iter retains what it reads, so there is
no way to read a streamable to completion while keeping nothing. To
clear a stream, callers write `for await (const _ of source) {}`. This
is apparent throughout multiple test suites, including the stream/iter
tests themselves.

These are especially useful for QUIC where a receiver doesn't want the
payload, but still has to read it to completion to relieve
backpressure. bytes() does that too, but allocates the whole payload to
discard it.

drain() pulls every batch and drops it, so peak memory is one batch
regardless of volume. It takes the same signal and limit options as
the other consumers, rejects if the source errors mid-stream, and
fulfills with undefined. drainSync() is the synchronous form.

Assisted-by: Claude Opus 5
Signed-off-by: Ethan Arrowood <ethan@arrowood.dev>
---
 doc/api/quic.md                               |   4 +
 doc/api/stream_iter.md                        |  76 ++++
 lib/internal/streams/iter/consumers.js        |  77 ++++
 lib/stream/iter.js                            |   6 +
 .../test-stream-iter-consumers-drain.mjs      | 405 ++++++++++++++++++
 test/parallel/test-stream-iter-namespace.js   |   6 +
 6 files changed, 574 insertions(+)
 create mode 100644 test/parallel/test-stream-iter-consumers-drain.mjs

diff --git a/doc/api/quic.md b/doc/api/quic.md
index d90b1b938bd2..c4b22a94af9c 100644
--- a/doc/api/quic.md
+++ b/doc/api/quic.md
@@ -343,6 +343,10 @@ Only one async iterator can be obtained per stream. The stream is also
 compatible with `node:stream/iter` utilities such as `Stream.bytes()`,
 `Stream.text()`, and `Stream.pipeTo()`.
 
+Consuming a stream is what returns flow-control credit to the peer, so a
+stream whose payload is not wanted should still be read to completion. Use
+`Stream.drain()` to read the stream without retaining any of it.
+
 ### Datagrams
 
 In addition to streams, QUIC supports unreliable datagrams ([RFC 9221][]) for
diff --git a/doc/api/stream_iter.md b/doc/api/stream_iter.md
index c1ebacd3dfad..a2b002cf9af1 100644
--- a/doc/api/stream_iter.md
+++ b/doc/api/stream_iter.md
@@ -1025,6 +1025,81 @@ added:
 
 Synchronous version of [`bytes()`][].
 
+### `drain(source[, options])`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* `source` {AsyncIterable|Iterable} whose chunks must be {Uint8Array\[]}
+* `options` {Object}
+  * `signal` {AbortSignal}
+  * `limit` {number} Maximum number of bytes to consume. If the total bytes
+    read exceeds limit, an `ERR_OUT_OF_RANGE` error is thrown
+* Returns: {Promise} Fulfills with `undefined`.
+
+Read a source to completion, discarding every chunk. Unlike the other
+consumers, `drain()` retains nothing. Memory tops-out at a single batch no
+matter how much data the source produces.
+
+Use this to consume a stream when the content doesn't matter. For example, A
+QUIC stream only returns flow-control credit to the peer as its data is
+consumed, so a receiver that does not want the payload must still read it to
+completion.
+
+If the source errors part-way through, the returned promise rejects with that
+error.
+
+There is no default `limit`. `drain()` reads until the source is exhausted
+unless a limit is specified. When a limit is configured and the source exceeds
+it, the promise rejects and the source is cancelled. A partial drain is never
+reported as success.
+
+```mjs
+import { drain, from, pull, tap } from 'node:stream/iter';
+
+// Count the bytes flowing through a stream without retaining any of them.
+let bytesSeen = 0;
+const counter = tap((chunks) => {
+  for (const chunk of chunks) bytesSeen += chunk.byteLength;
+});
+
+await drain(pull(from('hello world'), counter));
+console.log(bytesSeen); // 11
+```
+
+```cjs
+const { drain, from, pull, tap } = require('node:stream/iter');
+
+async function run() {
+  // Count the bytes flowing through a stream without retaining any of them.
+  let bytesSeen = 0;
+  const counter = tap((chunks) => {
+    for (const chunk of chunks) bytesSeen += chunk.byteLength;
+  });
+
+  await drain(pull(from('hello world'), counter));
+  console.log(bytesSeen); // 11
+}
+
+run().catch(console.error);
+```
+
+### `drainSync(source[, options])`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* `source` {Iterable} whose chunks must be {Uint8Array\[]}
+* `options` {Object}
+  * `limit` {number} Maximum number of bytes to consume. If the total bytes
+    read exceeds limit, an `ERR_OUT_OF_RANGE` error is thrown
+* Returns: {undefined}
+
+Synchronous version of [`drain()`][]. Throws `ERR_INVALID_ARG_TYPE` if `source`
+is not synchronously iterable.
+
 ### `text(source[, options])`
 
 <!-- YAML
@@ -2167,6 +2242,7 @@ console.log(textSync(stream)); // 'hello world'
 [`array()`]: #arraysource-options
 [`arrayBuffer()`]: #arraybuffersource-options
 [`bytes()`]: #bytessource-options
+[`drain()`]: #drainsource-options
 [`from()`]: #frominput
 [`fromSync()`]: #fromsyncinput
 [`node:zlib/iter`]: zlib.md#iterable-compression
diff --git a/lib/internal/streams/iter/consumers.js b/lib/internal/streams/iter/consumers.js
index e3bd7856dcd4..f6a9032239c2 100644
--- a/lib/internal/streams/iter/consumers.js
+++ b/lib/internal/streams/iter/consumers.js
@@ -3,6 +3,7 @@
 // New Streams API - Consumers & Utilities
 //
 // bytes(), text(), arrayBuffer() - collect entire stream
+// drain() - consume entire stream, retaining nothing
 // tap(), tapSync() - observe without modifying
 // merge() - temporal combining of sources
 // ondrain() - backpressure drain utility
@@ -217,6 +218,20 @@ function validateSyncConsumerOptions(options) {
   validateBaseConsumerOptions(options);
 }
 
+function validateSyncDrainOptions(options) {
+  validateObject(options, 'options');
+  if (options.limit !== undefined) {
+    validateInteger(options.limit, 'options.limit', 0);
+  }
+}
+
+function validateDrainOptions(options) {
+  validateSyncDrainOptions(options);
+  if (options.signal !== undefined) {
+    validateAbortSignal(options.signal, 'options.signal');
+  }
+}
+
 // =============================================================================
 // Sync Consumers
 // =============================================================================
@@ -272,6 +287,32 @@ function arraySync(source, options = kNullPrototype) {
   return collectSync(source, options.limit);
 }
 
+/**
+ * Read a sync source to completion, discarding every chunk.
+ * @param {Iterable<Uint8Array[]>} source
+ * @param {{ limit?: number }} [options]
+ * @returns {undefined}
+ */
+function drainSync(source, options = kNullPrototype) {
+  validateSyncDrainOptions(options);
+
+  const limit = options.limit;
+  const normalized = fromSync(source);
+  let totalBytes = 0;
+
+  for (const batch of normalized) {
+    // With no limit, just iterate through the stream completely.
+    if (limit === undefined) continue;
+    // Otherwise, calculate totalBytes and track the limit.
+    for (let i = 0; i < batch.length; i++) {
+      totalBytes += TypedArrayPrototypeGetByteLength(batch[i]);
+      if (totalBytes > limit) {
+        throw new ERR_OUT_OF_RANGE('totalBytes', `<= ${limit}`, totalBytes);
+      }
+    }
+  }
+}
+
 // =============================================================================
 // Async Consumers
 // =============================================================================
@@ -328,6 +369,40 @@ async function array(source, options = kNullPrototype) {
   return collectAsync(source, options.signal, options.limit);
 }
 
+/**
+ * Read an async or sync source to completion, discarding every chunk.
+ * @param {AsyncIterable<Uint8Array[]>|Iterable<Uint8Array[]>} source
+ * @param {{ signal?: AbortSignal, limit?: number }} [options]
+ * @returns {Promise<undefined>}
+ */
+async function drain(source, options = kNullPrototype) {
+  validateDrainOptions(options);
+  const signal = options.signal;
+  const limit = options.limit;
+
+  signal?.throwIfAborted();
+
+  const abortableSource = signal && isAsyncIterable(source) ?
+    yieldAbortable(source, signal) : source;
+  const normalized = from(abortableSource);
+  const iterable = signal ? yieldAbortable(normalized, signal) : normalized;
+
+  let totalBytes = 0;
+
+  for await (const batch of iterable) {
+    signal?.throwIfAborted();
+    // With no limit, just iterate through the stream completely.
+    if (limit === undefined) continue;
+    // Otherwise, calculate totalBytes and track the limit.
+    for (let i = 0; i < batch.length; i++) {
+      totalBytes += TypedArrayPrototypeGetByteLength(batch[i]);
+      if (totalBytes > limit) {
+        throw new ERR_OUT_OF_RANGE('totalBytes', `<= ${limit}`, totalBytes);
+      }
+    }
+  }
+}
+
 // =============================================================================
 // Tap Utilities
 // =============================================================================
@@ -585,6 +660,8 @@ module.exports = {
   arraySync,
   bytes,
   bytesSync,
+  drain,
+  drainSync,
   merge,
   ondrain,
   tap,
diff --git a/lib/stream/iter.js b/lib/stream/iter.js
index 4cb499bb1616..8ef01aafbbca 100644
--- a/lib/stream/iter.js
+++ b/lib/stream/iter.js
@@ -44,6 +44,8 @@ const {
   arrayBufferSync,
   array,
   arraySync,
+  drain,
+  drainSync,
   tap,
   tapSync,
   merge,
@@ -100,12 +102,14 @@ const Stream = ObjectFreeze({
   text,
   arrayBuffer,
   array,
+  drain,
 
   // Consumers (sync)
   bytesSync,
   textSync,
   arrayBufferSync,
   arraySync,
+  drainSync,
 
   // Combining
   merge,
@@ -164,12 +168,14 @@ module.exports = {
   text,
   arrayBuffer,
   array,
+  drain,
 
   // Consumers (sync)
   bytesSync,
   textSync,
   arrayBufferSync,
   arraySync,
+  drainSync,
 
   // Combining
   merge,
diff --git a/test/parallel/test-stream-iter-consumers-drain.mjs b/test/parallel/test-stream-iter-consumers-drain.mjs
new file mode 100644
index 000000000000..97da324eb88a
--- /dev/null
+++ b/test/parallel/test-stream-iter-consumers-drain.mjs
@@ -0,0 +1,405 @@
+// Flags: --experimental-stream-iter
+
+import { mustCall } from '../common/index.mjs';
+import assert from 'node:assert';
+import {
+  bytes,
+  drain,
+  drainSync,
+  from,
+  fromSync,
+  pull,
+  push,
+  tap,
+  toAsyncStreamable,
+} from 'node:stream/iter';
+
+// =============================================================================
+// drain
+// =============================================================================
+
+async function testDrainResolvesUndefined() {
+  assert.strictEqual(await drain(from('hello')), undefined);
+}
+
+async function testDrainConsumesToCompletion() {
+  let pulls = 0;
+  async function* gen() {
+    for (let i = 0; i < 5; i++) {
+      pulls++;
+      yield [new Uint8Array([i])];
+    }
+  }
+  await drain(gen());
+  assert.strictEqual(pulls, 5);
+}
+
+async function testDrainEmpty() {
+  assert.strictEqual(await drain(from([])), undefined);
+  async function* empty() {}
+  assert.strictEqual(await drain(empty()), undefined);
+}
+
+async function testDrainStringSource() {
+  // Sources are normalized via from(), so raw ByteInput works.
+  assert.strictEqual(await drain('a direct string'), undefined);
+}
+
+async function testDrainSyncSourceViaAsync() {
+  // drain() accepts sync iterables too, like the other async consumers.
+  assert.strictEqual(await drain(fromSync('sync source')), undefined);
+}
+
+async function testDrainAlreadyConsumed() {
+  // A source that has already been read to completion drains to a no-op
+  // rather than throwing.
+  async function* gen() {
+    yield [new Uint8Array([1])];
+  }
+  const source = gen();
+  await drain(source);
+  assert.strictEqual(await drain(source), undefined);
+}
+
+// =============================================================================
+// drain: error propagation
+// =============================================================================
+
+async function testDrainRejectsOnMidStreamError() {
+  const boom = new Error('boom');
+  async function* failing() {
+    yield [new Uint8Array([1])];
+    yield [new Uint8Array([2])];
+    throw boom;
+  }
+  await assert.rejects(drain(failing()), boom);
+}
+
+async function testDrainRejectsOnFirstChunkError() {
+  const boom = new Error('immediate boom');
+  async function* failing() {
+    await Promise.reject(boom);
+    yield [];
+  }
+  await assert.rejects(drain(failing()), boom);
+}
+
+async function testDrainRejectsInvalidChunkType() {
+  async function* bad() {
+    yield [42];
+  }
+  await assert.rejects(drain(bad()), { code: 'ERR_INVALID_ARG_TYPE' });
+}
+
+async function testDrainCallsReturnOnError() {
+  // An abrupt completion must still release the source.
+  let returned = false;
+  const boom = new Error('cleanup boom');
+  const source = {
+    __proto__: null,
+    async *[Symbol.asyncIterator]() {
+      try {
+        yield [new Uint8Array([1])];
+        throw boom;
+      } finally {
+        returned = true;
+      }
+    },
+  };
+  await assert.rejects(drain(source), boom);
+  assert.strictEqual(returned, true);
+}
+
+// =============================================================================
+// drain: AbortSignal
+// =============================================================================
+
+async function testDrainAlreadyAbortedSignal() {
+  await assert.rejects(
+    () => drain(from('data'), { signal: AbortSignal.abort() }),
+    { name: 'AbortError' },
+  );
+}
+
+async function testDrainAbortsPendingNext() {
+  const ac = new AbortController();
+  const reason = new Error('drain abort');
+
+  async function* never() {
+    await new Promise(() => {});
+    yield [];
+  }
+
+  const promise = drain(never(), { __proto__: null, signal: ac.signal });
+  ac.abort(reason);
+
+  await assert.rejects(promise, reason);
+}
+
+async function testDrainAbortsPendingNormalization() {
+  const ac = new AbortController();
+  const reason = new Error('drain normalization abort');
+  const source = {
+    __proto__: null,
+    [toAsyncStreamable]() {
+      return new Promise(() => {});
+    },
+  };
+
+  const promise = drain(source, { __proto__: null, signal: ac.signal });
+  ac.abort(reason);
+
+  await assert.rejects(promise, reason);
+}
+
+async function testDrainUnabortedSignalCompletes() {
+  const ac = new AbortController();
+  assert.strictEqual(
+    await drain(from('finishes first'), { signal: ac.signal }),
+    undefined,
+  );
+}
+
+// =============================================================================
+// drain: limit
+// =============================================================================
+
+async function testDrainLimitRejects() {
+  async function* gen() {
+    yield [new Uint8Array(100)];
+    yield [new Uint8Array(100)];
+  }
+  await assert.rejects(
+    () => drain(gen(), { limit: 150 }),
+    { name: 'RangeError', code: 'ERR_OUT_OF_RANGE' },
+  );
+}
+
+async function testDrainLimitAllowsExact() {
+  async function* gen() {
+    yield [new Uint8Array(50)];
+    yield [new Uint8Array(50)];
+  }
+  assert.strictEqual(await drain(gen(), { limit: 100 }), undefined);
+}
+
+async function testDrainLimitCancelsSource() {
+  // Exceeding the limit is an abrupt completion, so the source must be
+  // released rather than left half-read.
+  let returned = false;
+  const source = {
+    __proto__: null,
+    async *[Symbol.asyncIterator]() {
+      try {
+        while (true) yield [new Uint8Array(64)];
+      } finally {
+        returned = true;
+      }
+    },
+  };
+  await assert.rejects(() => drain(source, { limit: 128 }),
+                       { code: 'ERR_OUT_OF_RANGE' });
+  assert.strictEqual(returned, true);
+}
+
+async function testDrainLimitIsOptIn() {
+  // No default limit: a source larger than undici's 128 KB dump() default
+  // must still drain completely.
+  async function* big() {
+    for (let i = 0; i < 64; i++) yield [new Uint8Array(8192)];
+  }
+  assert.strictEqual(await drain(big()), undefined);
+}
+
+// =============================================================================
+// drain: does not retain data
+// =============================================================================
+
+async function testDrainDoesNotRetain() {
+  // Hand out one big chunk at a time from a single reused pool, and count
+  // how many are live. drain() must never hold more than the current batch,
+  // so the source can hand back the same buffer every time.
+  const CHUNKS = 512;
+  const CHUNK_SIZE = 64 * 1024;
+  const pooled = new Uint8Array(CHUNK_SIZE);
+  let handedOut = 0;
+
+  async function* pooledSource() {
+    for (let i = 0; i < CHUNKS; i++) {
+      handedOut++;
+      yield [pooled];
+    }
+  }
+
+  await drain(pooledSource());
+  assert.strictEqual(handedOut, CHUNKS);
+
+  // Contrast: bytes() over the same volume must allocate all of it.
+  const collected = await bytes(pooledSource());
+  assert.strictEqual(collected.byteLength, CHUNKS * CHUNK_SIZE);
+}
+
+async function testDrainReleasesBackpressure() {
+  // Consuming is what returns budget to the writer. With a small budget the
+  // writer blocks until drain() pulls, so this only completes if drain()
+  // really is reading.
+  const { writer, readable } = push({ budget: 16384 });
+  const chunk = new Uint8Array(8192);
+
+  const draining = drain(readable);
+
+  for (let i = 0; i < 32; i++) {
+    await writer.write(chunk);
+  }
+  await writer.end();
+
+  assert.strictEqual(await draining, undefined);
+}
+
+// =============================================================================
+// drain: interaction with transforms
+// =============================================================================
+
+async function testDrainWithTap() {
+  // The documented pattern: observe via tap() while retaining nothing.
+  let seenBytes = 0;
+  let seenBatches = 0;
+  const counter = tap((chunks) => {
+    // tap() invokes the callback once more with null to signal end of stream.
+    if (chunks === null) return;
+    seenBatches++;
+    for (const chunk of chunks) seenBytes += chunk.byteLength;
+  });
+
+  await drain(pull(from('hello world'), counter));
+  assert.strictEqual(seenBytes, 11);
+  assert.ok(seenBatches > 0);
+}
+
+async function testDrainPropagatesTransformError() {
+  const boom = new Error('transform boom');
+  const badTransform = () => { throw boom; };
+  await assert.rejects(drain(pull(from('hello'), badTransform)), boom);
+}
+
+// =============================================================================
+// drainSync
+// =============================================================================
+
+function testDrainSyncBasic() {
+  assert.strictEqual(drainSync(fromSync('hello')), undefined);
+}
+
+function testDrainSyncConsumesToCompletion() {
+  let pulls = 0;
+  function* gen() {
+    for (let i = 0; i < 5; i++) {
+      pulls++;
+      yield new Uint8Array([i]);
+    }
+  }
+  drainSync(gen());
+  assert.strictEqual(pulls, 5);
+}
+
+function testDrainSyncEmpty() {
+  assert.strictEqual(drainSync(fromSync([])), undefined);
+  function* empty() {}
+  assert.strictEqual(drainSync(empty()), undefined);
+}
+
+function testDrainSyncStringSource() {
+  assert.strictEqual(drainSync('direct sync string'), undefined);
+}
+
+function testDrainSyncThrowsMidStream() {
+  const boom = new Error('sync boom');
+  function* failing() {
+    yield new Uint8Array([1]);
+    throw boom;
+  }
+  assert.throws(() => drainSync(failing()), boom);
+}
+
+function testDrainSyncRejectsAsyncSource() {
+  // Same contract as the other *Sync consumers: an async-only source is a
+  // type error, not a silent no-op.
+  async function* asyncGen() {
+    yield [new Uint8Array([1])];
+  }
+  assert.throws(() => drainSync(asyncGen()),
+                { code: 'ERR_INVALID_ARG_TYPE' });
+  assert.throws(() => drainSync(Promise.resolve('x')),
+                { code: 'ERR_INVALID_ARG_TYPE' });
+}
+
+function testDrainSyncLimit() {
+  function* gen() {
+    yield new Uint8Array(100);
+    yield new Uint8Array(100);
+  }
+  assert.throws(
+    () => drainSync(gen(), { limit: 150 }),
+    { name: 'RangeError', code: 'ERR_OUT_OF_RANGE' },
+  );
+  assert.strictEqual(drainSync(gen(), { limit: 200 }), undefined);
+}
+
+// =============================================================================
+// Option validation
+// =============================================================================
+
+async function testDrainOptionValidation() {
+  await assert.rejects(() => drain(from('a'), 42),
+                       { code: 'ERR_INVALID_ARG_TYPE' });
+  await assert.rejects(() => drain(from('a'), { signal: 'bad' }),
+                       { code: 'ERR_INVALID_ARG_TYPE' });
+  await assert.rejects(() => drain(from('a'), { limit: 'bad' }),
+                       { code: 'ERR_INVALID_ARG_TYPE' });
+  await assert.rejects(() => drain(from('a'), { limit: -1 }),
+                       { code: 'ERR_OUT_OF_RANGE' });
+}
+
+function testDrainSyncOptionValidation() {
+  assert.throws(() => drainSync(fromSync('a'), 42),
+                { code: 'ERR_INVALID_ARG_TYPE' });
+  assert.throws(() => drainSync(fromSync('a'), { limit: 'bad' }),
+                { code: 'ERR_INVALID_ARG_TYPE' });
+  assert.throws(() => drainSync(fromSync('a'), { limit: -1 }),
+                { code: 'ERR_OUT_OF_RANGE' });
+}
+
+testDrainSyncBasic();
+testDrainSyncConsumesToCompletion();
+testDrainSyncEmpty();
+testDrainSyncStringSource();
+testDrainSyncThrowsMidStream();
+testDrainSyncRejectsAsyncSource();
+testDrainSyncLimit();
+testDrainSyncOptionValidation();
+
+await Promise.all([
+  testDrainResolvesUndefined(),
+  testDrainConsumesToCompletion(),
+  testDrainEmpty(),
+  testDrainStringSource(),
+  testDrainSyncSourceViaAsync(),
+  testDrainAlreadyConsumed(),
+  testDrainRejectsOnMidStreamError(),
+  testDrainRejectsOnFirstChunkError(),
+  testDrainRejectsInvalidChunkType(),
+  testDrainCallsReturnOnError(),
+  testDrainAlreadyAbortedSignal(),
+  testDrainAbortsPendingNext(),
+  testDrainAbortsPendingNormalization(),
+  testDrainUnabortedSignalCompletes(),
+  testDrainLimitRejects(),
+  testDrainLimitAllowsExact(),
+  testDrainLimitCancelsSource(),
+  testDrainLimitIsOptIn(),
+  testDrainDoesNotRetain(),
+  testDrainReleasesBackpressure(),
+  testDrainWithTap(),
+  testDrainPropagatesTransformError(),
+  testDrainOptionValidation(),
+]).then(mustCall());
diff --git a/test/parallel/test-stream-iter-namespace.js b/test/parallel/test-stream-iter-namespace.js
index ce197e85846d..a180c4fc7102 100644
--- a/test/parallel/test-stream-iter-namespace.js
+++ b/test/parallel/test-stream-iter-namespace.js
@@ -43,6 +43,7 @@ async function testStreamNamespaceAsyncConsumers() {
   assert.strictEqual(typeof Stream.text, 'function');
   assert.strictEqual(typeof Stream.arrayBuffer, 'function');
   assert.strictEqual(typeof Stream.array, 'function');
+  assert.strictEqual(typeof Stream.drain, 'function');
 }
 
 async function testStreamNamespaceSyncConsumers() {
@@ -52,6 +53,7 @@ async function testStreamNamespaceSyncConsumers() {
   assert.strictEqual(typeof Stream.textSync, 'function');
   assert.strictEqual(typeof Stream.arrayBufferSync, 'function');
   assert.strictEqual(typeof Stream.arraySync, 'function');
+  assert.strictEqual(typeof Stream.drainSync, 'function');
 }
 
 async function testStreamNamespaceCombining() {
@@ -108,6 +110,8 @@ async function testIndividualExports() {
   assert.strictEqual(typeof streamNew.arrayBufferSync, 'function');
   assert.strictEqual(typeof streamNew.array, 'function');
   assert.strictEqual(typeof streamNew.arraySync, 'function');
+  assert.strictEqual(typeof streamNew.drain, 'function');
+  assert.strictEqual(typeof streamNew.drainSync, 'function');
 
   // Combining
   assert.strictEqual(typeof streamNew.merge, 'function');
@@ -163,6 +167,8 @@ async function testNamespaceMatchesExports() {
   assert.strictEqual(Stream.textSync, streamNew.textSync);
   assert.strictEqual(Stream.arrayBufferSync, streamNew.arrayBufferSync);
   assert.strictEqual(Stream.arraySync, streamNew.arraySync);
+  assert.strictEqual(Stream.drain, streamNew.drain);
+  assert.strictEqual(Stream.drainSync, streamNew.drainSync);
   assert.strictEqual(Stream.merge, streamNew.merge);
   assert.strictEqual(Stream.broadcast, streamNew.broadcast);
   assert.strictEqual(Stream.share, streamNew.share);

From ef6ed85cbfcda41ed1f22aa4060204a9222d5c92 Mon Sep 17 00:00:00 2001
From: Ethan Arrowood <ethan@arrowood.dev>
Date: Thu, 27 Aug 2026 12:19:38 -0600
Subject: [PATCH 2/2] test: use drain() for discard loops in stream/iter tests

Replaces the `for await (const _ of source) {}` discard idiom with
drain(), which says what it does and does not need an eslint-disable
for the unused loop variable. 89 loops across 60 files, plus 32 now
redundant eslint-disable directives and comments removed.

Only files that already declare --experimental-stream-iter are
converted, so no test gains an experimental flag it did not already
opt into. That leaves 19 files with the old idiom; converting those
would change what those tests run under and belongs in a separate
discussion.

Loops that break early are left alone: those cancel the source rather
than reading it to completion, which is not what drain() does.

Four QUIC tests already bound a local `drain` to the writer-side
drainableProtocol promise. Those locals are renamed to drainPromise;
in test-quic-stream-writer-api.mjs the local shadowed the import and
broke at runtime rather than at parse time.

Assisted-by: Claude Opus 5
Signed-off-by: Ethan Arrowood <ethan@arrowood.dev>
---
 test/parallel/test-quic-cc-algorithm.mjs      |  3 ++-
 test/parallel/test-quic-datagram-multiple.mjs |  3 ++-
 .../test-quic-diagnostics-channel-stream.mjs  |  3 ++-
 test/parallel/test-quic-flow-control-blob.mjs |  3 ++-
 .../test-quic-flow-control-block-resume.mjs   |  3 ++-
 .../test-quic-flow-control-params.mjs         |  7 +++---
 ...-maxstreamdata-external-buffer-failure.mjs |  5 +++--
 test/parallel/test-quic-key-update-peer.mjs   |  3 ++-
 test/parallel/test-quic-key-update.mjs        |  3 ++-
 test/parallel/test-quic-max-payload-size.mjs  |  3 ++-
 test/parallel/test-quic-max-window.mjs        |  5 +++--
 ...-maxstreamdata-external-buffer-failure.mjs |  5 +++--
 test/parallel/test-quic-perf-hooks.mjs        |  3 ++-
 .../test-quic-session-application-options.mjs |  4 ++--
 .../test-quic-session-close-graceful.mjs      |  3 ++-
 .../test-quic-session-initial-rtt.mjs         |  3 ++-
 .../test-quic-session-stats-detailed.mjs      |  3 ++-
 test/parallel/test-quic-session-stats.mjs     |  3 ++-
 .../test-quic-session-transport-params.mjs    |  4 ++--
 ...test-quic-shared-endpoint-stream-close.mjs |  5 +++--
 test/parallel/test-quic-stateless-reset.mjs   |  5 +++--
 ...test-quic-stream-bidi-server-initiated.mjs |  4 ++--
 .../test-quic-stream-body-async-iterable.mjs  |  3 ++-
 .../test-quic-stream-body-pooled-buffer.mjs   |  3 ++-
 .../test-quic-stream-body-promise-reject.mjs  |  3 ++-
 .../test-quic-stream-body-promise.mjs         |  5 +++--
 .../test-quic-stream-body-readable-stream.mjs |  5 +++--
 .../test-quic-stream-body-sources.mjs         |  9 ++++----
 .../parallel/test-quic-stream-body-string.mjs |  3 ++-
 .../test-quic-stream-body-sync-iterable.mjs   |  3 ++-
 .../test-quic-stream-closed-promise.mjs       |  3 ++-
 .../test-quic-stream-iteration-double.mjs     |  3 ++-
 .../test-quic-stream-iteration-pull.mjs       |  3 ++-
 test/parallel/test-quic-stream-iteration.mjs  |  3 ++-
 .../test-quic-stream-limits-pending.mjs       |  6 ++---
 test/parallel/test-quic-stream-many-rapid.mjs |  3 ++-
 test/parallel/test-quic-stream-onblocked.mjs  |  3 ++-
 test/parallel/test-quic-stream-pending.mjs    |  3 ++-
 .../test-quic-stream-recv-coalescing.mjs      |  4 ++--
 .../test-quic-stream-slow-consumer.mjs        |  3 ++-
 test/parallel/test-quic-stream-stats.mjs      |  3 ++-
 .../test-quic-stream-write-partial-view.mjs   |  3 ++-
 test/parallel/test-quic-stream-writer-api.mjs | 15 +++++++------
 .../parallel/test-quic-stream-zero-length.mjs |  3 ++-
 .../test-quic-writer-async-dispose-ended.mjs  |  3 ++-
 .../test-quic-writer-backpressure.mjs         | 10 +++++----
 .../test-quic-writer-write-rejects.mjs        |  9 ++++----
 test/parallel/test-quic-zero-rtt.mjs          |  5 +++--
 .../test-stream-iter-broadcast-basic.js       |  5 ++---
 .../test-stream-iter-broadcast-from.js        |  5 ++---
 .../test-stream-iter-consumers-merge.js       | 18 +++++----------
 .../test-stream-iter-consumers-tap.js         |  4 ++--
 test/parallel/test-stream-iter-duplex.js      |  5 ++---
 test/parallel/test-stream-iter-pull-async.js  | 22 ++++++-------------
 test/parallel/test-stream-iter-push-writer.js | 20 ++++++-----------
 test/parallel/test-stream-iter-share-async.js |  7 +++---
 .../test-stream-iter-share-coverage.js        | 10 ++++-----
 test/parallel/test-stream-iter-share-from.js  |  9 ++++----
 test/parallel/test-stream-iter-to-readable.js | 17 +++++---------
 .../test-stream-iter-transform-output.js      |  8 +++----
 60 files changed, 171 insertions(+), 161 deletions(-)

diff --git a/test/parallel/test-quic-cc-algorithm.mjs b/test/parallel/test-quic-cc-algorithm.mjs
index 8b0a00feb5d6..65013d2d83b8 100644
--- a/test/parallel/test-quic-cc-algorithm.mjs
+++ b/test/parallel/test-quic-cc-algorithm.mjs
@@ -6,6 +6,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -39,7 +40,7 @@ for (const cc of ['reno', 'cubic', 'bbr']) {
     body: encoder.encode('congestion control test'),
   });
 
-  for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+  await drain(stream);
   await Promise.all([stream.closed, serverDone.promise]);
 
   // Verify the session stats show congestion control was active.
diff --git a/test/parallel/test-quic-datagram-multiple.mjs b/test/parallel/test-quic-datagram-multiple.mjs
index 733322ea047d..468e04e4d0af 100644
--- a/test/parallel/test-quic-datagram-multiple.mjs
+++ b/test/parallel/test-quic-datagram-multiple.mjs
@@ -10,6 +10,7 @@
 import { hasQuic, skip, mustCall, mustCallAtLeast } from '../common/index.mjs';
 import assert from 'node:assert';
 import * as fixtures from '../common/fixtures.mjs';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -72,7 +73,7 @@ for (let i = 0; i < numDatagrams; i++) {
 }
 
 // Complete the stream.
-for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(stream);
 await stream.closed;
 
 // At least some datagrams should have arrived.
diff --git a/test/parallel/test-quic-diagnostics-channel-stream.mjs b/test/parallel/test-quic-diagnostics-channel-stream.mjs
index 96a6074fc374..97fccc4c2259 100644
--- a/test/parallel/test-quic-diagnostics-channel-stream.mjs
+++ b/test/parallel/test-quic-diagnostics-channel-stream.mjs
@@ -9,6 +9,7 @@
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
 import dc from 'node:diagnostics_channel';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -59,7 +60,7 @@ const stream = await clientSession.createBidirectionalStream({
   body: encoder.encode('diagnostics test'),
 });
 
-for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(stream);
 
 await Promise.all([stream.closed, serverDone.promise, clientSession.closed]);
 await serverEndpoint.close();
diff --git a/test/parallel/test-quic-flow-control-blob.mjs b/test/parallel/test-quic-flow-control-blob.mjs
index 888562499216..52a18d36c5cb 100644
--- a/test/parallel/test-quic-flow-control-blob.mjs
+++ b/test/parallel/test-quic-flow-control-blob.mjs
@@ -7,6 +7,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -42,7 +43,7 @@ const stream = await clientSession.createBidirectionalStream({
   body: blob,
 });
 
-for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(stream);
 await Promise.all([stream.closed, serverDone.promise]);
 await clientSession.close();
 await serverEndpoint.close();
diff --git a/test/parallel/test-quic-flow-control-block-resume.mjs b/test/parallel/test-quic-flow-control-block-resume.mjs
index 552932651b1c..e8f065f69e50 100644
--- a/test/parallel/test-quic-flow-control-block-resume.mjs
+++ b/test/parallel/test-quic-flow-control-block-resume.mjs
@@ -8,6 +8,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -43,7 +44,7 @@ await clientSession.opened;
 const stream = await clientSession.createBidirectionalStream();
 stream.setBody(data);
 
-for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(stream);
 
 await Promise.all([stream.closed, serverDone.promise, clientSession.closed]);
 
diff --git a/test/parallel/test-quic-flow-control-params.mjs b/test/parallel/test-quic-flow-control-params.mjs
index 6b0e70ea2238..6b6f53c74f53 100644
--- a/test/parallel/test-quic-flow-control-params.mjs
+++ b/test/parallel/test-quic-flow-control-params.mjs
@@ -12,6 +12,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -57,13 +58,13 @@ const encoder = new TextEncoder();
   for (let offset = 0; offset < expected.byteLength; offset += chunkSize) {
     const chunk = expected.slice(offset, offset + chunkSize);
     while (!w.writeSync(chunk)) {
-      const drain = w[dp]();
-      if (drain) await drain;
+      const drainPromise = w[dp]();
+      if (drainPromise) await drainPromise;
     }
   }
   w.endSync();
 
-  for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+  await drain(stream);
   await Promise.all([stream.closed, serverDone.promise]);
   await clientSession.close();
   await serverEndpoint.close();
diff --git a/test/parallel/test-quic-h3-maxstreamdata-external-buffer-failure.mjs b/test/parallel/test-quic-h3-maxstreamdata-external-buffer-failure.mjs
index df74e4db408b..ab3e490fdfb2 100644
--- a/test/parallel/test-quic-h3-maxstreamdata-external-buffer-failure.mjs
+++ b/test/parallel/test-quic-h3-maxstreamdata-external-buffer-failure.mjs
@@ -7,6 +7,7 @@
 import { hasQuic, skip } from '../common/index.mjs';
 import { readFile } from 'node:fs/promises';
 import { setTimeout as sleep } from 'node:timers/promises';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -33,8 +34,8 @@ const serverMayRead = new Promise((resolve) => { letServerRead = resolve; });
 const endpoint = await listen((session) => {
   session.onstream = async (stream) => {
     await serverMayRead;
-    // eslint-disable-next-line no-unused-vars
-    for await (const _ of stream) { /* reading extends the window */ }
+    // Reading extends the window
+    await drain(stream);
   };
 }, {
   sni: { '*': { keys: [key], certs: [cert] } },
diff --git a/test/parallel/test-quic-key-update-peer.mjs b/test/parallel/test-quic-key-update-peer.mjs
index 1a0fed1a1672..ad2e7f209c17 100644
--- a/test/parallel/test-quic-key-update-peer.mjs
+++ b/test/parallel/test-quic-key-update-peer.mjs
@@ -6,6 +6,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -41,7 +42,7 @@ await clientSession.opened;
 const stream = await clientSession.createBidirectionalStream({
   body: encoder.encode('after key update'),
 });
-for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(stream);
 await Promise.all([stream.closed, serverDone.promise]);
 
 await clientSession.closed;
diff --git a/test/parallel/test-quic-key-update.mjs b/test/parallel/test-quic-key-update.mjs
index b01288e90f41..8643a7bc1706 100644
--- a/test/parallel/test-quic-key-update.mjs
+++ b/test/parallel/test-quic-key-update.mjs
@@ -7,6 +7,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -42,7 +43,7 @@ clientSession.updateKey();
 const stream = await clientSession.createBidirectionalStream();
 stream.setBody(data);
 
-for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(stream);
 await Promise.all([stream.closed, serverDone.promise]);
 await clientSession.close();
 await serverEndpoint.close();
diff --git a/test/parallel/test-quic-max-payload-size.mjs b/test/parallel/test-quic-max-payload-size.mjs
index 4aa2314f047e..db7da26d44cc 100644
--- a/test/parallel/test-quic-max-payload-size.mjs
+++ b/test/parallel/test-quic-max-payload-size.mjs
@@ -7,6 +7,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -38,7 +39,7 @@ async function transferAndGetPacketCount(maxPayloadSize) {
 
   const stream = await clientSession.createBidirectionalStream();
   stream.setBody(new Uint8Array(dataLength));
-  for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+  await drain(stream);
   await Promise.all([stream.closed, serverDone.promise]);
 
   const pktSent = clientSession.stats.pktSent;
diff --git a/test/parallel/test-quic-max-window.mjs b/test/parallel/test-quic-max-window.mjs
index b0ee67afa964..1ab4e7f80e68 100644
--- a/test/parallel/test-quic-max-window.mjs
+++ b/test/parallel/test-quic-max-window.mjs
@@ -8,6 +8,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -41,7 +42,7 @@ const dataLength = 8192;
 
   const stream = await clientSession.createBidirectionalStream();
   stream.setBody(new Uint8Array(dataLength));
-  for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+  await drain(stream);
   await Promise.all([stream.closed, serverDone.promise, clientSession.closed]);
   await serverEndpoint.close();
 }
@@ -69,7 +70,7 @@ const dataLength = 8192;
 
   const stream = await clientSession.createBidirectionalStream();
   stream.setBody(new Uint8Array(dataLength));
-  for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+  await drain(stream);
   await Promise.all([stream.closed, serverDone.promise, clientSession.closed]);
   await serverEndpoint.close();
 }
diff --git a/test/parallel/test-quic-maxstreamdata-external-buffer-failure.mjs b/test/parallel/test-quic-maxstreamdata-external-buffer-failure.mjs
index 79d11e05794a..91c955454df0 100644
--- a/test/parallel/test-quic-maxstreamdata-external-buffer-failure.mjs
+++ b/test/parallel/test-quic-maxstreamdata-external-buffer-failure.mjs
@@ -7,6 +7,7 @@
 import { hasQuic, skip } from '../common/index.mjs';
 import { readFile } from 'node:fs/promises';
 import { setTimeout as sleep } from 'node:timers/promises';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -33,8 +34,8 @@ const serverMayRead = new Promise((resolve) => { letServerRead = resolve; });
 const endpoint = await listen((session) => {
   session.onstream = async (stream) => {
     await serverMayRead;
-    // eslint-disable-next-line no-unused-vars
-    for await (const _ of stream) { /* reading extends the window */ }
+    // Reading extends the window
+    await drain(stream);
   };
 }, {
   alpn: 'foo',
diff --git a/test/parallel/test-quic-perf-hooks.mjs b/test/parallel/test-quic-perf-hooks.mjs
index fb9f90c5f472..de23fa3860b0 100644
--- a/test/parallel/test-quic-perf-hooks.mjs
+++ b/test/parallel/test-quic-perf-hooks.mjs
@@ -7,6 +7,7 @@
 import { hasQuic, skip, mustCall, mustCallAtLeast } from '../common/index.mjs';
 import assert from 'node:assert';
 import { PerformanceObserver } from 'node:perf_hooks';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -53,7 +54,7 @@ const stream = await clientSession.createBidirectionalStream({
   body: encoder.encode('perf test'),
 });
 
-for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(stream);
 await Promise.all([stream.closed, serverDone.promise, clientSession.closed]);
 await serverEndpoint.close();
 
diff --git a/test/parallel/test-quic-session-application-options.mjs b/test/parallel/test-quic-session-application-options.mjs
index b119207bcc1f..c2c6c08330f8 100644
--- a/test/parallel/test-quic-session-application-options.mjs
+++ b/test/parallel/test-quic-session-application-options.mjs
@@ -7,6 +7,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -92,8 +93,7 @@ assert.strictEqual(clientOpts.enableDatagrams, customAppOptions.enableDatagrams)
 const stream = await clientSession.createBidirectionalStream();
 stream.writer.endSync();
 
-// eslint-disable-next-line no-unused-vars
-for await (const _ of stream) { /* drain */ }
+await drain(stream);
 await Promise.all([stream.closed, serverDone.promise]);
 
 // After close, applicationOptions should return null.
diff --git a/test/parallel/test-quic-session-close-graceful.mjs b/test/parallel/test-quic-session-close-graceful.mjs
index b059b74f8324..1e81c8035f5b 100644
--- a/test/parallel/test-quic-session-close-graceful.mjs
+++ b/test/parallel/test-quic-session-close-graceful.mjs
@@ -7,6 +7,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -49,7 +50,7 @@ const encoder = new TextEncoder();
 
   // Wait for the stream to complete normally.
   await serverDone.promise;
-  for await (const batch of stream) { /* drain server FIN */ } // eslint-disable-line no-unused-vars
+  await drain(stream);
   await stream.closed;
 
   // Now the closed promise should resolve.
diff --git a/test/parallel/test-quic-session-initial-rtt.mjs b/test/parallel/test-quic-session-initial-rtt.mjs
index ce24f804dd04..611f1119e798 100644
--- a/test/parallel/test-quic-session-initial-rtt.mjs
+++ b/test/parallel/test-quic-session-initial-rtt.mjs
@@ -5,6 +5,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -42,7 +43,7 @@ const stream = await clientSession.createBidirectionalStream({
   body: payload,
 });
 
-for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(stream);
 await stream.closed;
 await serverDone.promise;
 
diff --git a/test/parallel/test-quic-session-stats-detailed.mjs b/test/parallel/test-quic-session-stats-detailed.mjs
index 58d769924ec0..ea60f890e56c 100644
--- a/test/parallel/test-quic-session-stats-detailed.mjs
+++ b/test/parallel/test-quic-session-stats-detailed.mjs
@@ -7,6 +7,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -35,7 +36,7 @@ const data = new Uint8Array(8192);
 const stream = await clientSession.createBidirectionalStream();
 stream.setBody(data);
 
-for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(stream);
 await Promise.all([stream.closed, serverDone.promise]);
 
 const stats = clientSession.stats;
diff --git a/test/parallel/test-quic-session-stats.mjs b/test/parallel/test-quic-session-stats.mjs
index b1a6c5aa7f75..708075cfce52 100644
--- a/test/parallel/test-quic-session-stats.mjs
+++ b/test/parallel/test-quic-session-stats.mjs
@@ -6,6 +6,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -50,7 +51,7 @@ const stream = await clientSession.createBidirectionalStream({
   body: payload,
 });
 
-for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(stream);
 await stream.closed;
 await serverDone.promise;
 
diff --git a/test/parallel/test-quic-session-transport-params.mjs b/test/parallel/test-quic-session-transport-params.mjs
index 6fd92419196f..e9af80fa919b 100644
--- a/test/parallel/test-quic-session-transport-params.mjs
+++ b/test/parallel/test-quic-session-transport-params.mjs
@@ -7,6 +7,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -118,8 +119,7 @@ assert.ok(clientRemoteParams.initialSCID.length > 0,
 const stream = await clientSession.createBidirectionalStream();
 stream.writer.endSync();
 
-// eslint-disable-next-line no-unused-vars
-for await (const _ of stream) { /* drain */ }
+await drain(stream);
 await stream.closed;
 await serverDone.promise;
 
diff --git a/test/parallel/test-quic-shared-endpoint-stream-close.mjs b/test/parallel/test-quic-shared-endpoint-stream-close.mjs
index 19e530e50b9e..83c20ba6fdb6 100644
--- a/test/parallel/test-quic-shared-endpoint-stream-close.mjs
+++ b/test/parallel/test-quic-shared-endpoint-stream-close.mjs
@@ -8,6 +8,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -59,7 +60,7 @@ await client1.opened;
 const s1 = await client1.createBidirectionalStream({
   body: encoder.encode('session1'),
 });
-for await (const _ of s1) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(s1);
 await s1.closed;
 
 await serverDone1.promise;
@@ -84,7 +85,7 @@ await client2.opened;
 const s2 = await client2.createBidirectionalStream({
   body: encoder.encode('session2'),
 });
-for await (const _ of s2) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(s2);
 await s2.closed;
 
 // If the bug is present, this never resolves and the test hangs.
diff --git a/test/parallel/test-quic-stateless-reset.mjs b/test/parallel/test-quic-stateless-reset.mjs
index 90ff9e2c80d5..f1bc484c029e 100644
--- a/test/parallel/test-quic-stateless-reset.mjs
+++ b/test/parallel/test-quic-stateless-reset.mjs
@@ -10,6 +10,7 @@
 
 import { hasQuic, skip, mustCall, expectsError, mustNotCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -54,7 +55,7 @@ const encoder = new TextEncoder();
   const stream1 = await clientSession.createBidirectionalStream({
     body: encoder.encode('hello'),
   });
-  for await (const _ of stream1) { /* drain */ } // eslint-disable-line no-unused-vars
+  await drain(stream1);
   await stream1.closed;
 
   // Wait for the server to destroy.
@@ -113,7 +114,7 @@ const encoder = new TextEncoder();
   const stream1 = await clientSession.createBidirectionalStream({
     body: encoder.encode('hello'),
   });
-  for await (const _ of stream1) { /* drain */ } // eslint-disable-line no-unused-vars
+  await drain(stream1);
   await stream1.closed;
 
   await serverDestroyed.promise;
diff --git a/test/parallel/test-quic-stream-bidi-server-initiated.mjs b/test/parallel/test-quic-stream-bidi-server-initiated.mjs
index d2279b28fe5e..b27e3b72268f 100644
--- a/test/parallel/test-quic-stream-bidi-server-initiated.mjs
+++ b/test/parallel/test-quic-stream-bidi-server-initiated.mjs
@@ -7,6 +7,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -29,8 +30,7 @@ const serverEndpoint = await listen(mustCall(async (serverSession) => {
     body: encoder.encode(message),
   });
 
-  // Drain the client's write side (client sends FIN with no data).
-  for await (const batch of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+  await drain(stream);
   await stream.closed;
 }));
 
diff --git a/test/parallel/test-quic-stream-body-async-iterable.mjs b/test/parallel/test-quic-stream-body-async-iterable.mjs
index 168144689eb8..7ee3ca5a4201 100644
--- a/test/parallel/test-quic-stream-body-async-iterable.mjs
+++ b/test/parallel/test-quic-stream-body-async-iterable.mjs
@@ -6,6 +6,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import * as assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -43,7 +44,7 @@ async function* generateChunks() {
 const stream = await clientSession.createBidirectionalStream();
 stream.setBody(generateChunks());
 
-for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(stream);
 await Promise.all([stream.closed, serverDone.promise]);
 await clientSession.close();
 await serverEndpoint.close();
diff --git a/test/parallel/test-quic-stream-body-pooled-buffer.mjs b/test/parallel/test-quic-stream-body-pooled-buffer.mjs
index 2a98882ac157..9fb2a48e46ba 100644
--- a/test/parallel/test-quic-stream-body-pooled-buffer.mjs
+++ b/test/parallel/test-quic-stream-body-pooled-buffer.mjs
@@ -6,6 +6,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -43,7 +44,7 @@ await clientSession.opened;
 const stream = await clientSession.createBidirectionalStream();
 stream.setBody(expected);
 
-for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(stream);
 await Promise.all([stream.closed, serverDone.promise]);
 await clientSession.close();
 await serverEndpoint.close();
diff --git a/test/parallel/test-quic-stream-body-promise-reject.mjs b/test/parallel/test-quic-stream-body-promise-reject.mjs
index ebfaf98a7bdf..f1f7022a53d4 100644
--- a/test/parallel/test-quic-stream-body-promise-reject.mjs
+++ b/test/parallel/test-quic-stream-body-promise-reject.mjs
@@ -6,6 +6,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -42,7 +43,7 @@ const { bytes } = await import('stream/iter');
     Promise.resolve(Promise.resolve('nested promise')),
   );
 
-  for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+  await drain(stream);
   await Promise.all([stream.closed, serverDone.promise]);
   await clientSession.close();
   await serverEndpoint.close();
diff --git a/test/parallel/test-quic-stream-body-promise.mjs b/test/parallel/test-quic-stream-body-promise.mjs
index 1aab4101d129..cd25f17a1b4f 100644
--- a/test/parallel/test-quic-stream-body-promise.mjs
+++ b/test/parallel/test-quic-stream-body-promise.mjs
@@ -9,6 +9,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -52,7 +53,7 @@ await clientSession.opened;
 {
   const stream = await clientSession.createBidirectionalStream();
   stream.setBody(Promise.resolve('resolved string'));
-  for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+  await drain(stream);
   await stream.closed;
 }
 
@@ -60,7 +61,7 @@ await clientSession.opened;
 {
   const stream = await clientSession.createBidirectionalStream();
   stream.setBody(Promise.resolve(null));
-  for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+  await drain(stream);
   await stream.closed;
 }
 
diff --git a/test/parallel/test-quic-stream-body-readable-stream.mjs b/test/parallel/test-quic-stream-body-readable-stream.mjs
index b235610736f9..e023c32f6c14 100644
--- a/test/parallel/test-quic-stream-body-readable-stream.mjs
+++ b/test/parallel/test-quic-stream-body-readable-stream.mjs
@@ -4,6 +4,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -46,7 +47,7 @@ await clientSession.opened;
   });
   const stream = await clientSession.createBidirectionalStream();
   stream.setBody(rs);
-  for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+  await drain(stream);
   await stream.closed;
 }
 
@@ -55,7 +56,7 @@ await clientSession.opened;
   const readable = Readable.from([encoder.encode(message)]);
   const stream = await clientSession.createBidirectionalStream();
   stream.setBody(readable);
-  for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+  await drain(stream);
   await stream.closed;
 }
 
diff --git a/test/parallel/test-quic-stream-body-sources.mjs b/test/parallel/test-quic-stream-body-sources.mjs
index d4f13d84b5ca..cc5591687838 100644
--- a/test/parallel/test-quic-stream-body-sources.mjs
+++ b/test/parallel/test-quic-stream-body-sources.mjs
@@ -7,6 +7,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import * as assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -45,7 +46,7 @@ await clientSession.opened;
   const buf = encoder.encode(message);
   const ab = buf.buffer.slice(buf.byteOffset, buf.byteOffset + buf.byteLength);
   const stream = await clientSession.createBidirectionalStream({ body: ab });
-  for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+  await drain(stream);
   await stream.closed;
 }
 
@@ -57,7 +58,7 @@ await clientSession.opened;
   fullView.set(expectedBytes, offset);
   const view = new Uint8Array(backing, offset, expectedBytes.byteLength);
   const stream = await clientSession.createBidirectionalStream({ body: view });
-  for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+  await drain(stream);
   await stream.closed;
 }
 
@@ -69,7 +70,7 @@ await clientSession.opened;
   const stream = await clientSession.createBidirectionalStream({ body: sabView });
   // The SharedArrayBuffer should still be usable (copied, not transferred).
   assert.strictEqual(sab.byteLength, expectedBytes.byteLength);
-  for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+  await drain(stream);
   await stream.closed;
 }
 
@@ -77,7 +78,7 @@ await clientSession.opened;
 {
   const blob = new Blob([expectedBytes]);
   const stream = await clientSession.createBidirectionalStream({ body: blob });
-  for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+  await drain(stream);
   await stream.closed;
 }
 
diff --git a/test/parallel/test-quic-stream-body-string.mjs b/test/parallel/test-quic-stream-body-string.mjs
index cfe2790028d0..0529c73e2ffe 100644
--- a/test/parallel/test-quic-stream-body-string.mjs
+++ b/test/parallel/test-quic-stream-body-string.mjs
@@ -4,6 +4,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -35,7 +36,7 @@ await clientSession.opened;
 const stream = await clientSession.createBidirectionalStream();
 stream.setBody(message);
 
-for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(stream);
 await Promise.all([stream.closed, serverDone.promise]);
 await clientSession.close();
 await serverEndpoint.close();
diff --git a/test/parallel/test-quic-stream-body-sync-iterable.mjs b/test/parallel/test-quic-stream-body-sync-iterable.mjs
index f65cb0da45ab..63cc01842669 100644
--- a/test/parallel/test-quic-stream-body-sync-iterable.mjs
+++ b/test/parallel/test-quic-stream-body-sync-iterable.mjs
@@ -4,6 +4,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -37,7 +38,7 @@ const bodyChunks = chunks.map((c) => encoder.encode(c));
 const stream = await clientSession.createBidirectionalStream();
 stream.setBody(bodyChunks);
 
-for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(stream);
 await Promise.all([stream.closed, serverDone.promise]);
 await clientSession.close();
 await serverEndpoint.close();
diff --git a/test/parallel/test-quic-stream-closed-promise.mjs b/test/parallel/test-quic-stream-closed-promise.mjs
index 2c30a44fb70b..c9185d351e82 100644
--- a/test/parallel/test-quic-stream-closed-promise.mjs
+++ b/test/parallel/test-quic-stream-closed-promise.mjs
@@ -3,6 +3,7 @@
 // Test: stream.closed promise resolves after normal completion.
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -32,7 +33,7 @@ const stream = await clientSession.createBidirectionalStream({
   body: encoder.encode('normal close'),
 });
 
-for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(stream);
 
 // Closed should resolve (not reject).
 await Promise.all([stream.closed, serverDone.promise]);
diff --git a/test/parallel/test-quic-stream-iteration-double.mjs b/test/parallel/test-quic-stream-iteration-double.mjs
index 122d1554559f..bb1ff2d549c0 100644
--- a/test/parallel/test-quic-stream-iteration-double.mjs
+++ b/test/parallel/test-quic-stream-iteration-double.mjs
@@ -8,6 +8,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import * as assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -52,7 +53,7 @@ await clientSession.opened;
 const stream = await clientSession.createBidirectionalStream({
   body: encoder.encode('double iter test'),
 });
-for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(stream);
 await Promise.all([stream.closed, serverDone.promise]);
 await clientSession.close();
 await serverEndpoint.close();
diff --git a/test/parallel/test-quic-stream-iteration-pull.mjs b/test/parallel/test-quic-stream-iteration-pull.mjs
index 3dba46c77680..462dfb4e5f8a 100644
--- a/test/parallel/test-quic-stream-iteration-pull.mjs
+++ b/test/parallel/test-quic-stream-iteration-pull.mjs
@@ -6,6 +6,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -44,7 +45,7 @@ const stream = await clientSession.createBidirectionalStream({
   body: encoder.encode(message),
 });
 
-for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(stream);
 await Promise.all([stream.closed, serverDone.promise]);
 await clientSession.close();
 await serverEndpoint.close();
diff --git a/test/parallel/test-quic-stream-iteration.mjs b/test/parallel/test-quic-stream-iteration.mjs
index 1f4635d0364a..d518c18f46de 100644
--- a/test/parallel/test-quic-stream-iteration.mjs
+++ b/test/parallel/test-quic-stream-iteration.mjs
@@ -9,6 +9,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -70,7 +71,7 @@ for (let i = 0; i < totalStreams; i++) {
   const stream = await clientSession.createBidirectionalStream({
     body: encoder.encode(message),
   });
-  for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+  await drain(stream);
   await stream.closed;
 }
 
diff --git a/test/parallel/test-quic-stream-limits-pending.mjs b/test/parallel/test-quic-stream-limits-pending.mjs
index bd7bcd51fff8..0f1a0a05cf7b 100644
--- a/test/parallel/test-quic-stream-limits-pending.mjs
+++ b/test/parallel/test-quic-stream-limits-pending.mjs
@@ -8,6 +8,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -53,8 +54,7 @@ const s2 = await clientSession.createBidirectionalStream({
 // more stream credits.
 assert.strictEqual(s2.pending, true);
 
-// Drain and close the first stream.
-for await (const _ of s1) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(s1);
 await s1.closed;
 
 // After s1 closes, the server sends MAX_STREAMS which opens s2.
@@ -62,7 +62,7 @@ await s1.closed;
 await allDone.promise;
 
 // s2 should no longer be pending.
-for await (const _ of s2) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(s2);
 await s2.closed;
 
 await clientSession.close();
diff --git a/test/parallel/test-quic-stream-many-rapid.mjs b/test/parallel/test-quic-stream-many-rapid.mjs
index 94660889718a..9b235be2e707 100644
--- a/test/parallel/test-quic-stream-many-rapid.mjs
+++ b/test/parallel/test-quic-stream-many-rapid.mjs
@@ -7,6 +7,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -47,7 +48,7 @@ for (let i = 0; i < streamCount; i++) {
 
 // Wait for all client streams to close.
 await Promise.all(streams.map(async (stream) => {
-  for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+  await drain(stream);
   await stream.closed;
 }));
 
diff --git a/test/parallel/test-quic-stream-onblocked.mjs b/test/parallel/test-quic-stream-onblocked.mjs
index b1b9d16c277b..e8a113cfdc44 100644
--- a/test/parallel/test-quic-stream-onblocked.mjs
+++ b/test/parallel/test-quic-stream-onblocked.mjs
@@ -12,6 +12,7 @@
 import { hasQuic, skip, mustCall, mustCallAtLeast } from '../common/index.mjs';
 import assert from 'node:assert';
 import dc from 'node:diagnostics_channel';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -61,7 +62,7 @@ stream.onblocked = mustCallAtLeast(() => {
 // NGTCP2_ERR_STREAM_DATA_BLOCKED, triggering onblocked.
 stream.setBody(body);
 
-for await (const _ of stream) { /* drain readable side */ } // eslint-disable-line no-unused-vars
+await drain(stream);
 await stream.closed;
 await serverDone.promise;
 
diff --git a/test/parallel/test-quic-stream-pending.mjs b/test/parallel/test-quic-stream-pending.mjs
index a51bc711585e..77341f825596 100644
--- a/test/parallel/test-quic-stream-pending.mjs
+++ b/test/parallel/test-quic-stream-pending.mjs
@@ -6,6 +6,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -49,7 +50,7 @@ await serverDone.promise;
 // After the server received data, the stream opened successfully.
 // The data arrival proves (pending stream opens after handshake).
 
-for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(stream);
 await stream.closed;
 await clientSession.close();
 await serverEndpoint.close();
diff --git a/test/parallel/test-quic-stream-recv-coalescing.mjs b/test/parallel/test-quic-stream-recv-coalescing.mjs
index c06e7facc830..68889521b050 100644
--- a/test/parallel/test-quic-stream-recv-coalescing.mjs
+++ b/test/parallel/test-quic-stream-recv-coalescing.mjs
@@ -7,6 +7,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -123,8 +124,7 @@ for (let i = 0; i < payloadSizes.length; i++) {
 w.endSync();
 
 // Drain the server's empty response to let the stream close cleanly.
-// eslint-disable-next-line no-unused-vars
-for await (const _ of stream) { /* drain response */ }
+await drain(stream);
 await Promise.all([stream.closed, serverDone.promise]);
 await clientSession.close();
 await serverEndpoint.close();
diff --git a/test/parallel/test-quic-stream-slow-consumer.mjs b/test/parallel/test-quic-stream-slow-consumer.mjs
index 5f1c2d48eb4e..1c077c19ccd0 100644
--- a/test/parallel/test-quic-stream-slow-consumer.mjs
+++ b/test/parallel/test-quic-stream-slow-consumer.mjs
@@ -7,6 +7,7 @@
 
 import { hasQuic, skip, mustCall, mustCallAtLeast } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -46,7 +47,7 @@ stream.onblocked = mustCallAtLeast(() => {
 
 stream.setBody(new Uint8Array(dataLength));
 
-for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(stream);
 await Promise.all([stream.closed, serverDone.promise]);
 
 // The sender should have been blocked multiple times.
diff --git a/test/parallel/test-quic-stream-stats.mjs b/test/parallel/test-quic-stream-stats.mjs
index 1911e1493f4f..57964e41b716 100644
--- a/test/parallel/test-quic-stream-stats.mjs
+++ b/test/parallel/test-quic-stream-stats.mjs
@@ -6,6 +6,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -59,7 +60,7 @@ const json = stream.stats.toJSON();
 assert.ok(json);
 assert.strictEqual(typeof json.createdAt, 'string');
 
-for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(stream);
 await stream.closed;
 await serverDone.promise;
 
diff --git a/test/parallel/test-quic-stream-write-partial-view.mjs b/test/parallel/test-quic-stream-write-partial-view.mjs
index 0fcd8eea69f8..69a09309986d 100644
--- a/test/parallel/test-quic-stream-write-partial-view.mjs
+++ b/test/parallel/test-quic-stream-write-partial-view.mjs
@@ -8,6 +8,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -64,7 +65,7 @@ assert.ok(source.every((b) => b === 0), 'source mutation should succeed');
 
 writer.endSync();
 
-for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(stream);
 await Promise.all([stream.closed, serverDone.promise]);
 await clientSession.close();
 await serverEndpoint.close();
diff --git a/test/parallel/test-quic-stream-writer-api.mjs b/test/parallel/test-quic-stream-writer-api.mjs
index 009cb4ff8a4a..31ba290db173 100644
--- a/test/parallel/test-quic-stream-writer-api.mjs
+++ b/test/parallel/test-quic-stream-writer-api.mjs
@@ -6,6 +6,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import * as assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -44,7 +45,7 @@ await clientSession.opened;
   await w.write(encoder.encode('async write'));
   const n = w.endSync();
   assert.strictEqual(n, 11);
-  for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+  await drain(stream);
   await stream.closed;
 }
 
@@ -59,7 +60,7 @@ await clientSession.opened;
   assert.strictEqual(result, true);
   const n = w.endSync();
   assert.strictEqual(n, 12);
-  for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+  await drain(stream);
   await stream.closed;
 }
 
@@ -73,7 +74,7 @@ await clientSession.opened;
   ]);
   const n = w.endSync();
   assert.strictEqual(n, 12);
-  for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+  await drain(stream);
   await stream.closed;
 }
 
@@ -84,7 +85,7 @@ await clientSession.opened;
   w.writeSync(encoder.encode('end async'));
   const n = await w.end();
   assert.strictEqual(n, 9);
-  for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+  await drain(stream);
   await stream.closed;
 }
 
@@ -99,11 +100,11 @@ await clientSession.opened;
   // or a promise when canWrite is false (backpressured). Either way, it
   // should not throw.
   const { drainableProtocol: dp } = await import('stream/iter');
-  const drain = w[dp]();
-  assert.ok(drain === null || drain instanceof Promise);
+  const drainPromise = w[dp]();
+  assert.ok(drainPromise === null || drainPromise instanceof Promise);
   w.writeSync(encoder.encode('capacity'));
   w.endSync();
-  for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+  await drain(stream);
   await stream.closed;
 }
 
diff --git a/test/parallel/test-quic-stream-zero-length.mjs b/test/parallel/test-quic-stream-zero-length.mjs
index d7af2cf30f0a..33657e094e02 100644
--- a/test/parallel/test-quic-stream-zero-length.mjs
+++ b/test/parallel/test-quic-stream-zero-length.mjs
@@ -6,6 +6,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -33,7 +34,7 @@ await clientSession.opened;
 const stream = await clientSession.createBidirectionalStream();
 stream.setBody(null);
 
-for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(stream);
 
 await Promise.all([stream.closed, serverDone.promise]);
 await clientSession.close();
diff --git a/test/parallel/test-quic-writer-async-dispose-ended.mjs b/test/parallel/test-quic-writer-async-dispose-ended.mjs
index 04337343ea87..45a7a366d754 100644
--- a/test/parallel/test-quic-writer-async-dispose-ended.mjs
+++ b/test/parallel/test-quic-writer-async-dispose-ended.mjs
@@ -5,6 +5,7 @@
 // should not fail — it's a no-op.
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -39,7 +40,7 @@ w.endSync();
 await w[Symbol.asyncDispose]();
 
 // The stream should close cleanly.
-for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(stream);
 
 await Promise.all([stream.closed, serverDone.promise]);
 await clientSession.close();
diff --git a/test/parallel/test-quic-writer-backpressure.mjs b/test/parallel/test-quic-writer-backpressure.mjs
index 253fc24024ea..b416d205629b 100644
--- a/test/parallel/test-quic-writer-backpressure.mjs
+++ b/test/parallel/test-quic-writer-backpressure.mjs
@@ -8,6 +8,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -56,11 +57,12 @@ for (let i = 0; i < numChunks; i++) {
     backpressureCount++;
 
     // drainableProtocol returns a promise when backpressured.
-    const drain = w[dp]();
-    assert.ok(drain instanceof Promise, 'drainableProtocol should return a Promise');
+    const drainPromise = w[dp]();
+    assert.ok(drainPromise instanceof Promise,
+              'drainableProtocol should return a Promise');
 
     // The promise resolves when drain fires.
-    await drain;
+    await drainPromise;
 
     // After drain, canWrite should be true.
     assert.ok(w.canWrite === true, `canWrite after drain should be true, got ${w.canWrite}`);
@@ -73,7 +75,7 @@ w.endSync();
 // and 1KB chunks (every 2 chunks fills the buffer).
 assert.ok(backpressureCount > 0, 'backpressure should have been hit');
 
-for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(stream);
 await Promise.all([stream.closed, serverDone.promise]);
 await clientSession.close();
 await serverEndpoint.close();
diff --git a/test/parallel/test-quic-writer-write-rejects.mjs b/test/parallel/test-quic-writer-write-rejects.mjs
index c29c63c075b2..cc646fd0c9eb 100644
--- a/test/parallel/test-quic-writer-write-rejects.mjs
+++ b/test/parallel/test-quic-writer-write-rejects.mjs
@@ -6,6 +6,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -48,16 +49,16 @@ await assert.rejects(
 );
 
 // Wait for drain, then write should succeed.
-const drain = w[dp]();
-assert.ok(drain instanceof Promise);
-await drain;
+const drainPromise = w[dp]();
+assert.ok(drainPromise instanceof Promise);
+await drainPromise;
 assert.ok(w.canWrite === true);
 
 // Now write succeeds.
 await w.write(new Uint8Array(100));
 
 w.endSync();
-for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(stream);
 await Promise.all([stream.closed, serverDone.promise]);
 await clientSession.close();
 await serverEndpoint.close();
diff --git a/test/parallel/test-quic-zero-rtt.mjs b/test/parallel/test-quic-zero-rtt.mjs
index e97c90391bcc..0a1fcf17d672 100644
--- a/test/parallel/test-quic-zero-rtt.mjs
+++ b/test/parallel/test-quic-zero-rtt.mjs
@@ -10,6 +10,7 @@
 
 import { hasQuic, skip, mustCall } from '../common/index.mjs';
 import assert from 'node:assert';
+import { drain } from 'node:stream/iter';
 
 if (!hasQuic) {
   skip('QUIC is not enabled');
@@ -76,7 +77,7 @@ await Promise.all([gotTicket.promise, gotToken.promise]);
 const s1 = await cs1.createBidirectionalStream({
   body: encoder.encode('first'),
 });
-for await (const _ of s1) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(s1);
 await Promise.all([s1.closed, cs1.closed]);
 
 // --- ZRTT-02: Second connection — 0-RTT with ticket + token ---
@@ -98,7 +99,7 @@ assert.strictEqual(info2.earlyDataAttempted, true);
 assert.strictEqual(info2.earlyDataAccepted, true);
 
 await s2.writer.end();
-for await (const _ of s2) { /* drain */ } // eslint-disable-line no-unused-vars
+await drain(s2);
 await s2.closed;
 
 // Verify the server saw the early data flag.
diff --git a/test/parallel/test-stream-iter-broadcast-basic.js b/test/parallel/test-stream-iter-broadcast-basic.js
index 3dbd3ce97512..1ddf9dde7a9c 100644
--- a/test/parallel/test-stream-iter-broadcast-basic.js
+++ b/test/parallel/test-stream-iter-broadcast-basic.js
@@ -4,7 +4,7 @@
 const common = require('../common');
 const assert = require('assert');
 const { setTimeout } = require('timers/promises');
-const { broadcast, text } = require('stream/iter');
+const { broadcast, drain, text } = require('stream/iter');
 
 // =============================================================================
 // Basic broadcast
@@ -320,8 +320,7 @@ async function testWriterFailIdempotent() {
   // Second call is a no-op (already errored)
   writer.fail(new Error('fail2'));
   await assert.rejects(async () => {
-    // eslint-disable-next-line no-unused-vars
-    for await (const _ of consumer) { /* consume */ }
+    await drain(consumer);
   }, { message: 'fail!' });
 }
 
diff --git a/test/parallel/test-stream-iter-broadcast-from.js b/test/parallel/test-stream-iter-broadcast-from.js
index 39d92c2aef49..cc33a8465d3d 100644
--- a/test/parallel/test-stream-iter-broadcast-from.js
+++ b/test/parallel/test-stream-iter-broadcast-from.js
@@ -3,7 +3,7 @@
 
 const common = require('../common');
 const assert = require('assert');
-const { broadcast, Broadcast, from, text } = require('stream/iter');
+const { Broadcast, broadcast, drain, from, text } = require('stream/iter');
 
 // =============================================================================
 // Broadcast.from
@@ -159,8 +159,7 @@ async function testBroadcastFromSourceError() {
   const { broadcast: bc } = Broadcast.from(failingSource());
   const consumer = bc.push();
   await assert.rejects(async () => {
-    // eslint-disable-next-line no-unused-vars
-    for await (const _ of consumer) { /* consume */ }
+    await drain(consumer);
   }, { message: 'broadcast source boom' });
 }
 
diff --git a/test/parallel/test-stream-iter-consumers-merge.js b/test/parallel/test-stream-iter-consumers-merge.js
index 84aeb24b6159..6749043d5285 100644
--- a/test/parallel/test-stream-iter-consumers-merge.js
+++ b/test/parallel/test-stream-iter-consumers-merge.js
@@ -4,10 +4,11 @@
 const common = require('../common');
 const assert = require('assert');
 const {
+  drain,
   from,
   fromSync,
-  push,
   merge,
+  push,
   text,
   toAsyncStreamable,
   toStreamable,
@@ -99,10 +100,7 @@ async function testMergeSourceError() {
   }
   await assert.rejects(
     async () => {
-      // eslint-disable-next-line no-unused-vars
-      for await (const _ of merge(goodSource(), badSource())) {
-        /* consume */
-      }
+      await drain(merge(goodSource(), badSource()));
     },
     { message: 'merge source boom' },
   );
@@ -287,10 +285,7 @@ async function testMergeCleanupErrorOnly() {
 
   await assert.rejects(
     async () => {
-      // eslint-disable-next-line no-unused-vars
-      for await (const _ of merge(source(), failingReturnSource())) {
-        // Consume all - no primary error
-      }
+      await drain(merge(source(), failingReturnSource()));
     },
     { message: 'cleanup boom' },
   );
@@ -314,10 +309,7 @@ async function testMergePrimaryAndCleanupError() {
 
   await assert.rejects(
     async () => {
-      // eslint-disable-next-line no-unused-vars
-      for await (const _ of merge(badSource(), failingReturnSource())) {
-        // Consume until error
-      }
+      await drain(merge(badSource(), failingReturnSource()));
     },
     (err) => {
       assert.ok(
diff --git a/test/parallel/test-stream-iter-consumers-tap.js b/test/parallel/test-stream-iter-consumers-tap.js
index b93f93eb242b..058c0f91ce93 100644
--- a/test/parallel/test-stream-iter-consumers-tap.js
+++ b/test/parallel/test-stream-iter-consumers-tap.js
@@ -4,6 +4,7 @@
 const common = require('../common');
 const assert = require('assert');
 const {
+  drain,
   from,
   fromSync,
   pull,
@@ -85,8 +86,7 @@ async function testTapInPipeline() {
 async function testTapAsyncErrorPropagation() {
   const badTap = tap(() => { throw new Error('tap error'); });
   await assert.rejects(async () => {
-    // eslint-disable-next-line no-unused-vars
-    for await (const _ of pull(from('hello'), badTap)) { /* consume */ }
+    await drain(pull(from('hello'), badTap));
   }, { message: 'tap error' });
 }
 
diff --git a/test/parallel/test-stream-iter-duplex.js b/test/parallel/test-stream-iter-duplex.js
index 0969b91e7d21..e941418c6d4a 100644
--- a/test/parallel/test-stream-iter-duplex.js
+++ b/test/parallel/test-stream-iter-duplex.js
@@ -3,7 +3,7 @@
 
 const common = require('../common');
 const assert = require('assert');
-const { duplex, text, bytes } = require('stream/iter');
+const { bytes, drain, duplex, text } = require('stream/iter');
 
 // =============================================================================
 // Basic duplex
@@ -162,8 +162,7 @@ async function testChannelFail() {
   const [a, b] = duplex();
   a.writer.fail(new Error('channel failed'));
   await assert.rejects(async () => {
-    // eslint-disable-next-line no-unused-vars
-    for await (const _ of b.readable) { /* consume */ }
+    await drain(b.readable);
   }, { message: 'channel failed' });
   await b.close();
 }
diff --git a/test/parallel/test-stream-iter-pull-async.js b/test/parallel/test-stream-iter-pull-async.js
index 3159b718f27e..a571f7892811 100644
--- a/test/parallel/test-stream-iter-pull-async.js
+++ b/test/parallel/test-stream-iter-pull-async.js
@@ -3,7 +3,7 @@
 
 const common = require('../common');
 const assert = require('assert');
-const { pull, from, text, tap } = require('stream/iter');
+const { drain, from, pull, tap, text } = require('stream/iter');
 
 async function testPullIdentity() {
   const data = await text(pull(from('hello-async')));
@@ -99,10 +99,7 @@ async function testTransformSignalListenerErrorOnSourceError() {
 
   await assert.rejects(
     async () => {
-      // eslint-disable-next-line no-unused-vars
-      for await (const _ of pull(failingSource(), throwingTransform)) {
-        // Consume
-      }
+      await drain(pull(failingSource(), throwingTransform));
     },
     { message: 'source error' },
   );
@@ -122,8 +119,7 @@ async function testPullSourceError() {
     throw new Error('source boom');
   }
   await assert.rejects(async () => {
-    // eslint-disable-next-line no-unused-vars
-    for await (const _ of pull(failingSource())) { /* consume */ }
+    await drain(pull(failingSource()));
   }, { message: 'source boom' });
 }
 
@@ -131,8 +127,7 @@ async function testPullSourceError() {
 async function testTapCallbackError() {
   const badTap = tap(() => { throw new Error('tap boom'); });
   await assert.rejects(async () => {
-    // eslint-disable-next-line no-unused-vars
-    for await (const _ of pull(from('hello'), badTap)) { /* consume */ }
+    await drain(pull(from('hello'), badTap));
   }, { message: 'tap boom' });
 }
 
@@ -235,8 +230,7 @@ async function testPullStatelessTransformError() {
     throw new Error('async stateless boom');
   };
   await assert.rejects(async () => {
-    // eslint-disable-next-line no-unused-vars
-    for await (const _ of pull(from('hello'), badTransform)) { /* consume */ }
+    await drain(pull(from('hello'), badTransform));
   }, { message: 'async stateless boom' });
 }
 
@@ -251,8 +245,7 @@ async function testPullStatefulTransformError() {
     },
   };
   await assert.rejects(async () => {
-    // eslint-disable-next-line no-unused-vars
-    for await (const _ of pull(from('hello'), badStateful)) { /* consume */ }
+    await drain(pull(from('hello'), badStateful));
   }, { message: 'async stateful boom' });
 }
 
@@ -290,8 +283,7 @@ async function testPullStatelessTransformFlushError() {
     return chunks;
   };
   await assert.rejects(async () => {
-    // eslint-disable-next-line no-unused-vars
-    for await (const _ of pull(from('hello'), badFlush)) { /* consume */ }
+    await drain(pull(from('hello'), badFlush));
   }, { message: 'async flush boom' });
 }
 
diff --git a/test/parallel/test-stream-iter-push-writer.js b/test/parallel/test-stream-iter-push-writer.js
index 8ce555ca9251..c6a6b560f838 100644
--- a/test/parallel/test-stream-iter-push-writer.js
+++ b/test/parallel/test-stream-iter-push-writer.js
@@ -3,7 +3,7 @@
 
 const common = require('../common');
 const assert = require('assert');
-const { push, ondrain, text } = require('stream/iter');
+const { drain, ondrain, push, text } = require('stream/iter');
 
 async function testOndrain() {
   const { writer } = push({ budget: 16384 });
@@ -242,8 +242,7 @@ async function testFail() {
   // Second fail is a no-op (already errored)
   writer.fail(new Error('boom2'));
   await assert.rejects(async () => {
-    // eslint-disable-next-line no-unused-vars
-    for await (const _ of readable) { /* consume */ }
+    await drain(readable);
   }, { message: 'boom' });
 }
 
@@ -252,8 +251,7 @@ async function testEndAsyncReturnValue() {
   writer.writeSync('hello');
   // Start consuming concurrently (end() waits for drain)
   const consume = (async () => {
-    // eslint-disable-next-line no-unused-vars
-    for await (const _ of readable) { /* drain */ }
+    await drain(readable);
   })();
   const total = await writer.end();
   assert.strictEqual(total, 5);
@@ -308,8 +306,7 @@ async function testEndAfterEndSyncWaitsForDrain() {
   await Promise.resolve();
   assert.strictEqual(ended, false);
 
-  // eslint-disable-next-line no-unused-vars
-  for await (const _ of readable) { /* drain */ }
+  await drain(readable);
   assert.strictEqual(await end, 5);
 }
 
@@ -475,8 +472,7 @@ async function testEndIdempotentWhenClosed() {
   await writer.write('hello');
   // Start consuming concurrently (end() waits for drain)
   const consume = (async () => {
-    // eslint-disable-next-line no-unused-vars
-    for await (const _ of readable) { /* drain */ }
+    await drain(readable);
   })();
   const first = await writer.end();
   assert.strictEqual(first, 5);
@@ -495,8 +491,7 @@ async function testAsyncDispose() {
   assert.strictEqual(writer.writeSync('fail'), false);
   // Drain readable
   try {
-    // eslint-disable-next-line no-unused-vars
-    for await (const _ of readable) { /* consume */ }
+    await drain(readable);
   } catch {
     // Expected - reader sees the error
   }
@@ -511,8 +506,7 @@ async function testSyncDispose() {
   assert.strictEqual(writer.writeSync('fail'), false);
   // Drain readable
   try {
-    // eslint-disable-next-line no-unused-vars
-    for await (const _ of readable) { /* consume */ }
+    await drain(readable);
   } catch {
     // Expected
   }
diff --git a/test/parallel/test-stream-iter-share-async.js b/test/parallel/test-stream-iter-share-async.js
index c96a0cb0f3c3..b0c522763415 100644
--- a/test/parallel/test-stream-iter-share-async.js
+++ b/test/parallel/test-stream-iter-share-async.js
@@ -4,6 +4,7 @@
 const common = require('../common');
 const assert = require('assert');
 const {
+  drain,
   from,
   share,
   text,
@@ -263,12 +264,10 @@ async function testShareSourceError() {
   const c2 = shared.pull();
 
   await assert.rejects(async () => {
-    // eslint-disable-next-line no-unused-vars
-    for await (const _ of c1) { /* consume */ }
+    await drain(c1);
   }, { message: 'share source boom' });
   await assert.rejects(async () => {
-    // eslint-disable-next-line no-unused-vars
-    for await (const _ of c2) { /* consume */ }
+    await drain(c2);
   }, { message: 'share source boom' });
 }
 
diff --git a/test/parallel/test-stream-iter-share-coverage.js b/test/parallel/test-stream-iter-share-coverage.js
index 48866e61deab..56116715fe11 100644
--- a/test/parallel/test-stream-iter-share-coverage.js
+++ b/test/parallel/test-stream-iter-share-coverage.js
@@ -9,12 +9,13 @@ const assert = require('assert');
 const {
   Share,
   SyncShare,
+  drain,
+  from,
+  fromSync,
   share,
-  shareSync,
   shareProtocol,
+  shareSync,
   shareSyncProtocol,
-  from,
-  fromSync,
   text,
   textSync,
 } = require('stream/iter');
@@ -97,8 +98,7 @@ async function testShareSourceThrowsNonError() {
   const shared = share(source());
   const consumer = shared.pull();
   await assert.rejects(async () => {
-    // eslint-disable-next-line no-unused-vars
-    for await (const batch of consumer) { /* consume */ }
+    await drain(consumer);
   }, { code: 'ERR_OPERATION_FAILED' });
 }
 
diff --git a/test/parallel/test-stream-iter-share-from.js b/test/parallel/test-stream-iter-share-from.js
index 806e30876302..0bd1dd9390e1 100644
--- a/test/parallel/test-stream-iter-share-from.js
+++ b/test/parallel/test-stream-iter-share-from.js
@@ -4,14 +4,14 @@
 const common = require('../common');
 const assert = require('assert');
 const {
+  Share,
+  SyncShare,
+  drain,
   from,
   fromSync,
   share,
-  Share,
-  SyncShare,
   text,
   textSync,
-
 } = require('stream/iter');
 
 // =============================================================================
@@ -235,8 +235,7 @@ async function testShareStrictBackpressure() {
   // 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 */ }
+    await drain(fast);
   }, { code: 'ERR_OUT_OF_RANGE' });
 }
 
diff --git a/test/parallel/test-stream-iter-to-readable.js b/test/parallel/test-stream-iter-to-readable.js
index d8287036e7e7..6596384d81f9 100644
--- a/test/parallel/test-stream-iter-to-readable.js
+++ b/test/parallel/test-stream-iter-to-readable.js
@@ -8,6 +8,7 @@ const common = require('../common');
 const assert = require('assert');
 const { Writable } = require('stream');
 const {
+  drain,
   from,
   fromSync,
   pull,
@@ -98,10 +99,7 @@ async function testErrorAsync() {
   const readable = toReadable(gen());
 
   await assert.rejects(async () => {
-    // eslint-disable-next-line no-unused-vars
-    for await (const chunk of readable) {
-      // Consume until error
-    }
+    await drain(readable);
   }, { message: 'source failed' });
 }
 
@@ -314,10 +312,8 @@ async function testSignalAlreadyAborted() {
   const readable = toReadable(gen(), { signal: AbortSignal.abort() });
 
   await assert.rejects(async () => {
-    // eslint-disable-next-line no-unused-vars
-    for await (const chunk of readable) {
-      // Should not receive any data
-    }
+    // Should not receive any data
+    await drain(readable);
   }, { name: 'AbortError' });
   assert.ok(readable.destroyed);
 }
@@ -465,10 +461,7 @@ async function testErrorSync() {
   const readable = toReadableSync(gen());
 
   await assert.rejects(async () => {
-    // eslint-disable-next-line no-unused-vars
-    for await (const chunk of readable) {
-      // Consume until error
-    }
+    await drain(readable);
   }, { message: 'sync source failed' });
 }
 
diff --git a/test/parallel/test-stream-iter-transform-output.js b/test/parallel/test-stream-iter-transform-output.js
index 90261a33785a..352dd7c2d54c 100644
--- a/test/parallel/test-stream-iter-transform-output.js
+++ b/test/parallel/test-stream-iter-transform-output.js
@@ -7,12 +7,13 @@
 const common = require('../common');
 const assert = require('assert');
 const {
-  pull,
-  pullSync,
   bytes,
   bytesSync,
+  drain,
   from,
   fromSync,
+  pull,
+  pullSync,
 } = require('stream/iter');
 
 // Stateless transform returns ArrayBuffer (async)
@@ -236,8 +237,7 @@ async function testTransformReturnsInvalidType() {
   };
   await assert.rejects(
     async () => {
-      // eslint-disable-next-line no-unused-vars
-      for await (const batch of pull(from('x'), tx)) { /* consume */ }
+      await drain(pull(from('x'), tx));
     },
     { code: 'ERR_INVALID_ARG_TYPE' },
   );