From 85bd8607038dc5bcbe444fd005cdecf4320bf1f0 Mon Sep 17 00:00:00 2001 From: James M Snell Date: Sun, 30 Aug 2026 17:13:53 +0000 Subject: [PATCH] stream: ensure streamable protocols precede fast paths Signed-off-by: James M Snell Assisted-by: Opencode PR-URL: https://github.com/nodejs/node/pull/66030 Reviewed-By: Filip Skokan Reviewed-By: Trivikram Kamat --- lib/internal/streams/iter/from.js | 70 +++++++++---------- .../test-stream-iter-property-access.js | 40 ++++++++++- 2 files changed, 73 insertions(+), 37 deletions(-) diff --git a/lib/internal/streams/iter/from.js b/lib/internal/streams/iter/from.js index 0d95b813e01..cac8fd3565b 100644 --- a/lib/internal/streams/iter/from.js +++ b/lib/internal/streams/iter/from.js @@ -661,6 +661,13 @@ function fromSync(input) { }; } + // Check toStreamable protocol (takes precedence over iteration protocols). + // toAsyncStreamable is ignored entirely in fromSync. + const streamableMethod = getProtocolMethod(input, toStreamable); + if (streamableMethod !== undefined) { + return fromSync(FunctionPrototypeCall(streamableMethod, input)); + } + // Fast path: Uint8Array[] - yield in bounded sub-batches. // Yielding the entire array as one batch forces downstream transforms // to process all data at once, causing peak memory proportional to total @@ -696,13 +703,6 @@ function fromSync(input) { } } - // Check toStreamable protocol (takes precedence over iteration protocols). - // toAsyncStreamable is ignored entirely in fromSync. - const streamableMethod = getProtocolMethod(input, toStreamable); - if (streamableMethod !== undefined) { - return fromSync(FunctionPrototypeCall(streamableMethod, input)); - } - const isIterable = isSyncIterable(input); // Reject explicit async-only inputs @@ -751,11 +751,6 @@ function from(input) { throw new ERR_INVALID_ARG_TYPE('input', 'a non-null value', input); } - // Fast path: validated source already yields valid Uint8Array[] batches - if (input[kValidatedSource]) { - return input; - } - // Check for primitives first (ByteInput) if (isPrimitiveChunk(input)) { const chunk = primitiveToUint8Array(input); @@ -767,6 +762,34 @@ function from(input) { }; } + // Check toAsyncStreamable protocol (takes precedence over toStreamable and + // iteration protocols) + const asyncStreamableMethod = getProtocolMethod(input, toAsyncStreamable); + if (asyncStreamableMethod !== undefined) { + let result = FunctionPrototypeCall(asyncStreamableMethod, input); + if (isPromise(result)) { + result = PromisePrototypeThen(result, undefined, undefined); + markPromiseAsHandled(result); + } + // Synchronous validated source (e.g. Readable batched iterator) + if (result?.[kValidatedSource]) { + return result; + } + return createNormalizationSource( + (context) => normalizeAsyncStreamableResult(result, context)); + } + + // Check toStreamable protocol (takes precedence over iteration protocols) + const streamableMethod = getProtocolMethod(input, toStreamable); + if (streamableMethod !== undefined) { + return from(FunctionPrototypeCall(streamableMethod, input)); + } + + // Fast path: validated source already yields valid Uint8Array[] batches + if (input[kValidatedSource]) { + return input; + } + // Fast path: Uint8Array[] - yield in bounded sub-batches. // Yielding the entire array as one batch forces downstream transforms // to process all data at once, causing peak memory proportional to total @@ -801,29 +824,6 @@ function from(input) { } } - // Check toAsyncStreamable protocol (takes precedence over toStreamable and - // iteration protocols) - const asyncStreamableMethod = getProtocolMethod(input, toAsyncStreamable); - if (asyncStreamableMethod !== undefined) { - let result = FunctionPrototypeCall(asyncStreamableMethod, input); - if (isPromise(result)) { - result = PromisePrototypeThen(result, undefined, undefined); - markPromiseAsHandled(result); - } - // Synchronous validated source (e.g. Readable batched iterator) - if (result?.[kValidatedSource]) { - return result; - } - return createNormalizationSource( - (context) => normalizeAsyncStreamableResult(result, context)); - } - - // Check toStreamable protocol (takes precedence over iteration protocols) - const streamableMethod = getProtocolMethod(input, toStreamable); - if (streamableMethod !== undefined) { - return from(FunctionPrototypeCall(streamableMethod, input)); - } - // Must be a Streamable (sync or async iterable) if (!isSyncIterable(input) && !isAsyncIterable(input)) { throw new ERR_INVALID_ARG_TYPE( diff --git a/test/parallel/test-stream-iter-property-access.js b/test/parallel/test-stream-iter-property-access.js index 344a481b081..5ba1a5a587f 100644 --- a/test/parallel/test-stream-iter-property-access.js +++ b/test/parallel/test-stream-iter-property-access.js @@ -3,6 +3,7 @@ const common = require('../common'); const assert = require('assert'); +const { Readable } = require('stream'); const { Broadcast, Share, @@ -21,13 +22,14 @@ const { shareProtocol, shareSync, shareSyncProtocol, + text, + textSync, toAsyncStreamable, toStreamable, } = require('stream/iter'); -function protocolFixture(symbol, result) { +function protocolFixture(symbol, result, input = {}) { let accesses = 0; - const input = {}; const method = common.mustCall(function() { assert.strictEqual(this, input); return result; @@ -83,6 +85,38 @@ async function testFromSyncSnapshotsProtocolMethods() { assert.strictEqual(fixture.accesses, 1); } +async function testArrayFastPathsHonorProtocols() { + const asyncInputs = [ + protocolFixture(toAsyncStreamable, 'empty-async', []), + protocolFixture(toStreamable, 'batch-async', [new Uint8Array([0])]), + ]; + for (const fixture of asyncInputs) { + assert.match(await text(from(fixture.input)), /-async$/); + assert.strictEqual(fixture.accesses, 1); + } + + const syncInputs = [ + protocolFixture(toStreamable, 'empty-sync', []), + protocolFixture(toStreamable, 'batch-sync', [new Uint8Array([0])]), + ]; + for (const fixture of syncInputs) { + assert.match(textSync(fromSync(fixture.input)), /-sync$/); + assert.strictEqual(fixture.accesses, 1); + } +} + +async function testValidatedSourceHonorsProtocol() { + const readable = Readable.from(['ignored']); + const validated = readable[toAsyncStreamable](); + assert.strictEqual(from(validated), validated); + + const fixture = protocolFixture( + toAsyncStreamable, 'validated-protocol', validated); + assert.strictEqual(await text(from(fixture.input)), 'validated-protocol'); + assert.strictEqual(fixture.accesses, 1); + readable.destroy(); +} + async function testPullSnapshotsStatefulTransform() { const fixture = statefulTransformFixture(); const controller = new AbortController(); @@ -136,6 +170,8 @@ async function testDrainableProtocolSnapshotMethod() { Promise.all([ testFromSnapshotsProtocolMethods(), testFromSyncSnapshotsProtocolMethods(), + testArrayFastPathsHonorProtocols(), + testValidatedSourceHonorsProtocol(), testPullSnapshotsStatefulTransform(), testPullSyncSnapshotsStatefulTransform(), testMultiConsumerProtocolsSnapshotMethods(),