stream: add drain()/drainSync() for stream/iter

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

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

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,

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,

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.

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.

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();

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();

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]);
 

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();

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] } },

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;