stream: add stream/iter Implementation

Experimental implementation of https://stream-iter.jasnell.me/

Signed-off-by: James M Snell <jasnell@gmail.com>
Assisted-By: Claude/Opus 4.6
PR-URL: #62066
Backport-PR-URL: #64675
Reviewed-By: Robert Nagy <ronagy@icloud.com>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>

Author: jasnell

doc/api/cli.md

@@ -1203,6 +1203,16 @@ added:
 
 Enable experimental support for storage inspection
 
+### `--experimental-stream-iter`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+Enable the experimental [`node:stream/iter`][] module.
+
 ### `--experimental-test-coverage`
 
 <!-- YAML
@@ -3574,6 +3584,7 @@ one is included in the list below.
 * `--experimental-require-module`
 * `--experimental-shadow-realm`
 * `--experimental-specifier-resolution`
+* `--experimental-stream-iter`
 * `--experimental-test-isolation`
 * `--experimental-top-level-await`
 * `--experimental-transform-types`
@@ -4212,6 +4223,7 @@ node --stack-trace-limit=12 -p -e "Error.stackTraceLimit" # prints 12
 [`import` specifier]: esm.md#import-specifiers
 [`net.getDefaultAutoSelectFamilyAttemptTimeout()`]: net.md#netgetdefaultautoselectfamilyattempttimeout
 [`node:sqlite`]: sqlite.md
+[`node:stream/iter`]: stream_iter.md
 [`process.setUncaughtExceptionCaptureCallback()`]: process.md#processsetuncaughtexceptioncapturecallbackfn
 [`tls.DEFAULT_MAX_VERSION`]: tls.md#tlsdefault_max_version
 [`tls.DEFAULT_MIN_VERSION`]: tls.md#tlsdefault_min_version

doc/api/fs.md

@@ -377,6 +377,154 @@ added: v10.0.0
 
 * Type: {number} The numeric file descriptor managed by the {FileHandle} object.
 
+#### `filehandle.pull([...transforms][, options])`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+* `...transforms` {Function|Object} Optional transforms to apply via
+  [`stream/iter pull()`][].
+* `options` {Object}
+  * `signal` {AbortSignal}
+  * `autoClose` {boolean} Close the file handle when the stream ends.
+    **Default:** `false`.
+  * `start` {number} Byte offset to begin reading from. When specified,
+    reads use explicit positioning (`pread` semantics). **Default:** current
+    file position.
+  * `limit` {number} Maximum number of bytes to read before ending the
+    iterator. Reads stop when `limit` bytes have been delivered or EOF is
+    reached, whichever comes first. **Default:** read until EOF.
+  * `chunkSize` {number} Size in bytes of the buffer allocated for each
+    read operation. **Default:** `131072` (128 KB).
+* Returns: {AsyncIterable\<Uint8Array\[]>}
+
+Return the file contents as an async iterable using the
+[`node:stream/iter`][] pull model. Reads are performed in `chunkSize`-byte
+chunks (default 128 KB). If transforms are provided, they are applied
+via [`stream/iter pull()`][].
+
+The file handle is locked while the iterable is being consumed and unlocked
+when iteration completes, an error occurs, or the consumer breaks.
+
+This function is only available when the `--experimental-stream-iter` flag is
+enabled.
+
+```mjs
+import { open } from 'node:fs/promises';
+import { text } from 'node:stream/iter';
+import { compressGzip } from 'node:zlib/iter';
+
+const fh = await open('input.txt', 'r');
+
+// Read as text
+console.log(await text(fh.pull({ autoClose: true })));
+
+// Read 1 KB starting at byte 100
+const fh2 = await open('input.txt', 'r');
+console.log(await text(fh2.pull({ start: 100, limit: 1024, autoClose: true })));
+
+// Read with compression
+const fh3 = await open('input.txt', 'r');
+const compressed = fh3.pull(compressGzip(), { autoClose: true });
+```
+
+```cjs
+const { open } = require('node:fs/promises');
+const { text } = require('node:stream/iter');
+const { compressGzip } = require('node:zlib/iter');
+
+async function run() {
+  const fh = await open('input.txt', 'r');
+
+  // Read as text
+  console.log(await text(fh.pull({ autoClose: true })));
+
+  // Read 1 KB starting at byte 100
+  const fh2 = await open('input.txt', 'r');
+  console.log(await text(fh2.pull({ start: 100, limit: 1024, autoClose: true })));
+
+  // Read with compression
+  const fh3 = await open('input.txt', 'r');
+  const compressed = fh3.pull(compressGzip(), { autoClose: true });
+}
+
+run().catch(console.error);
+```
+
+#### `filehandle.pullSync([...transforms][, options])`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+* `...transforms` {Function|Object} Optional transforms to apply via
+  [`stream/iter pullSync()`][].
+* `options` {Object}
+  * `autoClose` {boolean} Close the file handle when the stream ends.
+    **Default:** `false`.
+  * `start` {number} Byte offset to begin reading from. When specified,
+    reads use explicit positioning. **Default:** current file position.
+  * `limit` {number} Maximum number of bytes to read before ending the
+    iterator. **Default:** read until EOF.
+  * `chunkSize` {number} Size in bytes of the buffer allocated for each
+    read operation. **Default:** `131072` (128 KB).
+* Returns: {Iterable\<Uint8Array\[]>}
+
+Synchronous counterpart of [`filehandle.pull()`][]. Returns a sync iterable
+that reads the file using synchronous I/O on the main thread. Reads are
+performed in `chunkSize`-byte chunks (default 128 KB).
+
+The file handle is locked while the iterable is being consumed. Unlike the
+async `pull()`, this method does not support `AbortSignal` since all
+operations are synchronous.
+
+This function is only available when the `--experimental-stream-iter` flag is
+enabled.
+
+```mjs
+import { open } from 'node:fs/promises';
+import { textSync, pipeToSync } from 'node:stream/iter';
+import { compressGzipSync, decompressGzipSync } from 'node:zlib/iter';
+
+const fh = await open('input.txt', 'r');
+
+// Read as text (sync)
+console.log(textSync(fh.pullSync({ autoClose: true })));
+
+// Sync compress pipeline: file -> gzip -> file
+const src = await open('input.txt', 'r');
+const dst = await open('output.gz', 'w');
+pipeToSync(src.pullSync(compressGzipSync(), { autoClose: true }), dst.writer({ autoClose: true }));
+```
+
+```cjs
+const { open } = require('node:fs/promises');
+const { textSync, pipeToSync } = require('node:stream/iter');
+const { compressGzipSync, decompressGzipSync } = require('node:zlib/iter');
+
+async function run() {
+  const fh = await open('input.txt', 'r');
+
+  // Read as text (sync)
+  console.log(textSync(fh.pullSync({ autoClose: true })));
+
+  // Sync compress pipeline: file -> gzip -> file
+  const src = await open('input.txt', 'r');
+  const dst = await open('output.gz', 'w');
+  pipeToSync(
+    src.pullSync(compressGzipSync(), { autoClose: true }),
+    dst.writer({ autoClose: true }),
+  );
+}
+
+run().catch(console.error);
+```
+
 #### `filehandle.read(buffer, offset, length, position)`
 
 <!-- YAML
@@ -905,6 +1053,121 @@ On Linux, positional writes don't work when the file is opened in append mode.
 The kernel ignores the position argument and always appends the data to
 the end of the file.
 
+#### `filehandle.writer([options])`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+* `options` {Object}
+  * `autoClose` {boolean} Close the file handle when the writer ends or
+    fails. **Default:** `false`.
+  * `start` {number} Byte offset to start writing at. When specified,
+    writes use explicit positioning. **Default:** current file position.
+  * `limit` {number} Maximum number of bytes the writer will accept.
+    Async writes (`write()`, `writev()`) that would exceed the limit reject
+    with `ERR_OUT_OF_RANGE`. Sync writes (`writeSync()`, `writevSync()`)
+    return `false`. **Default:** no limit.
+  * `chunkSize` {number} Maximum chunk size in bytes for synchronous write
+    operations. Writes larger than this threshold fall back to async I/O.
+    Set this to match the reader's `chunkSize` for optimal `pipeTo()`
+    performance. **Default:** `131072` (128 KB).
+* Returns: {Object}
+  * `write(chunk[, options])` {Function} Returns {Promise\<void>}.
+    Accepts `Uint8Array`, `Buffer`, or string (UTF-8 encoded).
+    * `chunk` {Buffer|TypedArray|DataView|string}
+    * `options` {Object}
+      * `signal` {AbortSignal} If the signal is already aborted, the write
+        rejects with `AbortError` without performing I/O.
+  * `writev(chunks[, options])` {Function} Returns {Promise\<void>}. Uses
+    scatter/gather I/O via a single `writev()` syscall. Accepts mixed
+    `Uint8Array`/string arrays.
+    * `chunks` {Array\<Buffer|TypedArray|DataView|string>}
+    * `options` {Object}
+      * `signal` {AbortSignal} If the signal is already aborted, the write
+        rejects with `AbortError` without performing I/O.
+  * `writeSync(chunk)` {Function} Returns {boolean}. Attempts a synchronous
+    write. Returns `true` if the write succeeded, `false` if the caller
+    should fall back to async `write()`. Returns `false` when: the writer
+    is closed/errored, an async operation is in flight, the chunk exceeds
+    `chunkSize`, or the write would exceed `limit`.
+    * `chunk` {Buffer|TypedArray|DataView|string}
+  * `writevSync(chunks)` {Function} Returns {boolean}. Synchronous batch
+    write. Same fallback semantics as `writeSync()`.
+    * `chunks` {Array\<Buffer|TypedArray|DataView|string>}
+  * `end([options])` {Function} Returns {Promise\<number>} total bytes
+    written. Idempotent: returns `totalBytesWritten` if already closed,
+    returns the pending promise if already closing. Rejects if the writer
+    is in an errored state.
+    * `options` {Object}
+      * `signal` {AbortSignal} If the signal is already aborted, `end()`
+        rejects with `AbortError` and the writer remains open.
+  * `endSync()` {Function} Returns {number|number} total bytes written on
+    success, `-1` if the writer is errored or an async operation is in
+    flight. Idempotent when already closed.
+  * `fail(reason)` {Function} Puts the writer into a terminal error state.
+    Synchronous. If the writer is already closed or errored, this is a
+    no-op. If `autoClose` is true, closes the file handle synchronously.
+
+Return a [`node:stream/iter`][] writer backed by this file handle.
+
+The writer supports both `Symbol.asyncDispose` and `Symbol.dispose`:
+
+* `await using w = fh.writer()` — if the writer is still open (no `end()`
+  called), `asyncDispose` calls `fail()`. If `end()` is pending, it waits
+  for it to complete.
+* `using w = fh.writer()` — calls `fail()` unconditionally.
+
+The `writeSync()` and `writevSync()` methods enable the try-sync fast path
+used by [`stream/iter pipeTo()`][]. When the reader's chunk size matches the
+writer's `chunkSize`, all writes in a `pipeTo()` pipeline complete
+synchronously with zero promise overhead.
+
+This function is only available when the `--experimental-stream-iter` flag is
+enabled.
+
+```mjs
+import { open } from 'node:fs/promises';
+import { from, pipeTo } from 'node:stream/iter';
+import { compressGzip } from 'node:zlib/iter';
+
+// Async pipeline
+const fh = await open('output.gz', 'w');
+await pipeTo(from('Hello!'), compressGzip(), fh.writer({ autoClose: true }));
+
+// Sync pipeline with limit
+const src = await open('input.txt', 'r');
+const dst = await open('output.txt', 'w');
+const w = dst.writer({ limit: 1024 * 1024 }); // Max 1 MB
+await pipeTo(src.pull({ autoClose: true }), w);
+await w.end();
+await dst.close();
+```
+
+```cjs
+const { open } = require('node:fs/promises');
+const { from, pipeTo } = require('node:stream/iter');
+const { compressGzip } = require('node:zlib/iter');
+
+async function run() {
+  // Async pipeline
+  const fh = await open('output.gz', 'w');
+  await pipeTo(from('Hello!'), compressGzip(), fh.writer({ autoClose: true }));
+
+  // Sync pipeline with limit
+  const src = await open('input.txt', 'r');
+  const dst = await open('output.txt', 'w');
+  const w = dst.writer({ limit: 1024 * 1024 }); // Max 1 MB
+  await pipeTo(src.pull({ autoClose: true }), w);
+  await w.end();
+  await dst.close();
+}
+
+run().catch(console.error);
+```
+
 #### `filehandle[Symbol.asyncDispose]()`
 
 <!-- YAML
@@ -8948,6 +9211,7 @@ the file contents.
 [`event ports`]: https://illumos.org/man/port_create
 [`filehandle.createReadStream()`]: #filehandlecreatereadstreamoptions
 [`filehandle.createWriteStream()`]: #filehandlecreatewritestreamoptions
+[`filehandle.pull()`]: #filehandlepulltransforms-options
 [`filehandle.writeFile()`]: #filehandlewritefiledata-options
 [`fs.access()`]: #fsaccesspath-mode-callback
 [`fs.accessSync()`]: #fsaccesssyncpath-mode
@@ -8998,7 +9262,11 @@ the file contents.
 [`inotify(7)`]: https://man7.org/linux/man-pages/man7/inotify.7.html
 [`kqueue(2)`]: https://www.freebsd.org/cgi/man.cgi?query=kqueue&sektion=2
 [`minimatch`]: https://github.com/isaacs/minimatch
+[`node:stream/iter`]: stream_iter.md
 [`statfs.bsize`]: #statfsbsize
+[`stream/iter pipeTo()`]: stream_iter.md#pipetosource-transforms-writer
+[`stream/iter pull()`]: stream_iter.md#pullsource-transforms-options
+[`stream/iter pullSync()`]: stream_iter.md#pullsyncsource-transforms
 [`util.promisify()`]: util.md#utilpromisifyoriginal
 [bigints]: https://tc39.github.io/proposal-bigint
 [caveats]: #caveats

doc/api/index.md

@@ -43,6 +43,7 @@
 * [Modules: Packages](packages.md)
 * [Modules: TypeScript](typescript.md)
 * [Net](net.md)
+* [Iterable Streams API](stream_iter.md)
 * [OS](os.md)
 * [Path](path.md)
 * [Performance hooks](perf_hooks.md)
@@ -72,6 +73,7 @@
 * [Web Streams API](webstreams.md)
 * [Worker threads](worker_threads.md)
 * [Zlib](zlib.md)
+* [Zlib Iterable Compression](zlib_iter.md)
 
 <hr class="line"/>