quic: add proper inbound/outbound for stream

Author: jasnell

doc/api/errors.md

@@ -2633,6 +2633,19 @@ added:
 
 Opening a QUIC stream failed.
 
+<a id="ERR_QUIC_STREAM_RESET"></a>
+
+### `ERR_QUIC_STREAM_RESET`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+A QUIC stream was reset by the peer. The error includes the reset code
+provided by the peer.
+
 <a id="ERR_QUIC_TRANSPORT_ERROR"></a>
 
 ### `ERR_QUIC_TRANSPORT_ERROR`

doc/api/quic.md

@@ -1149,13 +1149,85 @@ Sets the priority of the stream. Throws `ERR_INVALID_STATE` if the session
 does not support priority (e.g. non-HTTP/3). Has no effect if the stream
 has been destroyed.
 
-### `stream.readable`
+### `stream[Symbol.asyncIterator]()`
 
 <!-- YAML
-added: v23.8.0
+added: REPLACEME
 -->
 
-* Type: {ReadableStream}
+* Returns: {AsyncIterableIterator} yielding {Uint8Array\[]}
+
+The stream implements `Symbol.asyncIterator`, making it directly usable
+in `for await...of` loops. Each iteration yields a batch of `Uint8Array`
+chunks.
+
+Only one async iterator can be obtained per stream. A second call throws
+`ERR_INVALID_STATE`. Non-readable streams (outbound-only unidirectional
+or closed) return an immediately-finished iterator.
+
+```mjs
+for await (const chunks of stream) {
+  for (const chunk of chunks) {
+    // Process each Uint8Array chunk
+  }
+}
+```
+
+Compatible with stream/iter utilities:
+
+```mjs
+import Stream from 'node:stream/iter';
+const body = await Stream.bytes(stream);
+const text = await Stream.text(stream);
+await Stream.pipeTo(stream, someWriter);
+```
+
+### `stream.writer`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* Type: {Object}
+
+Returns a Writer object for pushing data to the stream incrementally.
+The Writer implements the stream/iter Writer interface with the
+try-sync-fallback-to-async pattern.
+
+Only available when no `body` source was provided at creation time or via
+[`stream.setBody()`][]. Non-writable streams return an already-closed
+Writer. Throws `ERR_INVALID_STATE` if the outbound is already configured.
+
+The Writer has the following methods:
+
+* `writeSync(chunk)` — Synchronous write. Returns `true` if accepted,
+  `false` if flow-controlled. Data is NOT accepted on `false`.
+* `write(chunk[, options])` — Async write with drain wait. `options.signal`
+  is checked at entry but not observed during the write.
+* `writevSync(chunks)` — Synchronous vectored write. All-or-nothing.
+* `writev(chunks[, options])` — Async vectored write.
+* `endSync()` — Synchronous close. Returns total bytes or `-1`.
+* `end([options])` — Async close.
+* `fail(reason)` — Errors the stream (sends RESET\_STREAM to peer).
+* `desiredSize` — Available capacity in bytes, or `null` if closed/errored.
+
+### `stream.setBody(body)`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* `body` {string|ArrayBuffer|SharedArrayBuffer|TypedArray|Blob|AsyncIterable|Iterable|Promise|null}
+
+Sets the outbound body source for the stream. Can only be called once.
+Mutually exclusive with [`stream.writer`][].
+
+If `body` is `null`, the writable side is closed immediately (FIN sent).
+If `body` is a `Promise`, it is awaited and the resolved value is used.
+Other types are handled per their optimization tier (see below).
+
+Throws `ERR_INVALID_STATE` if the outbound is already configured or if
+the writer has been accessed.
 
 ### `stream.session`
 
@@ -2235,4 +2307,6 @@ added: v23.8.0
 [`stream.onwanttrailers`]: #streamonwanttrailers
 [`stream.pendingTrailers`]: #streampendingtrailers
 [`stream.sendTrailers()`]: #streamsendtrailersheaders
+[`stream.setBody()`]: #streamsetbodybody
 [`stream.setPriority()`]: #streamsetpriorityoptions
+[`stream.writer`]: #streamwriter

lib/internal/blob.js

@@ -2,6 +2,7 @@
 
 const {
   ArrayFrom,
+  ArrayPrototypePush,
   MathMax,
   MathMin,
   ObjectDefineProperties,
@@ -523,13 +524,75 @@ function createBlobReaderStream(reader) {
   }, { highWaterMark: 0 });
 }
 
+// Maximum number of chunks to collect in a single batch to prevent
+// unbounded memory growth when the DataQueue has a large burst of data.
+const kMaxBatchChunks = 16;
+
+async function* createBlobReaderIterable(reader, options = {}) {
+  const { getReadError } = options;
+  let wakeup = PromiseWithResolvers();
+  reader.setWakeup(wakeup.resolve);
+
+  try {
+    while (true) {
+      const batch = [];
+      let blocked = false;
+      let eos = false;
+      let error = null;
+
+      // Pull as many chunks as available synchronously.
+      // reader.pull(callback) calls the callback synchronously via
+      // MakeCallback, so we can collect multiple chunks per iteration
+      // step without any async overhead.
+      while (true) {
+        let pullResult;
+        reader.pull((status, buffer) => {
+          pullResult = { status, buffer };
+        });
+
+        if (pullResult.status === 0) {
+          eos = true;
+          break;
+        }
+        if (pullResult.status < 0) {
+          error = typeof getReadError === 'function' ?
+            getReadError(pullResult.status) :
+            new ERR_INVALID_STATE('The reader is not readable');
+          break;
+        }
+        if (pullResult.status === 2) {
+          blocked = true;
+          break;
+        }
+        ArrayPrototypePush(batch, new Uint8Array(pullResult.buffer));
+        if (batch.length >= kMaxBatchChunks) break;
+      }
+
+      if (batch.length > 0) {
+        yield batch;
+      }
+
+      if (eos) return;
+      if (error) throw error;
+
+      if (blocked) {
+        await wakeup.promise;
+        wakeup = PromiseWithResolvers();
+      }
+    }
+  } finally {
+    reader.setWakeup(undefined);
+  }
+}
+
 module.exports = {
   Blob,
   createBlob,
   createBlobFromFilePath,
+  createBlobReaderIterable,
+  createBlobReaderStream,
   isBlob,
   kHandle,
   resolveObjectURL,
   TransferableBlob,
-  createBlobReaderStream,
 };

lib/internal/errors.js

@@ -1686,6 +1686,8 @@ E('ERR_QUIC_APPLICATION_ERROR', 'A QUIC application error occurred. %d [%s]', Er
 E('ERR_QUIC_CONNECTION_FAILED', 'QUIC connection failed', Error);
 E('ERR_QUIC_ENDPOINT_CLOSED', 'QUIC endpoint closed: %s (%d)', Error);
 E('ERR_QUIC_OPEN_STREAM_FAILED', 'Failed to open QUIC stream', Error);
+E('ERR_QUIC_STREAM_RESET',
+  'The QUIC stream was reset by the peer with error code %d', Error);
 E('ERR_QUIC_TRANSPORT_ERROR', 'A QUIC transport error occurred. %d [%s]', Error);
 E('ERR_QUIC_VERSION_NEGOTIATION_ERROR', 'The QUIC session requires version negotiation', Error);
 E('ERR_REQUIRE_ASYNC_MODULE', function(filename, parentFilename) {