Commit 778306c0018 for nodejs
commit 778306c00186ab53b4364087754ddadef3eaf9ae
Author: James M Snell <jasnell@gmail.com>
Date: Sun Oct 4 01:47:21 2026 +0000
stream: create stream/iter iterator results with a constructor
The iterators of push(), share(), shareSync(), broadcast() consumers
and from() normalization created a `{ __proto__: null, done, value }`
literal for every result. V8 creates such literals in dictionary mode,
which makes them several times more expensive to create and read than
ordinary objects.
Create them with an IterResult constructor whose prototype is a single
frozen, null-prototype object instead. Results have fast properties
and a single shape, and still have no %Object.prototype% in their
prototype chain, so a polluted Object.prototype.then still cannot turn
a result into a thenable. Creating and reading a result is about 5x
faster in a microbenchmark.
benchmark/streams/iter-throughput-share-sync.js improves by 7% to 42%
(more with more consumers) and iter-throughput-share.js by 4-6%;
pipeTo() with small chunks is about 2.5% faster.
This is observable: results are no longer null-prototype objects, so
deepStrictEqual() comparisons against `{ __proto__: null, ... }` no
longer match, and util.inspect() prints them as
`IterResult { done, value }` (the prototype has a non-enumerable
`constructor` for that purpose). The five tests that compared results
that way now compare their own properties, and a new test covers the
result contract and the prototype pollution case.
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/lib/internal/streams/iter/broadcast.js b/lib/internal/streams/iter/broadcast.js
index e862f8d2a7a..df2a701f20c 100644
--- a/lib/internal/streams/iter/broadcast.js
+++ b/lib/internal/streams/iter/broadcast.js
@@ -57,6 +57,7 @@ const {
} = require('internal/streams/iter/pull');
const {
+ IterResult,
kMultiConsumerDefaultBudget,
kResolvedPromise,
convertChunks,
@@ -207,13 +208,13 @@ class BroadcastImpl {
const self = this;
const kDone = PromiseResolve(
- { __proto__: null, done: true, value: undefined });
+ new IterResult(true, undefined));
function detach() {
state.detached = true;
self.#waiters.delete(state);
if (state.resolve) {
- state.resolve({ __proto__: null, done: true, value: undefined });
+ state.resolve(new IterResult(true, undefined));
}
self.#resolvePendingDone(state);
if (self.#deleteConsumer(state)) {
@@ -245,7 +246,7 @@ class BroadcastImpl {
self.#tryTrimBuffer();
}
return PromiseResolve(
- { __proto__: null, done: false, value: chunk });
+ new IterResult(false, chunk));
}
if (self.#errored) {
@@ -307,7 +308,7 @@ class BroadcastImpl {
if (hasReason) {
consumer.reject?.(reason);
} else {
- consumer.resolve({ __proto__: null, done: true, value: undefined });
+ consumer.resolve(new IterResult(true, undefined));
}
consumer.resolve = null;
consumer.reject = null;
@@ -397,9 +398,9 @@ class BroadcastImpl {
--this.#cachedMinCursorConsumers === 0) {
this.#tryTrimBuffer();
}
- consumer.resolve({ __proto__: null, done: false, value: chunk });
+ consumer.resolve(new IterResult(false, chunk));
} else {
- consumer.resolve({ __proto__: null, done: true, value: undefined });
+ consumer.resolve(new IterResult(true, undefined));
this.#resolvePendingDone(consumer);
consumer.detached = true;
}
@@ -529,7 +530,7 @@ class BroadcastImpl {
const resolve = consumer.resolve;
consumer.resolve = null;
consumer.reject = null;
- resolve({ __proto__: null, done: false, value: chunk });
+ resolve(new IterResult(false, chunk));
if (consumer.detached && this.#deleteConsumer(consumer)) {
this.#tryTrimBuffer();
} else if (this.#promotePending(consumer)) {
@@ -573,7 +574,7 @@ class BroadcastImpl {
}
while (consumer.pending.length > 0) {
ArrayPrototypeShift(consumer.pending).resolve(
- { __proto__: null, done: true, value: undefined });
+ new IterResult(true, undefined));
}
}
diff --git a/lib/internal/streams/iter/from.js b/lib/internal/streams/iter/from.js
index 9f39e0f614a..78b61b267de 100644
--- a/lib/internal/streams/iter/from.js
+++ b/lib/internal/streams/iter/from.js
@@ -53,6 +53,7 @@ const {
} = require('internal/streams/iter/types');
const {
+ IterResult,
getProtocolMethod,
toUint8Array,
} = require('internal/streams/iter/utils');
@@ -388,7 +389,7 @@ function yieldNormalizationAbortable(source, context) {
return ObjectSetPrototypeOf({
async next() {
if (completed) {
- return { __proto__: null, done: true, value: undefined };
+ return new IterResult(true, undefined);
}
throwIfNormalizationCancelled(context);
reading = true;
@@ -406,12 +407,12 @@ function yieldNormalizationAbortable(source, context) {
throwIfNormalizationCancelled(context);
completed = true;
closed = true;
- return { __proto__: null, done: true, value: result.value };
+ return new IterResult(true, result.value);
}
const value = result.value;
reading = false;
throwIfNormalizationCancelled(context);
- return { __proto__: null, done: false, value };
+ return new IterResult(false, value);
} catch (error) {
if (context.cancelled) await closeSource(true);
reading = false;
@@ -421,7 +422,7 @@ function yieldNormalizationAbortable(source, context) {
async return(value) {
await closeSource(
context.suppressCleanup || (context.cancelled && reading));
- return { __proto__: null, done: true, value };
+ return new IterResult(true, value);
},
async throw(error) {
await closeSource(
diff --git a/lib/internal/streams/iter/pull.js b/lib/internal/streams/iter/pull.js
index 70e89a99423..9a52a5fdfed 100644
--- a/lib/internal/streams/iter/pull.js
+++ b/lib/internal/streams/iter/pull.js
@@ -50,6 +50,7 @@ const {
} = require('internal/streams/iter/from');
const {
+ IterResult,
createBatchEntry,
isTransformObject,
parsePullArgs,
@@ -914,7 +915,7 @@ function createAbortablePullIterator(source, transforms, signal) {
},
return(value) {
if (aborted) {
- return PromiseResolve({ __proto__: null, done: true, value });
+ return PromiseResolve(new IterResult(true, value));
}
controller.abort(lazyDOMException('Aborted', 'AbortError'));
return iterator.return(value);
diff --git a/lib/internal/streams/iter/push.js b/lib/internal/streams/iter/push.js
index 039e6b9f39d..d15f5752fa6 100644
--- a/lib/internal/streams/iter/push.js
+++ b/lib/internal/streams/iter/push.js
@@ -29,6 +29,7 @@ const {
} = require('internal/streams/iter/types');
const {
+ IterResult,
kPushDefaultBudget,
kResolvedPromise,
createBatchEntry,
@@ -438,7 +439,7 @@ class PushQueue {
async read() {
if (this.#consumerState === 'returned') {
- return { __proto__: null, done: true, value: undefined };
+ return new IterResult(true, undefined);
}
if (this.#consumerState === 'thrown') {
throw this.#consumerError;
@@ -448,17 +449,17 @@ class PushQueue {
if (this.#slots.length > 0) {
const result = this.#drain();
this.#resolvePendingWrites();
- return { __proto__: null, done: false, value: result };
+ return new IterResult(false, result);
}
// Buffer empty and writer closing = drain complete
if (this.#writerState === 'closing') {
this.endDrained();
- return { __proto__: null, done: true, value: undefined };
+ return new IterResult(true, undefined);
}
if (this.#writerState === 'closed') {
- return { __proto__: null, done: true, value: undefined };
+ return new IterResult(true, undefined);
}
if (this.#writerState === 'errored') {
@@ -540,7 +541,7 @@ class PushQueue {
while (this.#pendingReads.length > 0) {
if (this.#consumerState === 'returned') {
const pending = this.#pendingReads.shift();
- pending.resolve({ __proto__: null, done: true, value: undefined });
+ pending.resolve(new IterResult(true, undefined));
} else if (this.#consumerState === 'thrown') {
const pending = this.#pendingReads.shift();
pending.reject(this.#consumerError);
@@ -549,7 +550,7 @@ class PushQueue {
try {
const result = this.#drain();
this.#resolvePendingWrites();
- pending.resolve({ __proto__: null, done: false, value: result });
+ pending.resolve(new IterResult(false, result));
} catch (error) {
pending.reject(error);
}
@@ -558,10 +559,10 @@ class PushQueue {
this.#pendingWrites.length === 0) {
this.endDrained();
const pending = this.#pendingReads.shift();
- pending.resolve({ __proto__: null, done: true, value: undefined });
+ pending.resolve(new IterResult(true, undefined));
} else if (this.#writerState === 'closed') {
const pending = this.#pendingReads.shift();
- pending.resolve({ __proto__: null, done: true, value: undefined });
+ pending.resolve(new IterResult(true, undefined));
} else if (this.#writerState === 'errored') {
const pending = this.#pendingReads.shift();
pending.reject(this.#writerError);
@@ -740,7 +741,7 @@ function createReadable(queue) {
},
async return() {
queue.consumerReturn();
- return { __proto__: null, done: true, value: undefined };
+ return new IterResult(true, undefined);
},
async throw(error) {
queue.consumerThrow(error);
diff --git a/lib/internal/streams/iter/share.js b/lib/internal/streams/iter/share.js
index a61d4627ad5..937528b3929 100644
--- a/lib/internal/streams/iter/share.js
+++ b/lib/internal/streams/iter/share.js
@@ -38,6 +38,7 @@ const {
} = require('internal/streams/iter/pull');
const {
+ IterResult,
kMultiConsumerDefaultBudget,
createBatchEntry,
getProtocolMethod,
@@ -175,7 +176,7 @@ class ShareImpl {
for (;;) {
if (state.detached) {
if (state.error !== kNoShareError) throw state.error;
- return { __proto__: null, done: true, value: undefined };
+ return new IterResult(true, undefined);
}
if (self.#cancelled) {
@@ -183,7 +184,7 @@ class ShareImpl {
state.error = self.#cancelError;
self.#deleteConsumer(state);
if (state.error !== kNoShareError) throw state.error;
- return { __proto__: null, done: true, value: undefined };
+ return new IterResult(true, undefined);
}
// Check if data is available in buffer
@@ -196,7 +197,7 @@ class ShareImpl {
--self.#cachedMinCursorConsumers === 0) {
self.#tryTrimBuffer();
}
- return { __proto__: null, done: false, value: chunk };
+ return new IterResult(false, chunk);
}
if (self.#sourceExhausted) {
@@ -206,7 +207,7 @@ class ShareImpl {
state.error = self.#sourceError;
throw state.error;
}
- return { __proto__: null, done: true, value: undefined };
+ return new IterResult(true, undefined);
}
// Need to pull from source - check buffer limit
@@ -225,7 +226,7 @@ class ShareImpl {
state.error = self.#cancelError;
self.#deleteConsumer(state);
if (state.error !== kNoShareError) throw state.error;
- return { __proto__: null, done: true, value: undefined };
+ return new IterResult(true, undefined);
}
await self.#pullFromSource(!shouldBuffer);
@@ -253,7 +254,7 @@ class ShareImpl {
if (self.#deleteConsumer(state)) {
self.#tryTrimBuffer();
}
- return { __proto__: null, done: true, value: undefined };
+ return new IterResult(true, undefined);
},
async throw() {
@@ -263,7 +264,7 @@ class ShareImpl {
if (self.#deleteConsumer(state)) {
self.#tryTrimBuffer();
}
- return { __proto__: null, done: true, value: undefined };
+ return new IterResult(true, undefined);
},
}, null);
},
@@ -302,7 +303,7 @@ class ShareImpl {
if (hasReason) {
consumer.reject?.(reason);
} else {
- consumer.resolve({ __proto__: null, done: true, value: undefined });
+ consumer.resolve(new IterResult(true, undefined));
}
consumer.resolve = null;
consumer.reject = null;
@@ -406,7 +407,7 @@ class ShareImpl {
},
async return() {
return syncIterator.return?.() ??
- { __proto__: null, done: true, value: undefined };
+ new IterResult(true, undefined);
},
}, null);
} else {
@@ -594,7 +595,7 @@ class SyncShareImpl {
next() {
if (state.detached) {
if (state.error !== kNoShareError) throw state.error;
- return { __proto__: null, done: true, value: undefined };
+ return new IterResult(true, undefined);
}
if (self.#sourceError !== kNoShareError) {
state.detached = true;
@@ -605,7 +606,7 @@ class SyncShareImpl {
if (self.#cancelled) {
state.detached = true;
self.#deleteConsumer(state);
- return { __proto__: null, done: true, value: undefined };
+ return new IterResult(true, undefined);
}
const bufferIndex = state.cursor - self.#bufferStart;
@@ -617,13 +618,13 @@ class SyncShareImpl {
--self.#cachedMinCursorConsumers === 0) {
self.#tryTrimBuffer();
}
- return { __proto__: null, done: false, value: chunk };
+ return new IterResult(false, chunk);
}
if (self.#sourceExhausted) {
state.detached = true;
self.#deleteConsumer(state);
- return { __proto__: null, done: true, value: undefined };
+ return new IterResult(true, undefined);
}
// Check buffer limit. 'unbounded' and 'drop-newest' are rejected
@@ -680,16 +681,16 @@ class SyncShareImpl {
--self.#cachedMinCursorConsumers === 0) {
self.#tryTrimBuffer();
}
- return { __proto__: null, done: false, value: chunk };
+ return new IterResult(false, chunk);
}
if (self.#sourceExhausted) {
state.detached = true;
self.#deleteConsumer(state);
- return { __proto__: null, done: true, value: undefined };
+ return new IterResult(true, undefined);
}
- return { __proto__: null, done: true, value: undefined };
+ return new IterResult(true, undefined);
},
return() {
@@ -697,7 +698,7 @@ class SyncShareImpl {
if (self.#deleteConsumer(state)) {
self.#tryTrimBuffer();
}
- return { __proto__: null, done: true, value: undefined };
+ return new IterResult(true, undefined);
},
throw() {
@@ -705,7 +706,7 @@ class SyncShareImpl {
if (self.#deleteConsumer(state)) {
self.#tryTrimBuffer();
}
- return { __proto__: null, done: true, value: undefined };
+ return new IterResult(true, undefined);
},
}, null);
},
diff --git a/lib/internal/streams/iter/utils.js b/lib/internal/streams/iter/utils.js
index 2f72a78c0d2..0c6e1daaa32 100644
--- a/lib/internal/streams/iter/utils.js
+++ b/lib/internal/streams/iter/utils.js
@@ -7,6 +7,8 @@ const {
ArrayBufferPrototypeGetResizable,
ArrayPrototypePush,
ArrayPrototypeSlice,
+ ObjectDefineProperty,
+ ObjectFreeze,
PromiseResolve,
PromiseWithResolvers,
SafePromisePrototypeFinally,
@@ -61,6 +63,30 @@ const kPushDefaultBudget = 16384;
/** Default byte budget for broadcast and share streams (multi-consumer). */
const kMultiConsumerDefaultBudget = 65536;
+/**
+ * Iterator result object (`{ done, value }`) for the iterators returned by
+ * this module. A result is created for every chunk, so it must be cheap.
+ *
+ * `new IterResult(done, value)` literals are created in V8 dictionary
+ * mode, which costs several times more than an ordinary object. Instances
+ * of this constructor have fast properties, always with the same shape
+ * (`done` before `value`), and still have no %Object.prototype% in their
+ * prototype chain, so a polluted `Object.prototype.then` cannot turn a
+ * result into a thenable when an async `next()` resolves with it. The
+ * prototype is a single empty, frozen, null-prototype object (V8 gives
+ * objects fast properties once they are used as a prototype).
+ * @param {boolean} done
+ * @param {any} value
+ */
+function IterResult(done, value) {
+ this.done = done;
+ this.value = value;
+}
+// The non-enumerable `constructor` lets util.inspect() print results as
+// `IterResult { done, value }`.
+IterResult.prototype = ObjectFreeze(ObjectDefineProperty(
+ { __proto__: null }, 'constructor', { __proto__: null, value: IterResult }));
+
/**
* Register a handler for an AbortSignal, handling the already-aborted case.
* If the signal is already aborted, calls handler immediately.
@@ -539,6 +565,7 @@ function validateBackpressure(value) {
}
module.exports = {
+ IterResult,
kMultiConsumerDefaultBudget,
kPushDefaultBudget,
kResolvedPromise,
diff --git a/test/parallel/test-stream-iter-broadcast-basic.js b/test/parallel/test-stream-iter-broadcast-basic.js
index 232439cdf7d..c59689c2d12 100644
--- a/test/parallel/test-stream-iter-broadcast-basic.js
+++ b/test/parallel/test-stream-iter-broadcast-basic.js
@@ -385,8 +385,7 @@ async function testOverlappingNextKeepsEarlierRead() {
assert.strictEqual(Buffer.concat(result.value).toString(), 'x');
writer.endSync();
- assert.deepStrictEqual(await second, {
- __proto__: null,
+ assert.deepStrictEqual({ ...await second }, {
done: true,
value: undefined,
});
diff --git a/test/parallel/test-stream-iter-broadcast-from.js b/test/parallel/test-stream-iter-broadcast-from.js
index b76a2b31b7f..8bc2abde2e1 100644
--- a/test/parallel/test-stream-iter-broadcast-from.js
+++ b/test/parallel/test-stream-iter-broadcast-from.js
@@ -144,8 +144,7 @@ async function testBroadcastFromCancelWhileBlocked() {
let writesAfterCancel = 0;
writer.writevSync = () => { writesAfterCancel++; return true; };
bc.cancel();
- assert.deepStrictEqual(await pendingRead, {
- __proto__: null,
+ assert.deepStrictEqual({ ...await pendingRead }, {
done: true,
value: undefined,
});
diff --git a/test/parallel/test-stream-iter-iterator-result.js b/test/parallel/test-stream-iter-iterator-result.js
new file mode 100644
index 00000000000..4f29bf05534
--- /dev/null
+++ b/test/parallel/test-stream-iter-iterator-result.js
@@ -0,0 +1,79 @@
+// Flags: --experimental-stream-iter
+'use strict';
+
+// Iterator results created by the stream/iter iterators themselves do not
+// inherit from Object.prototype, so prototype pollution cannot affect them.
+// (Iterators implemented as async generators, such as the one returned by
+// pull(), return ordinary iterator results created by the engine.)
+
+const common = require('../common');
+const assert = require('assert');
+const { inspect } = require('util');
+const {
+ broadcast,
+ from,
+ push,
+ share,
+ shareSync,
+} = require('stream/iter');
+
+function assertResult(result, done) {
+ assert.strictEqual(result instanceof Object, false);
+ assert.deepStrictEqual(Object.keys(result), ['done', 'value']);
+ assert.strictEqual(result.done, done);
+ assert.match(inspect(result), /^IterResult \{ done: (true|false), value: /);
+}
+
+async function testAsyncIterators() {
+ const sources = {
+ 'push()': () => {
+ const { writer, readable } = push();
+ writer.writeSync('a');
+ writer.endSync();
+ return readable;
+ },
+ 'share()': () => share(from('a')).pull(),
+ 'broadcast()': () => {
+ const { writer, broadcast: bc } = broadcast();
+ const consumer = bc.push();
+ writer.writeSync('a');
+ writer.endSync();
+ return consumer;
+ },
+ };
+ for (const create of Object.values(sources)) {
+ const iterator = create()[Symbol.asyncIterator]();
+ assertResult(await iterator.next(), false);
+ assertResult(await iterator.next(), true);
+ }
+}
+
+function testShareSync() {
+ const iterator = shareSync(['a']).pull()[Symbol.iterator]();
+ assertResult(iterator.next(), false);
+ assertResult(iterator.next(), true);
+}
+
+async function testPollutedThen() {
+ // Resolving an async next() with an object looks up `then`. Results must
+ // not pick it up from a polluted Object.prototype.
+ const { writer, readable } = push();
+ writer.writeSync('ab');
+ writer.endSync();
+ const iterator = readable[Symbol.asyncIterator]();
+ Object.prototype.then = common.mustNotCall('Object.prototype.then');
+ try {
+ const first = await iterator.next();
+ assert.strictEqual(first.done, false);
+ assert.strictEqual(new TextDecoder().decode(first.value[0]), 'ab');
+ assert.strictEqual((await iterator.next()).done, true);
+ } finally {
+ delete Object.prototype.then;
+ }
+}
+
+(async () => {
+ await testAsyncIterators();
+ testShareSync();
+ await testPollutedThen();
+})().then(common.mustCall());
diff --git a/test/parallel/test-stream-iter-pull-async.js b/test/parallel/test-stream-iter-pull-async.js
index 632306fa589..71fc27426ea 100644
--- a/test/parallel/test-stream-iter-pull-async.js
+++ b/test/parallel/test-stream-iter-pull-async.js
@@ -79,8 +79,8 @@ async function testPullWithAbortSignal() {
await assert.rejects(iterator.next(), (error) => error === signal.reason);
await assert.rejects(iterator.next(), (error) => error === signal.reason);
assert.strictEqual(started, false);
- assert.deepStrictEqual(await iterator.return(),
- { __proto__: null, done: true, value: undefined });
+ assert.deepStrictEqual({ ...await iterator.return() },
+ { done: true, value: undefined });
await assert.rejects(text(pull(gen(), (chunks) => chunks, { signal })),
(error) => error === signal.reason);
diff --git a/test/parallel/test-stream-iter-reason-propagation.js b/test/parallel/test-stream-iter-reason-propagation.js
index 7511dbb2c41..58ae7d4e5f4 100644
--- a/test/parallel/test-stream-iter-reason-propagation.js
+++ b/test/parallel/test-stream-iter-reason-propagation.js
@@ -249,8 +249,7 @@ async function testCompletedBroadcastConsumerStaysCompleted() {
await iterator.return();
writer.fail(undefined);
- assert.deepStrictEqual(await iterator.next(), {
- __proto__: null,
+ assert.deepStrictEqual({ ...await iterator.next() }, {
done: true,
value: undefined,
});
diff --git a/test/parallel/test-stream-iter-share-coverage.js b/test/parallel/test-stream-iter-share-coverage.js
index fee28ae1568..73827f2d40f 100644
--- a/test/parallel/test-stream-iter-share-coverage.js
+++ b/test/parallel/test-stream-iter-share-coverage.js
@@ -113,8 +113,7 @@ async function testCompletedSyncConsumerStaysCompleted() {
assert.strictEqual(error, reason);
}
assert.strictEqual(caught, true);
- assert.deepStrictEqual(completed.next(), {
- __proto__: null,
+ assert.deepStrictEqual({ ...completed.next() }, {
done: true,
value: undefined,
});