Commit a7c156fd377 for nodejs
commit a7c156fd37724fd5cbefb1d6592a722618f29377
Author: James M Snell <jasnell@gmail.com>
Date: Sun Oct 4 02:35:56 2026 +0000
stream: construct the options passed to stream/iter transforms
pull() passes every stateless transform call a new
`{ __proto__: null, signal }` options object, which V8 creates in
dictionary mode, once per batch and transform.
Create the options with a TransformOptions constructor instead, for
stateful transforms as well so that both forms receive the same kind of
object. Each call still gets its own object, as the pipeline requires.
Its prototype is a single frozen object with no %Object.prototype% in
its chain, so a transform cannot pass state to other transforms through
it, and with a non-enumerable `constructor` so that util.inspect()
prints `TransformOptions { signal }`.
With 300,000 single-chunk batches, pull() is about 2% faster with one
stateless transform and about 8% faster with four.
This is observable: the options object's prototype is no longer null.
A new test covers the options contract, and the documentation now
describes it, including that pullSync() passes transforms no options.
Assisted-by: OpenCode
Signed-off-by: James M Snell <jasnell@gmail.com>
PR-URL: https://github.com/nodejs/node/pull/66648
Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>
Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day>
diff --git a/doc/api/stream_iter.md b/doc/api/stream_iter.md
index e2f30ce6a94..893dcfbf4bc 100644
--- a/doc/api/stream_iter.md
+++ b/doc/api/stream_iter.md
@@ -140,6 +140,12 @@ Both forms receive an `options` parameter with the following property:
can check `signal.aborted` or listen for the `'abort'` event to perform
early cleanup.
+In `pull()`, stateless transforms receive a new `options` object for every
+call, and stateful transforms one for the pipeline, so a transform can modify
+its `options` without affecting other transforms. The object does not inherit
+from `Object.prototype`. Transforms passed to [`pullSync()`][] receive no
+`options`.
+
The flush signal (`null`) is sent after the source ends, giving transforms
a chance to emit trailing data (e.g., compression footers).
diff --git a/lib/internal/streams/iter/pull.js b/lib/internal/streams/iter/pull.js
index 9a52a5fdfed..35836dd27d6 100644
--- a/lib/internal/streams/iter/pull.js
+++ b/lib/internal/streams/iter/pull.js
@@ -11,6 +11,8 @@ const {
ArrayPrototypePush,
ArrayPrototypeSlice,
FunctionPrototypeCall,
+ ObjectDefineProperty,
+ ObjectFreeze,
ObjectSetPrototypeOf,
PromisePrototypeThen,
PromiseReject,
@@ -564,6 +566,19 @@ function* createSyncPipeline(source, transforms) {
// Async Pipeline Implementation
// =============================================================================
+// The options object passed to async transforms. Stateless transforms get a
+// new one for every call, so it is constructed rather than created as a
+// `{ __proto__: null, signal }` literal (a dictionary-mode object). The
+// prototype is frozen and has no %Object.prototype% in its chain, and the
+// non-enumerable `constructor` lets util.inspect() print the options as
+// `TransformOptions { signal }`.
+function TransformOptions(signal) {
+ this.signal = signal;
+}
+TransformOptions.prototype = ObjectFreeze(ObjectDefineProperty(
+ { __proto__: null }, 'constructor',
+ { __proto__: null, value: TransformOptions }));
+
/**
* Apply a single stateless async transform to a source.
* @yields {Uint8Array[]}
@@ -574,7 +589,7 @@ function* createSyncPipeline(source, transforms) {
* avoiding the overhead of N async generator ticks for N transforms.
*
* INVARIANT: This function accepts a signal, NOT a pre-built options object.
- * A fresh { __proto__: null, signal } options object is created for each
+ * A fresh TransformOptions object is created for each
* transform invocation to prevent cross-transform mutation.
* @param {AsyncIterable<Uint8Array[]>} source
* @param {Array<Function>} run - Array of stateless transform functions
@@ -585,7 +600,7 @@ async function* applyFusedStatelessAsyncTransforms(source, run, signal) {
for await (const chunks of source) {
let current = chunks;
for (let i = 0; i < run.length; i++) {
- let result = run[i](current, { __proto__: null, signal });
+ let result = run[i](current, new TransformOptions(signal));
if (isPromise(result)) result = await result;
if (result === null) {
current = null;
@@ -628,14 +643,14 @@ async function* applyFusedStatelessAsyncTransforms(source, run, signal) {
for (let j = 0; j < pending.length; j++) {
const pendingResult = appendTransformResultAsync(
next,
- run[i](pending[j], { __proto__: null, signal }));
+ run[i](pending[j], new TransformOptions(signal)));
if (pendingResult !== undefined) {
await pendingResult;
}
}
const flushResult = appendTransformResultAsync(
next,
- run[i](null, { __proto__: null, signal }));
+ run[i](null, new TransformOptions(signal)));
if (flushResult !== undefined) {
await flushResult;
}
@@ -742,7 +757,7 @@ async function* createAsyncPipeline(source, transforms, signal) {
// generator layer to avoid unnecessary async generator ticks.
//
// INVARIANT: Each transform invocation MUST receive its own fresh options
- // object ({ __proto__: null, signal }). Transforms may mutate the options
+ // object (new TransformOptions(signal)). Transforms may mutate the options
// object, so sharing a single object across invocations would allow one
// transform to corrupt the options seen by another. The signal is shared
// across calls (mutations to it are acceptable), but the containing options
@@ -763,7 +778,7 @@ async function* createAsyncPipeline(source, transforms, signal) {
transformSignal);
statelessRun = [];
}
- const opts = { __proto__: null, signal: transformSignal };
+ const opts = new TransformOptions(transformSignal);
if (transform[kValidatedTransform]) {
current = applyValidatedStatefulAsyncTransform(
current, transform.transform, transform.receiver, opts);
diff --git a/test/parallel/test-stream-iter-pull-async.js b/test/parallel/test-stream-iter-pull-async.js
index 71fc27426ea..5bffd49d511 100644
--- a/test/parallel/test-stream-iter-pull-async.js
+++ b/test/parallel/test-stream-iter-pull-async.js
@@ -16,6 +16,7 @@ const {
} = require('stream/iter');
const { setImmediate } = require('timers/promises');
+const { inspect } = require('util');
async function testPullIdentity() {
const data = await text(pull(from('hello-async')));
@@ -573,6 +574,41 @@ async function testTransformOptionsNotShared() {
assert.strictEqual(seen[1].mutated, undefined);
}
+// Stateless transforms get a new options object for every call, and stateful
+// transforms one for the pipeline. The options object has only `signal`, does
+// not inherit from Object.prototype, and its prototype is frozen, so that a
+// transform cannot pass state to others through it.
+async function testTransformOptionsShape() {
+ const seen = [];
+ const stateless = (chunks, options) => {
+ seen.push(options);
+ return chunks;
+ };
+ const stateful = {
+ async* transform(source, options) {
+ seen.push(options);
+ for await (const chunks of source) yield chunks;
+ },
+ };
+ const ac = new AbortController();
+ await text(pull(from(['a', 'b']), stateless, stateful,
+ { signal: ac.signal }));
+ // Stateless: one call per batch plus the flush call.
+ assert.strictEqual(seen.length, 4);
+ assert.strictEqual(new Set(seen).size, seen.length);
+ for (const options of seen) {
+ assert.strictEqual(options instanceof Object, false);
+ assert.deepStrictEqual(Object.keys(options), ['signal']);
+ assert.ok(options.signal instanceof AbortSignal);
+ assert.strictEqual(Object.isFrozen(Object.getPrototypeOf(options)), true);
+ assert.match(inspect(options), /^TransformOptions \{ signal: /);
+ }
+ assert.strictEqual(Object.getPrototypeOf(seen[0]),
+ Object.getPrototypeOf(seen[3]));
+ assert.throws(() => { Object.getPrototypeOf(seen[0]).leak = true; },
+ TypeError);
+}
+
// Run the uncaughtException test sequentially (it installs a global handler
// that would interfere with concurrent tests).
(async () => {
@@ -609,6 +645,7 @@ async function testTransformOptionsNotShared() {
testTransformReturnsArrayBuffer(),
testPipeToStringSource(),
testTransformOptionsNotShared(),
+ testTransformOptionsShape(),
]);
// Run after all concurrent tests complete to avoid global handler races
await testTransformSignalListenerErrorOnSourceError();