diff --git a/lib/internal/streams/iter/pull.js b/lib/internal/streams/iter/pull.js index a9a77377a84..c1e42f03dc3 100644 --- a/lib/internal/streams/iter/pull.js +++ b/lib/internal/streams/iter/pull.js @@ -24,6 +24,7 @@ const { codes: { ERR_INVALID_ARG_TYPE, ERR_INVALID_ARG_VALUE, + ERR_INVALID_STATE, ERR_OUT_OF_RANGE, }, } = require('internal/errors'); @@ -981,6 +982,13 @@ function pullWithConsumerCleanup(source, transforms, signal) { */ function pipeToSync(source, ...args) { const { transforms, writer, options } = parsePipeToArgs(args, 'writeSync'); + const hasWritevSync = typeof writer.writevSync === 'function'; + const endSync = writer.endSync; + + if (!options?.preventClose && typeof endSync !== 'function') { + throw new ERR_INVALID_ARG_TYPE( + 'writer.endSync', 'Function', endSync); + } // Normalize source and create pipeline const normalized = fromSync(source); @@ -989,8 +997,6 @@ function pipeToSync(source, ...args) { normalized; let totalBytes = 0; - const hasWritevSync = typeof writer.writevSync === 'function'; - const hasEndSync = typeof writer.endSync === 'function'; try { for (const batch of pipeline) { @@ -1019,8 +1025,9 @@ function pipeToSync(source, ...args) { } if (!options?.preventClose) { - if (!hasEndSync || writer.endSync() < 0) { - writer.end?.(); + if (FunctionPrototypeCall(endSync, writer) < 0) { + throw new ERR_INVALID_STATE( + 'Writer could not be closed synchronously'); } } } catch (error) { diff --git a/test/parallel/test-stream-iter-pipeto-edge.js b/test/parallel/test-stream-iter-pipeto-edge.js index 3f09c4dfd42..13223a226a1 100644 --- a/test/parallel/test-stream-iter-pipeto-edge.js +++ b/test/parallel/test-stream-iter-pipeto-edge.js @@ -1,33 +1,41 @@ // Flags: --experimental-stream-iter 'use strict'; -// Edge case tests for pipeToSync: endSync fallback, preventFail. +// Edge case tests for pipeToSync close and failure behavior. const common = require('../common'); const assert = require('assert'); const { pipeToSync, fromSync } = require('stream/iter'); -// pipeToSync endSync returns negative → falls back to end() -async function testPipeToSyncEndSyncFallback() { +// pipeToSync cannot complete when endSync() requires async fallback. +async function testPipeToSyncEndSyncFailure() { let endCalled = false; const writer = { writeSync() { return true; }, - endSync() { return -1; }, // Negative → triggers end() fallback + endSync() { return -1; }, end() { endCalled = true; }, }; - pipeToSync(fromSync('data'), writer); - assert.strictEqual(endCalled, true); + assert.throws( + () => pipeToSync(fromSync('data'), writer, { preventFail: true }), + { code: 'ERR_INVALID_STATE' }, + ); + assert.strictEqual(endCalled, false); } -// pipeToSync endSync missing → falls back to end() +// pipeToSync requires endSync() when closing is enabled. async function testPipeToSyncNoEndSync() { + let writeCalled = false; let endCalled = false; const writer = { - writeSync() { return true; }, + writeSync() { writeCalled = true; return true; }, end() { endCalled = true; }, }; - pipeToSync(fromSync('data'), writer); - assert.strictEqual(endCalled, true); + assert.throws( + () => pipeToSync(fromSync('data'), writer), + { code: 'ERR_INVALID_ARG_TYPE' }, + ); + assert.strictEqual(writeCalled, false); + assert.strictEqual(endCalled, false); } // pipeToSync with preventFail: true — source error does NOT call fail() @@ -61,7 +69,7 @@ async function testPipeToSyncPreventClose() { } Promise.all([ - testPipeToSyncEndSyncFallback(), + testPipeToSyncEndSyncFailure(), testPipeToSyncNoEndSync(), testPipeToSyncPreventFail(), testPipeToSyncPreventClose(), diff --git a/test/parallel/test-stream-iter-pipeto.js b/test/parallel/test-stream-iter-pipeto.js index 5d8b5088f54..f16d7ae8997 100644 --- a/test/parallel/test-stream-iter-pipeto.js +++ b/test/parallel/test-stream-iter-pipeto.js @@ -71,6 +71,7 @@ async function testPipeToSyncSourceError() { let failCalled = false; const writer = { writeSync() { return true; }, + endSync: common.mustNotCall(), fail(reason) { failCalled = true; }, }; function* failingSource() { @@ -127,6 +128,7 @@ async function testPipeToSyncWithTransforms() { const chunks = []; const writer = { writeSync(chunk) { chunks.push(new TextDecoder().decode(chunk)); return true; }, + endSync() { return 0; }, }; const upper = (batch) => { if (batch === null) return null; @@ -160,6 +162,7 @@ async function testPipeToSyncWriterTransformMethodIgnored() { chunks.push(new TextDecoder().decode(chunk)); return true; }, + endSync() { return 0; }, }; pipeToSync(fromSync('hello'), writer); @@ -240,7 +243,7 @@ async function testPipeToSyncMinimalWriter() { }, }; - pipeToSync(fromSync('minimal-sync'), minimalWriter); + pipeToSync(fromSync('minimal-sync'), minimalWriter, { preventClose: true }); assert.strictEqual(chunks.length > 0, true); } diff --git a/test/parallel/test-stream-iter-resizable-buffers.js b/test/parallel/test-stream-iter-resizable-buffers.js index c4be26774ac..b66973393ca 100644 --- a/test/parallel/test-stream-iter-resizable-buffers.js +++ b/test/parallel/test-stream-iter-resizable-buffers.js @@ -161,7 +161,8 @@ async function testPipeRejectsWriterResize() { fail: common.mustCall(), }; assert.throws( - () => pipeToSync([new Uint8Array(syncBuffer)], syncWriter), + () => pipeToSync( + [new Uint8Array(syncBuffer)], syncWriter, { preventClose: true }), kResizeError, ); }