From b329f009c31807d93bd8a153b3512c8b6fe3ae3a Mon Sep 17 00:00:00 2001 From: Yagiz Nizipli Date: Thu, 20 Aug 2026 16:02:58 +0000 Subject: [PATCH 1/3] stream: process CompressionStream chunks without threadpool round trips Rewrite CompressionStream and DecompressionStream on top of a TransformStream that drives the zlib (or brotli) handle synchronously, instead of wrapping a zlib stream.Duplex in the web streams adapters. The previous implementation dispatched every written chunk to the threadpool and waited for the event loop to observe its completion, which dominates the cost of streaming small chunks: a 64 MiB body written in 4 KiB chunks paid for 16384 threadpool round trips plus the Transform and adapter machinery around them. Processing the chunks inline removes that latency entirely while performing the same work. Inputs larger than 64 KiB are processed in slices with a turn of the event loop in between so that huge chunks cannot block the loop for their full duration. Output is emitted in up to 64 KiB chunks, either as zero-copy views or as right-sized copies (so small outputs do not retain large buffers), which also reduces the per-chunk overhead imposed on the rest of the pipeline downstream. Assisted-by: Cursor Signed-off-by: Yagiz Nizipli --- benchmark/webstreams/compression.js | 67 ++++ lib/internal/webstreams/compression.js | 405 ++++++++++++++++++++++--- 2 files changed, 431 insertions(+), 41 deletions(-) create mode 100644 benchmark/webstreams/compression.js diff --git a/benchmark/webstreams/compression.js b/benchmark/webstreams/compression.js new file mode 100644 index 000000000000..8bc0fdc6225c --- /dev/null +++ b/benchmark/webstreams/compression.js @@ -0,0 +1,67 @@ +'use strict'; +const common = require('../common.js'); +const { gzipSync, deflateSync } = require('node:zlib'); +const assert = require('node:assert'); + +const bench = common.createBenchmark(main, { + chunkSize: [4096, 65536], + totalBytes: [64 << 20], + kind: ['compress', 'decompress'], + format: ['gzip', 'deflate'], +}); + +// Repetitive but non-trivial payload. +function makePayload(totalBytes, chunkSize) { + const line = 'time=2024-01-01T00:00:00.000Z level=info request=42 ' + + 'method=GET path=/api/v1/items status=200 duration=13ms\n'; + const chunk = Buffer.alloc(chunkSize, line); + const chunks = []; + for (let n = 0; n < totalBytes; n += chunk.length) + chunks.push(chunk); + return chunks; +} + +function makeSource(chunks) { + let i = 0; + return new ReadableStream({ + pull(controller) { + if (i < chunks.length) { + controller.enqueue(chunks[i++]); + } else { + controller.close(); + } + }, + }); +} + +async function main({ chunkSize, totalBytes, kind, format }) { + let chunks = makePayload(totalBytes, chunkSize); + + let stream; + if (kind === 'compress') { + stream = new CompressionStream(format); + } else { + // Rechunk the compressed payload so the decompressor sees the + // configured chunk size. + const compress = format === 'gzip' ? gzipSync : deflateSync; + const compressed = compress(Buffer.concat(chunks)); + chunks = []; + for (let off = 0; off < compressed.length; off += chunkSize) + chunks.push(compressed.subarray(off, off + chunkSize)); + stream = new DecompressionStream(format); + } + + const source = makeSource(chunks); + + let outputBytes = 0; + bench.start(); + for await (const chunk of source.pipeThrough(stream)) { + outputBytes += chunk.byteLength; + } + bench.end(totalBytes / (1024 * 1024)); + + if (kind === 'decompress') + assert.strictEqual(outputBytes, totalBytes); + else + assert.notStrictEqual(outputBytes, 0); +} diff --git a/lib/internal/webstreams/compression.js b/lib/internal/webstreams/compression.js index 70359d1158ee..ee00a64a501a 100644 --- a/lib/internal/webstreams/compression.js +++ b/lib/internal/webstreams/compression.js @@ -1,52 +1,163 @@ 'use strict'; const { + ArrayPrototypeFilter, + MathMin, ObjectDefineProperties, + ObjectKeys, + Promise, + SafeSet, + StringPrototypeStartsWith, SymbolToStringTag, + TypeError, + TypedArrayPrototypeGetByteLength, + TypedArrayPrototypeSet, + TypedArrayPrototypeSubarray, + Uint32Array, + Uint8Array, } = primordials; const { - newReadableWritablePairFromDuplex, - kValidateChunk, - kDestroyOnSyncError, -} = require('internal/webstreams/adapters'); + TransformStream, +} = require('internal/webstreams/transformstream'); const { customInspect } = require('internal/webstreams/util'); const { isArrayBufferView, + isAnyArrayBuffer, isSharedArrayBuffer, + isUint8Array, } = require('internal/util/types'); const { customInspectSymbol: kInspect, kEnumerableProperty, + setOwnProperty, } = require('internal/util'); const { codes: { ERR_INVALID_ARG_TYPE, + ERR_STREAM_NULL_VALUES, + ERR_TRAILING_JUNK_AFTER_STREAM_END, }, + genericNodeError, } = require('internal/errors'); const { createEnumConverter } = require('internal/webidl'); -let zlib; -function lazyZlib() { - zlib ??= require('zlib'); - return zlib; +const { + Zlib, + BrotliDecoder, + BrotliEncoder, +} = internalBinding('zlib'); + +const { zlib: constants } = internalBinding('constants'); +const { + BROTLI_DECODE, + BROTLI_ENCODE, + BROTLI_OPERATION_FINISH, + BROTLI_OPERATION_PROCESS, + DEFLATE, + DEFLATERAW, + GUNZIP, + GZIP, + INFLATE, + INFLATERAW, + Z_DEFAULT_COMPRESSION, + Z_DEFAULT_MEMLEVEL, + Z_DEFAULT_STRATEGY, + Z_DEFAULT_WINDOWBITS, + Z_FINISH, + Z_NO_FLUSH, +} = constants; + +const { Buffer } = require('buffer'); + +let setImmediate; + +// Output is accumulated in fixed-size buffers and emitted as soon as it is +// produced, mirroring how the zlib streams emit their output. Produced +// regions at least half a buffer large are emitted as zero-copy views (and +// the buffer is retired so no two emitted chunks ever share memory); +// smaller regions are copied out so that tiny chunks do not retain large +// allocations. +const kOutputBufferSize = 65536; +const kEmitViewThreshold = kOutputBufferSize / 2; + +// Inputs larger than this are processed in slices, with a turn of the event +// loop in between, so that compressing/decompressing a huge chunk does not +// block the event loop for its full duration. +const kInputSliceSize = 65536; + +// Collect all negative (error) ZLIB codes and Z_NEED_DICT. +const ZLIB_FAILURES = new SafeSet( + ArrayPrototypeFilter( + ObjectKeys(constants), + (code) => code === 'Z_NEED_DICT' || constants[code] < 0, + ), +); + +// Compression error codes are surfaced as TypeError to align with the +// WHATWG Compression Streams specification. +function convertToTypeError(message, errno, code) { + const cause = genericNodeError(message, { errno, code }); + cause.errno = errno; + cause.code = code; + if (ZLIB_FAILURES.has(code) || + // Brotli decoder error codes are formatted as 'ERR_' + + // BrotliDecoderErrorString(), where the latter returns strings like + // '_ERROR_FORMAT_...', '_ERROR_ALLOC_...', '_ERROR_UNREACHABLE', etc. + // The resulting JS error codes all start with 'ERR__ERROR_'. + StringPrototypeStartsWith(code, 'ERR__ERROR_')) { + // eslint-disable-next-line no-restricted-syntax + const error = new TypeError(undefined, { cause }); + setOwnProperty(error, 'code', code); + return error; + } + return cause; } // Per the Compression Streams spec, chunks must be BufferSource // (ArrayBuffer or ArrayBufferView not backed by SharedArrayBuffer). -function validateBufferSourceChunk(chunk) { - if (isSharedArrayBuffer(isArrayBufferView(chunk) ? chunk.buffer : chunk)) { - throw new ERR_INVALID_ARG_TYPE( - 'chunk', - ['ArrayBuffer', 'Buffer', 'TypedArray', 'DataView'], - chunk, - ); +// Additionally, strings are accepted for backwards compatibility with the +// previous Node.js streams-based implementation. +function normalizeChunk(chunk) { + if (chunk === null) { + throw new ERR_STREAM_NULL_VALUES(); + } + if (typeof chunk === 'string') { + return Buffer.from(chunk); + } + if (isArrayBufferView(chunk)) { + if (isSharedArrayBuffer(chunk.buffer)) { + throw new ERR_INVALID_ARG_TYPE( + 'chunk', + ['ArrayBuffer', 'Buffer', 'TypedArray', 'DataView'], + chunk, + ); + } + if (isUint8Array(chunk)) { + return chunk; + } + return new Uint8Array(chunk.buffer, chunk.byteOffset, chunk.byteLength); + } + if (isAnyArrayBuffer(chunk)) { + if (isSharedArrayBuffer(chunk)) { + throw new ERR_INVALID_ARG_TYPE( + 'chunk', + ['ArrayBuffer', 'Buffer', 'TypedArray', 'DataView'], + chunk, + ); + } + return new Uint8Array(chunk); } + throw new ERR_INVALID_ARG_TYPE( + 'chunk', + ['string', 'Buffer', 'TypedArray', 'DataView'], + chunk, + ); } const formatConverter = createEnumConverter('CompressionFormat', [ @@ -56,13 +167,242 @@ const formatConverter = createEnumConverter('CompressionFormat', [ 'brotli', ]); +// Processes chunks synchronously on the current thread using the raw zlib +// (or brotli) handle, avoiding both the threadpool round trip that the +// zlib streams make for every write and the stream.Duplex adapter layers. +// For the small chunks that flow through streams pipelines, running the +// compression inline is significantly cheaper than dispatching each chunk +// to the threadpool and waiting for the event loop to observe completion. +class CompressionHandle { + #handle; + #writeState = new Uint32Array(2); + #error; + #closed = false; + #outBuffer = null; + #outOffset = 0; + #noFlushFlag; + #finishFlag; + #rejectTrailingInput; + + constructor(mode) { + const onerror = (message, errno, code) => { + this.#error = convertToTypeError(message, errno, code); + }; + if (mode === BROTLI_ENCODE || mode === BROTLI_DECODE) { + this.#handle = mode === BROTLI_DECODE ? + new BrotliDecoder(mode) : new BrotliEncoder(mode); + this.#handle.init( + getBrotliDefaultParams(), + this.#writeState, + noopOnWriteComplete, + undefined, + ); + this.#noFlushFlag = BROTLI_OPERATION_PROCESS; + this.#finishFlag = BROTLI_OPERATION_FINISH; + this.#rejectTrailingInput = mode === BROTLI_DECODE; + } else { + const decompress = + mode === INFLATE || mode === INFLATERAW || mode === GUNZIP; + // A windowBits value of 0 tells zlib to use the window size stored + // in the header of the compressed stream. + const windowBits = mode === INFLATE || mode === GUNZIP ? + 0 : Z_DEFAULT_WINDOWBITS; + this.#handle = new Zlib(mode); + this.#handle.init( + windowBits, + Z_DEFAULT_COMPRESSION, + Z_DEFAULT_MEMLEVEL, + Z_DEFAULT_STRATEGY, + this.#writeState, + noopOnWriteComplete, + undefined, + decompress, + ); + this.#noFlushFlag = Z_NO_FLUSH; + this.#finishFlag = Z_FINISH; + this.#rejectTrailingInput = decompress; + } + this.#handle.onerror = onerror; + } + + transform(chunk, controller) { + chunk = normalizeChunk(chunk); + if (TypedArrayPrototypeGetByteLength(chunk) <= kInputSliceSize) { + this.#process(chunk, this.#noFlushFlag, controller); + return; + } + return this.#processSlices(chunk, controller); + } + + async #processSlices(chunk, controller) { + let offset = 0; + while (offset < TypedArrayPrototypeGetByteLength(chunk)) { + if (this.#closed) return; + const end = MathMin( + offset + kInputSliceSize, TypedArrayPrototypeGetByteLength(chunk)); + this.#process( + TypedArrayPrototypeSubarray(chunk, offset, end), + this.#noFlushFlag, + controller); + offset = end; + if (offset < TypedArrayPrototypeGetByteLength(chunk)) { + setImmediate ??= require('timers').setImmediate; + await new Promise(setImmediate); + } + } + } + + flush(controller) { + try { + this.#process(kEmptyInput, this.#finishFlag, controller); + } finally { + this.close(); + } + } + + close() { + if (!this.#closed) { + this.#closed = true; + this.#handle.close(); + } + } + + #process(chunk, flushFlag, controller) { + let availIn = TypedArrayPrototypeGetByteLength(chunk); + let inOff = 0; + const writeState = this.#writeState; + const handle = this.#handle; + + let availOutAfter; + let availInAfter; + do { + if (this.#outBuffer === null) { + this.#outBuffer = new Uint8Array(kOutputBufferSize); + this.#outOffset = 0; + } + const availOutBefore = kOutputBufferSize - this.#outOffset; + handle.writeSync(flushFlag, + chunk, // in + inOff, // in_off + availIn, // in_len + this.#outBuffer, // out + this.#outOffset, // out_off + availOutBefore); // out_len + if (this.#error !== undefined) { + const error = this.#error; + this.close(); + throw error; + } + + availOutAfter = writeState[0]; + availInAfter = writeState[1]; + + const have = availOutBefore - availOutAfter; + if (have > 0) { + this.#emit(have, controller); + } + + // Exhausted the output buffer: emit and reprocess the rest of the + // input against a fresh buffer. + inOff += availIn - availInAfter; + availIn = availInAfter; + } while (availOutAfter === 0); + + if (availInAfter > 0 && this.#rejectTrailingInput) { + // The compression library was not interested in receiving more data: + // the compressed stream has ended, with junk data trailing behind it. + const error = new ERR_TRAILING_JUNK_AFTER_STREAM_END(); + this.close(); + throw error; + } + } + + #emit(have, controller) { + const offset = this.#outOffset; + const buffer = this.#outBuffer; + let chunk; + if (have >= kEmitViewThreshold) { + // Emit a zero-copy view and retire the buffer so that no two emitted + // chunks ever share the same backing memory. + chunk = new Uint8Array(buffer.buffer, offset, have); + this.#outBuffer = null; + } else { + chunk = new Uint8Array(have); + TypedArrayPrototypeSet( + chunk, TypedArrayPrototypeSubarray(buffer, offset, offset + have)); + this.#outOffset = offset + have; + if (this.#outOffset === kOutputBufferSize) { + this.#outBuffer = null; + } + } + controller.enqueue(chunk); + } +} + +const kEmptyInput = new Uint8Array(0); + +// The write callback is required by the handle's init function, but it is +// only ever invoked by asynchronous writes, which are never issued here. +function noopOnWriteComplete() {} + +let brotliDefaultParams; +function getBrotliDefaultParams() { + if (brotliDefaultParams === undefined) { + let maxParam = 0; + for (const key of ObjectKeys(constants)) { + if (StringPrototypeStartsWith(key, 'BROTLI_PARAM_') && + constants[key] > maxParam) { + maxParam = constants[key]; + } + } + // -1 (as an unsigned 32-bit value) marks a parameter as unset. + brotliDefaultParams = new Uint32Array(maxParam + 1); + brotliDefaultParams.fill(-1); + } + return brotliDefaultParams; +} + +// These match the strategies that the previous stream.Duplex-based +// implementation derived from the zlib streams' high water marks. +function getWritableStrategy() { + return { + highWaterMark: 16384, + size(chunk) { + return chunk?.byteLength ?? chunk?.length ?? 1; + }, + }; +} + +function getReadableStrategy() { + return { + highWaterMark: 16384, + size(chunk) { + return chunk.byteLength; + }, + }; +} + +function createTransform(mode) { + const handle = new CompressionHandle(mode); + return new TransformStream({ + transform(chunk, controller) { + return handle.transform(chunk, controller); + }, + flush(controller) { + handle.flush(controller); + }, + cancel() { + handle.close(); + }, + }, getWritableStrategy(), getReadableStrategy()); +} + /** * @typedef {import('./readablestream').ReadableStream} ReadableStream * @typedef {import('./writablestream').WritableStream} WritableStream */ class CompressionStream { - #handle; #transform; /** @@ -75,22 +415,18 @@ class CompressionStream { }); switch (format) { case 'deflate': - this.#handle = lazyZlib().createDeflate(); + this.#transform = createTransform(DEFLATE); break; case 'deflate-raw': - this.#handle = lazyZlib().createDeflateRaw(); + this.#transform = createTransform(DEFLATERAW); break; case 'gzip': - this.#handle = lazyZlib().createGzip(); + this.#transform = createTransform(GZIP); break; case 'brotli': - this.#handle = lazyZlib().createBrotliCompress(); + this.#transform = createTransform(BROTLI_ENCODE); break; } - this.#transform = newReadableWritablePairFromDuplex(this.#handle, { - [kValidateChunk]: validateBufferSourceChunk, - [kDestroyOnSyncError]: true, - }); } /** @@ -118,7 +454,6 @@ class CompressionStream { } class DecompressionStream { - #handle; #transform; /** @@ -131,30 +466,18 @@ class DecompressionStream { }); switch (format) { case 'deflate': - this.#handle = lazyZlib().createInflate({ - rejectGarbageAfterEnd: true, - }); + this.#transform = createTransform(INFLATE); break; case 'deflate-raw': - this.#handle = lazyZlib().createInflateRaw({ - rejectGarbageAfterEnd: true, - }); + this.#transform = createTransform(INFLATERAW); break; case 'gzip': - this.#handle = lazyZlib().createGunzip({ - rejectGarbageAfterEnd: true, - }); + this.#transform = createTransform(GUNZIP); break; case 'brotli': - this.#handle = lazyZlib().createBrotliDecompress({ - rejectGarbageAfterEnd: true, - }); + this.#transform = createTransform(BROTLI_DECODE); break; } - this.#transform = newReadableWritablePairFromDuplex(this.#handle, { - [kValidateChunk]: validateBufferSourceChunk, - [kDestroyOnSyncError]: true, - }); } /** From e8933330533720d939666f1a19fee3492519ab85 Mon Sep 17 00:00:00 2001 From: Yagiz Nizipli Date: Thu, 20 Aug 2026 16:02:59 +0000 Subject: [PATCH 2/3] stream: fast-path TextEncoderStream chunk encoding The encode-and-enqueue algorithm was implemented as a literal transcription of the spec's per-code-unit loop: it extracted a single-character string, called charCodeAt(), and appended to an accumulator string for every code unit of every chunk, allocating millions of temporary strings for large payloads. The only observable effects of that loop are that a high surrogate at the end of a chunk is held back to pair with a low surrogate starting the next chunk, and that unpaired surrogates encode as U+FFFD, which TextEncoder already does. Handle the chunk boundary explicitly and encode the rest of the chunk with a single encode() call. Assisted-by: Cursor Signed-off-by: Yagiz Nizipli --- benchmark/webstreams/encoding.js | 52 +++++++++++++++++++++++++++++ lib/internal/webstreams/encoding.js | 47 +++++++++++++------------- 2 files changed, 75 insertions(+), 24 deletions(-) create mode 100644 benchmark/webstreams/encoding.js diff --git a/benchmark/webstreams/encoding.js b/benchmark/webstreams/encoding.js new file mode 100644 index 000000000000..b548e5a69b05 --- /dev/null +++ b/benchmark/webstreams/encoding.js @@ -0,0 +1,52 @@ +'use strict'; +const common = require('../common.js'); +const assert = require('node:assert'); + +const bench = common.createBenchmark(main, { + chunkSize: [4096], + totalBytes: [64 << 20], + kind: ['encode', 'decode', 'transcode'], +}); + +async function main({ chunkSize, totalBytes, kind }) { + const line = 'time=2024-01-01T00:00:00.000Z level=info request=42 ' + + 'method=GET path=/api/v1/items status=200 duration=13ms\n'; + const byteChunk = Buffer.alloc(chunkSize, line); + const stringChunk = byteChunk.toString(); + const nChunks = Math.ceil(totalBytes / chunkSize); + + let i = 0; + const input = kind === 'encode' ? stringChunk : byteChunk; + const source = new ReadableStream({ + pull(controller) { + if (i++ < nChunks) { + controller.enqueue(input); + } else { + controller.close(); + } + }, + }); + + let stream = source; + switch (kind) { + case 'encode': + stream = source.pipeThrough(new TextEncoderStream()); + break; + case 'decode': + stream = source.pipeThrough(new TextDecoderStream()); + break; + case 'transcode': + stream = source + .pipeThrough(new TextDecoderStream()) + .pipeThrough(new TextEncoderStream()); + break; + } + + let processed = 0; + bench.start(); + for await (const chunk of stream) { + processed += chunk.length; + } + bench.end(totalBytes / (1024 * 1024)); + assert.strictEqual(processed, nChunks * chunkSize); +} diff --git a/lib/internal/webstreams/encoding.js b/lib/internal/webstreams/encoding.js index f316222ccbf0..8e3af430d1a9 100644 --- a/lib/internal/webstreams/encoding.js +++ b/lib/internal/webstreams/encoding.js @@ -4,6 +4,7 @@ const { ObjectDefineProperties, String, StringPrototypeCharCodeAt, + StringPrototypeSlice, Uint8Array, } = primordials; @@ -46,32 +47,30 @@ class TextEncoderStream { this.#transform = new TransformStream({ transform: (chunk, controller) => { // https://encoding.spec.whatwg.org/#encode-and-enqueue-a-chunk + // + // The spec describes a per-code-unit loop whose only observable + // effects are (a) holding back a high surrogate that ends a chunk + // so it can be paired with a low surrogate starting the next chunk + // and (b) replacing unpaired surrogates with U+FFFD. The encoder + // already performs (b), so only the chunk boundaries need special + // handling here. chunk = String(chunk); - let finalChunk = ''; - for (let i = 0; i < chunk.length; i++) { - const item = chunk[i]; - const codeUnit = StringPrototypeCharCodeAt(item, 0); - if (this.#pendingHighSurrogate !== null) { - const highSurrogate = this.#pendingHighSurrogate; - this.#pendingHighSurrogate = null; - if (0xDC00 <= codeUnit && codeUnit <= 0xDFFF) { - finalChunk += highSurrogate + item; - continue; - } - finalChunk += '\uFFFD'; - } - if (0xD800 <= codeUnit && codeUnit <= 0xDBFF) { - this.#pendingHighSurrogate = item; - continue; - } - if (0xDC00 <= codeUnit && codeUnit <= 0xDFFF) { - finalChunk += '\uFFFD'; - continue; - } - finalChunk += item; + if (chunk.length === 0) { + return; } - if (finalChunk) { - const value = this.#handle.encode(finalChunk); + if (this.#pendingHighSurrogate !== null) { + chunk = this.#pendingHighSurrogate + chunk; + this.#pendingHighSurrogate = null; + } + const lastCodeUnit = StringPrototypeCharCodeAt(chunk, chunk.length - 1); + if (0xD800 <= lastCodeUnit && lastCodeUnit <= 0xDBFF) { + // A high surrogate at the end of the chunk may pair with a low + // surrogate at the start of the next one: hold it back. + this.#pendingHighSurrogate = StringPrototypeSlice(chunk, -1); + chunk = StringPrototypeSlice(chunk, 0, -1); + } + if (chunk) { + const value = this.#handle.encode(chunk); controller.enqueue(value); } }, From e3f6bd26b8e0f2111849e11648ccdf20d01e38bf Mon Sep 17 00:00:00 2001 From: Yagiz Nizipli Date: Thu, 20 Aug 2026 16:06:19 +0000 Subject: [PATCH 3/3] stream: remove unused sync-error destroy plumbing from adapters The kValidateChunk and kDestroyOnSyncError hooks existed only for the previous stream.Duplex-based CompressionStream implementation, which no longer uses the adapters. Assisted-by: Cursor Signed-off-by: Yagiz Nizipli --- lib/internal/webstreams/adapters.js | 49 +++++-------------- ...st-webstreams-adapters-sync-write-error.js | 3 +- 2 files changed, 13 insertions(+), 39 deletions(-) diff --git a/lib/internal/webstreams/adapters.js b/lib/internal/webstreams/adapters.js index 42e1221bb307..8488bd04424f 100644 --- a/lib/internal/webstreams/adapters.js +++ b/lib/internal/webstreams/adapters.js @@ -100,9 +100,6 @@ const { UV_EOF } = internalBinding('uv'); const encoder = new TextEncoder(); -const kValidateChunk = Symbol('kValidateChunk'); -const kDestroyOnSyncError = Symbol('kDestroyOnSyncError'); - // Collect all negative (error) ZLIB codes and Z_NEED_DICT const ZLIB_FAILURES = new SafeSet( ArrayPrototypeFilter( @@ -236,32 +233,18 @@ function newWritableStreamFromStreamWritable(streamWritable, options = kEmptyObj start(c) { controller = c; }, write(chunk) { - try { - options[kValidateChunk]?.(chunk); - if (!streamWritable.writableObjectMode && isAnyArrayBuffer(chunk)) { - chunk = new Uint8Array(chunk); - } - if (streamWritable.writableNeedDrain || !streamWritable.write(chunk)) { - backpressurePromise = PromiseWithResolvers(); - if (!streamWritable.writableNeedDrain) { - backpressurePromise.resolve(); - } - return SafePromisePrototypeFinally( - backpressurePromise.promise, () => { - backpressurePromise = undefined; - }); - } - } catch (error) { - // When the kDestroyOnSyncError flag is set (e.g. for - // CompressionStream), a sync throw must also destroy the - // stream so the readable side is errored too. Without this - // the readable side hangs forever. This replicates the - // TransformStream semantics: error both sides on any throw - // in the transform path. - if (options[kDestroyOnSyncError]) { - destroy(streamWritable, error); + if (!streamWritable.writableObjectMode && isAnyArrayBuffer(chunk)) { + chunk = new Uint8Array(chunk); + } + if (streamWritable.writableNeedDrain || !streamWritable.write(chunk)) { + backpressurePromise = PromiseWithResolvers(); + if (!streamWritable.writableNeedDrain) { + backpressurePromise.resolve(); } - throw error; + return SafePromisePrototypeFinally( + backpressurePromise.promise, () => { + backpressurePromise = undefined; + }); } }, @@ -694,15 +677,9 @@ function newReadableWritablePairFromDuplex(duplex, options = kEmptyObject) { return { readable, writable }; } - const writableOptions = { - __proto__: null, - [kValidateChunk]: options[kValidateChunk], - [kDestroyOnSyncError]: options[kDestroyOnSyncError], - }; - const writable = isWritable(duplex) ? - newWritableStreamFromStreamWritable(duplex, writableOptions) : + newWritableStreamFromStreamWritable(duplex) : new WritableStream(); if (!isWritable(duplex)) @@ -1093,6 +1070,4 @@ module.exports = { newStreamDuplexFromReadableWritablePair, newWritableStreamFromStreamBase, newReadableStreamFromStreamBase, - kValidateChunk, - kDestroyOnSyncError, }; diff --git a/test/parallel/test-webstreams-adapters-sync-write-error.js b/test/parallel/test-webstreams-adapters-sync-write-error.js index 748f682365ee..7123ed2e3e71 100644 --- a/test/parallel/test-webstreams-adapters-sync-write-error.js +++ b/test/parallel/test-webstreams-adapters-sync-write-error.js @@ -11,8 +11,7 @@ const { // Verify that when the underlying Node.js stream throws synchronously from // write(), the writable web stream properly rejects but does not destroy -// the stream (destroy-on-sync-throw is only used internally by -// CompressionStream/DecompressionStream). +// the stream. test('WritableStream from Node.js stream handles sync write throw', async () => { const error = new TypeError('invalid chunk');