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-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-drain.mjs b/test/parallel/test-stream-iter-consumers-drain.mjs
new file mode 100644
index 000000000000..f658246ee2e4
--- /dev/null
+++ b/test/parallel/test-stream-iter-consumers-drain.mjs
@@ -0,0 +1,424 @@
+// Flags: --experimental-stream-iter --expose-gc
+
+import { mustCall } from '../common/index.mjs';
+import { createRequire } from 'node:module';
+import assert from 'node:assert';
+import {
+  array,
+  drain,
+  drainSync,
+  from,
+  fromSync,
+  pull,
+  push,
+  tap,
+  toAsyncStreamable,
+} from 'node:stream/iter';
+
+const require = createRequire(import.meta.url);
+const { gcUntil } = require('../common/gc');
+
+// =============================================================================
+// 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() {
+  // Allocate a distinct buffer per chunk and keep only weak references to
+  // them. After the consumer finishes, anything it retained is still
+  // reachable; anything it discarded is collectable.
+  const kChunks = 64;
+
+  async function measure(consumer, keepResult) {
+    const refs = [];
+    async function* source() {
+      for (let i = 0; i < kChunks; i++) {
+        const buf = new Uint8Array(64 * 1024);
+        refs.push(new WeakRef(buf));
+        yield [buf];
+      }
+    }
+    const result = await consumer(source());
+    let live = kChunks;
+    const settled = await gcUntil('drain retention', () => {
+      live = refs.filter((ref) => ref.deref() !== undefined).length;
+      return live === 0;
+    }).then(() => true, () => false);
+    // Keep the collecting consumer's result reachable across the GC, so its
+    // retention is what is being measured rather than the result being dropped.
+    if (keepResult) assert.strictEqual(result.length, kChunks);
+    return { settled, live };
+  }
+
+  // drain() keeps nothing, so every chunk becomes collectable.
+  const drained = await measure(drain, false);
+  assert.ok(drained.settled,
+            `drain() retained ${drained.live}/${kChunks} chunks`);
+
+  // Positive control: array() does retain, which proves the measurement above
+  // is capable of observing retention rather than always passing.
+  const collected = await measure(array, true);
+  assert.ok(collected.live === kChunks,
+            `array() should retain every chunk, kept ${collected.live}`);
+}
+
+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-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-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);
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' },
   );