From a1e91d8f68d50457a9cfbd5c6d8c0a67969fda21 Mon Sep 17 00:00:00 2001 From: seungwoo Date: Sat, 18 Jul 2026 23:49:24 +0900 Subject: [PATCH 1/2] stream: preserve mutable chunks in Web Stream adapters Wait for native write callbacks when they can safely represent chunk consumption. For Duplex streams, corked writes, and custom or legacy write methods, pass private BufferSource copies to preserve completion timing. Coordinate callback completion with backpressure, aborts, and stream errors so a settled Web Streams write no longer exposes mutable bytes still retained by the native stream. Preserve native HTTP validation and SharedArrayBuffer backing when fallback copies are required. Signed-off-by: seungwoo --- lib/_http_outgoing.js | 3 + lib/internal/streams/utils.js | 1 + lib/internal/streams/writable.js | 8 +- lib/internal/webstreams/adapters.js | 356 ++++++++++++++++++++-- lib/internal/webstreams/writablestream.js | 5 + 5 files changed, 345 insertions(+), 28 deletions(-) diff --git a/lib/_http_outgoing.js b/lib/_http_outgoing.js index 7694fe4b5b3f3e..6070eb79467c63 100644 --- a/lib/_http_outgoing.js +++ b/lib/_http_outgoing.js @@ -904,6 +904,8 @@ OutgoingMessage.prototype.write = function write(chunk, encoding, callback) { return ret; }; +const outgoingMessagePrototypeWrite = OutgoingMessage.prototype.write; + function onError(msg, err, callback) { if (msg.destroyed) { return; @@ -1260,4 +1262,5 @@ module.exports = { validateHeaderName, validateHeaderValue, OutgoingMessage, + outgoingMessagePrototypeWrite, }; diff --git a/lib/internal/streams/utils.js b/lib/internal/streams/utils.js index 45f55316104fe8..06b05bab01846f 100644 --- a/lib/internal/streams/utils.js +++ b/lib/internal/streams/utils.js @@ -345,6 +345,7 @@ module.exports = { isWritableEnded, isWritableFinished, isWritableErrored, + isOutgoingMessage, isServerRequest, isServerResponse, willEmitClose, diff --git a/lib/internal/streams/writable.js b/lib/internal/streams/writable.js index 47e003ea3a8a88..91e9c38ad631f6 100644 --- a/lib/internal/streams/writable.js +++ b/lib/internal/streams/writable.js @@ -512,6 +512,8 @@ Writable.prototype.write = function(chunk, encoding, cb) { return _write(this, chunk, encoding, cb) === true; }; +const writablePrototypeWrite = Writable.prototype.write; + Writable.prototype.cork = function() { const state = this._writableState; @@ -1150,7 +1152,11 @@ Writable.fromWeb = function(writableStream, options) { }; Writable.toWeb = function(streamWritable) { - return lazyWebStreams().newWritableStreamFromStreamWritable(streamWritable); + return lazyWebStreams().newWritableStreamFromStreamWritable( + streamWritable, + undefined, + writablePrototypeWrite, + ); }; Writable.prototype[SymbolAsyncDispose] = async function() { diff --git a/lib/internal/webstreams/adapters.js b/lib/internal/webstreams/adapters.js index d3f8212f1a135b..6617ac72fca753 100644 --- a/lib/internal/webstreams/adapters.js +++ b/lib/internal/webstreams/adapters.js @@ -1,25 +1,46 @@ 'use strict'; const { + ArrayBufferPrototypeGetDetached, ArrayPrototypeFilter, + BigInt64Array, + BigUint64Array, + DataView, + Float32Array, + Float64Array, + FunctionPrototypeCall, + Int16Array, + Int32Array, + Int8Array, ObjectKeys, PromisePrototypeThen, PromiseResolve, PromiseWithResolvers, SafePromiseAllReturnVoid, - SafePromisePrototypeFinally, SafeSet, StringPrototypeStartsWith, Symbol, + SymbolDispose, TypeError, TypedArrayPrototypeGetBuffer, TypedArrayPrototypeGetByteLength, TypedArrayPrototypeGetByteOffset, + TypedArrayPrototypeGetSymbolToStringTag, + TypedArrayPrototypeSet, + Uint16Array, + Uint32Array, Uint8Array, + Uint8ClampedArray, + globalThis: { + Float16Array, + }, } = primordials; const { TextEncoder } = require('internal/encoding'); +let addAbortListener; +let outgoingMessagePrototypeWrite; + const { ReadableStream, isReadableStream, @@ -27,6 +48,7 @@ const { const { WritableStream, + getWritableStreamDefaultControllerSignal, isWritableStream, } = require('internal/webstreams/writablestream'); @@ -44,6 +66,7 @@ const { const { isDestroyed, + isOutgoingMessage, isReadable, isWritable, isWritableEnded, @@ -55,8 +78,18 @@ const { const { isAnyArrayBuffer, + isArrayBufferView, + isDataView, + isSharedArrayBuffer, + isUint8Array, } = require('internal/util/types'); +const { + ArrayBufferViewGetBuffer, + ArrayBufferViewGetByteLength, + ArrayBufferViewGetByteOffset, +} = require('internal/webstreams/util'); + const { AbortError, ErrnoException, @@ -69,6 +102,7 @@ const { } = require('internal/errors'); const { + constructSharedArrayBuffer, getDeprecationWarningEmitter, kEmptyObject, normalizeEncoding, @@ -141,6 +175,62 @@ function handleKnownInternalErrors(cause) { const noop = () => {}; +function cloneArrayBufferView(view) { + const viewIsDataView = isDataView(view); + const buffer = ArrayBufferViewGetBuffer(view); + + // A detached backing buffer cannot be changed by the caller. Passing the + // original view through also preserves the native stream's validation. + if (!isSharedArrayBuffer(buffer) && + ArrayBufferPrototypeGetDetached(buffer)) { + return view; + } + + const byteLength = ArrayBufferViewGetByteLength(view); + const byteOffset = ArrayBufferViewGetByteOffset(view); + const copiedBytes = new Uint8Array( + isSharedArrayBuffer(buffer) ? + constructSharedArrayBuffer(byteLength) : + byteLength, + ); + TypedArrayPrototypeSet( + copiedBytes, + new Uint8Array(buffer, byteOffset, byteLength), + ); + const copiedBuffer = ArrayBufferViewGetBuffer(copiedBytes); + + if (viewIsDataView) { + return new DataView(copiedBuffer); + } + + switch (TypedArrayPrototypeGetSymbolToStringTag(view)) { + case 'BigInt64Array': + return new BigInt64Array(copiedBuffer); + case 'BigUint64Array': + return new BigUint64Array(copiedBuffer); + case 'Float16Array': + return new Float16Array(copiedBuffer); + case 'Float32Array': + return new Float32Array(copiedBuffer); + case 'Float64Array': + return new Float64Array(copiedBuffer); + case 'Int8Array': + return new Int8Array(copiedBuffer); + case 'Int16Array': + return new Int16Array(copiedBuffer); + case 'Int32Array': + return new Int32Array(copiedBuffer); + case 'Uint8Array': + return Buffer.isBuffer(view) ? Buffer.from(copiedBuffer) : copiedBytes; + case 'Uint8ClampedArray': + return new Uint8ClampedArray(copiedBuffer); + case 'Uint16Array': + return new Uint16Array(copiedBuffer); + case 'Uint32Array': + return new Uint32Array(copiedBuffer); + } +} + /** * @typedef {import('../../stream').Writable} Writable * @typedef {import('../../stream').Readable} Readable @@ -155,9 +245,14 @@ const noop = () => {}; /** * @param {Writable} streamWritable * @param {object} [options] + * @param {Function} [writablePrototypeWrite] * @returns {WritableStream} */ -function newWritableStreamFromStreamWritable(streamWritable, options = kEmptyObject) { +function newWritableStreamFromStreamWritable( + streamWritable, + options = kEmptyObject, + writablePrototypeWrite, +) { // Not using the internal/streams/utils isWritableNodeStream utility // here because it will return false if streamWritable is a Duplex // whose writable option is false. For a Duplex that is not writable, @@ -193,22 +288,188 @@ function newWritableStreamFromStreamWritable(streamWritable, options = kEmptyObj }; let controller; - let backpressurePromise; + let pendingWriteState; + let isTerminated = false; + let terminalReason; let closed; + let abortSignal; + let abortDisposable; + + function disposeAbortListener() { + abortDisposable?.[SymbolDispose](); + abortDisposable = undefined; + } + + function recordTerminalReason(reason) { + if (!isTerminated) { + isTerminated = true; + terminalReason = reason; + } + } + + function resolvePendingWrite(state) { + if (pendingWriteState !== state) { + return; + } + pendingWriteState = undefined; + disposeAbortListener(); + state.deferred.resolve(); + } + + function rejectPendingWrite(error) { + const state = pendingWriteState; + if (state === undefined) { + return; + } + pendingWriteState = undefined; + disposeAbortListener(); + state.deferred.reject(error); + } + + function listenForPendingAbort() { + addAbortListener ??= + require('internal/events/abort_listener').addAbortListener; + abortDisposable = addAbortListener(abortSignal, () => { + // Release an in-flight write so the sink's abort algorithm can run. + // The abort algorithm is solely responsible for destroying the stream. + recordTerminalReason(abortSignal.reason); + rejectPendingWrite(abortSignal.reason); + }); + } + + function maybeResolvePendingWrite(state) { + if (state.writeComplete && state.backpressureCleared) { + resolvePendingWrite(state); + } + } + + function markWriteComplete(state, error) { + if (pendingWriteState !== state) { + return; + } + if (error != null) { + rejectPendingWrite(error); + return; + } + state.writeComplete = true; + maybeResolvePendingWrite(state); + } + + function createPendingWriteState(waitForWriteCallback) { + const state = { + __proto__: null, + deferred: PromiseWithResolvers(), + writeComplete: !waitForWriteCallback, + backpressureCleared: false, + }; + pendingWriteState = state; + listenForPendingAbort(); + return state; + } function onDrain() { - backpressurePromise?.resolve(); + const state = pendingWriteState; + if (state === undefined) { + return; + } + state.backpressureCleared = true; + maybeResolvePendingWrite(state); + } + + function onWrite(error) { + const state = pendingWriteState; + if (state === undefined) { + return; + } + if (error != null) { + error = handleKnownInternalErrors(error); + } + markWriteComplete(state, error); + } + + function throwSyncWriteError(error) { + disposeAbortListener(); + // 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); + } + throw error; + } + + function throwIfTerminated() { + if (abortSignal.aborted) { + recordTerminalReason(abortSignal.reason); + } + if (isTerminated) { + throw terminalReason; + } + } + + function writeChunk( + writeMethod, + chunk, + needDrainBeforeWrite, + waitForWriteCallback, + ) { + if (!waitForWriteCallback && !needDrainBeforeWrite) { + try { + const writeReturn = FunctionPrototypeCall( + writeMethod, + streamWritable, + chunk, + ); + throwIfTerminated(); + if (controller === undefined || + writeReturn || + !streamWritable.writableNeedDrain) { + return; + } + } catch (error) { + throwSyncWriteError(error); + } + + const state = createPendingWriteState(false); + return state.deferred.promise; + } + + const state = createPendingWriteState(waitForWriteCallback); + try { + // The return value only reports backpressure. The callback reports when + // the chunk has been handled and can safely be mutated by the caller. + const writeReturn = waitForWriteCallback ? + FunctionPrototypeCall(writeMethod, streamWritable, chunk, onWrite) : + FunctionPrototypeCall(writeMethod, streamWritable, chunk); + throwIfTerminated(); + const shouldWaitForDrain = + (needDrainBeforeWrite || !writeReturn) && + streamWritable.writableNeedDrain; + state.backpressureCleared = !shouldWaitForDrain; + maybeResolvePendingWrite(state); + } catch (error) { + resolvePendingWrite(state); + PromisePrototypeThen(state.deferred.promise, undefined, noop); + throwSyncWriteError(error); + } + + return state.deferred.promise; } const cleanup = eos(streamWritable, (error) => { error = handleKnownInternalErrors(error); cleanup(); + disposeAbortListener(); // This is a protection against non-standard, legacy streams // that happen to emit an error event again after finished is called. streamWritable.on('error', () => {}); if (error != null) { - backpressurePromise?.reject(error); + recordTerminalReason(error); + rejectPendingWrite(error); // If closed is not undefined, the error is happening // after the WritableStream close has already started. // We need to reject it here. @@ -226,57 +487,98 @@ function newWritableStreamFromStreamWritable(streamWritable, options = kEmptyObj closed = undefined; return; } - controller.error(new AbortError()); + const abortError = new AbortError(); + recordTerminalReason(abortError); + rejectPendingWrite(abortError); + controller.error(abortError); controller = undefined; }); streamWritable.on('drain', onDrain); return new WritableStream({ - start(c) { controller = c; }, + start(c) { + controller = c; + abortSignal = getWritableStreamDefaultControllerSignal(c); + }, write(chunk) { + let writeMethod; + let needDrainBeforeWrite; + let waitForWriteCallback; try { options[kValidateChunk]?.(chunk); - if (!streamWritable.writableObjectMode && isAnyArrayBuffer(chunk)) { + writeMethod = streamWritable.write; + const writableState = streamWritable._writableState; + const objectMode = streamWritable.writableObjectMode; + const isBufferSourceChunk = + isAnyArrayBuffer(chunk) || isArrayBufferView(chunk); + if (!objectMode && isAnyArrayBuffer(chunk)) { chunk = new Uint8Array(chunk); } - if (streamWritable.writableNeedDrain || !streamWritable.write(chunk)) { - backpressurePromise = PromiseWithResolvers(); - if (!streamWritable.writableNeedDrain) { - backpressurePromise.resolve(); + needDrainBeforeWrite = streamWritable.writableNeedDrain; + const isArrayBufferViewChunk = isArrayBufferView(chunk); + if (!objectMode && + isOutgoingMessage(streamWritable) && + isArrayBufferViewChunk && + !isUint8Array(chunk)) { + outgoingMessagePrototypeWrite ??= + require('_http_outgoing').outgoingMessagePrototypeWrite; + if (writeMethod === outgoingMessagePrototypeWrite) { + throw new ERR_INVALID_ARG_TYPE( + 'chunk', ['string', 'Buffer', 'Uint8Array'], chunk); } - 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); + const canAwaitWriteCallback = + streamWritable instanceof Writable && + writeMethod === writablePrototypeWrite && + streamWritable._readableState === undefined && + writableState?.corked === 0; + waitForWriteCallback = + isBufferSourceChunk && canAwaitWriteCallback; + if (!objectMode && + isArrayBufferViewChunk && + !waitForWriteCallback) { + // Duplex streams can retain chunks in their readable side or + // forward them after their own write callback. A corked stream does + // not invoke callbacks until uncork() or end(), and streams that + // override write() may not support callbacks at all. Preserve the + // existing completion timing for those cases by passing a private, + // same-type copy to the native stream. + chunk = cloneArrayBufferView(chunk); } - throw error; + } catch (error) { + throwSyncWriteError(error); } + + return writeChunk( + writeMethod, + chunk, + needDrainBeforeWrite, + waitForWriteCallback, + ); }, abort(reason) { + disposeAbortListener(); destroy(streamWritable, reason); }, close() { if (closed === undefined && !isWritableEnded(streamWritable)) { closed = PromiseWithResolvers(); - streamWritable.end(); + try { + streamWritable.end(); + } catch (error) { + closed = undefined; + disposeAbortListener(); + throw error; + } return closed.promise; } controller = undefined; + disposeAbortListener(); return PromiseResolve(); }, }, strategy); diff --git a/lib/internal/webstreams/writablestream.js b/lib/internal/webstreams/writablestream.js index f0eb687be129d1..3b570a0af12d10 100644 --- a/lib/internal/webstreams/writablestream.js +++ b/lib/internal/webstreams/writablestream.js @@ -596,6 +596,10 @@ const isWritableStreamDefaultWriter = const isWritableStreamDefaultController = isBrandCheck('WritableStreamDefaultController'); +function getWritableStreamDefaultControllerSignal(controller) { + return controller[kState].abortController.signal; +} + function createWritableStreamState() { return { __proto__: null, @@ -1357,6 +1361,7 @@ module.exports = { isWritableStreamDefaultController, isWritableStreamDefaultWriter, + getWritableStreamDefaultControllerSignal, isWritableStreamLocked, setupWritableStreamDefaultWriter, writableStreamAbort, From 97f53188e56a369f98b3f79c58727c3e9f348ce9 Mon Sep 17 00:00:00 2001 From: seungwoo Date: Sun, 19 Jul 2026 00:25:42 +0900 Subject: [PATCH 2/2] test: cover mutable Web Stream adapter writes Cover native callback completion, fallback copies, aborts, and error propagation for mutable BufferSource chunks passed to Node.js Web Streams adapters. Verify HTTP validation, SharedArrayBuffer backing, and Duplex and compression paths. Signed-off-by: seungwoo --- test/parallel/test-stream-duplex.js | 4 +- ...st-webstreams-adapters-sync-write-error.js | 75 +++- ...treams-adapters-writable-buffer-sources.js | 360 +++++++++++++++++- ...st-webstreams-compression-buffer-source.js | 29 ++ 4 files changed, 463 insertions(+), 5 deletions(-) diff --git a/test/parallel/test-stream-duplex.js b/test/parallel/test-stream-duplex.js index de67290a7e76cd..9908d000cb7ec8 100644 --- a/test/parallel/test-stream-duplex.js +++ b/test/parallel/test-stream-duplex.js @@ -120,7 +120,7 @@ process.on('exit', () => { this.push(null); }, write: common.mustCall((chunk) => { - assert.strictEqual(chunk, dataToWrite); + assert.deepStrictEqual(chunk, dataToWrite); }) }); @@ -143,7 +143,7 @@ process.on('exit', () => { this.push(null); }, write: common.mustCall((chunk) => { - assert.strictEqual(chunk, dataToWrite); + assert.deepStrictEqual(chunk, dataToWrite); }) }); diff --git a/test/parallel/test-webstreams-adapters-sync-write-error.js b/test/parallel/test-webstreams-adapters-sync-write-error.js index 748f682365ee93..ff6dbddf377f85 100644 --- a/test/parallel/test-webstreams-adapters-sync-write-error.js +++ b/test/parallel/test-webstreams-adapters-sync-write-error.js @@ -1,6 +1,6 @@ 'use strict'; // Flags: --no-warnings --expose-internals -require('../common'); +const common = require('../common'); const assert = require('assert'); const test = require('node:test'); const { Duplex, Writable } = require('stream'); @@ -9,6 +9,13 @@ const { newReadableWritablePairFromDuplex, } = require('internal/webstreams/adapters'); +function isSameError(expected) { + return common.mustCall((actual) => { + assert.strictEqual(actual, expected); + return true; + }); +} + // 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 @@ -34,6 +41,72 @@ test('WritableStream from Node.js stream handles sync write throw', async () => assert.strictEqual(writable.destroyed, false); }); +test('WritableStream from Node.js stream handles async write error', async () => { + const error = new Error('boom'); + const writable = new Writable({ + write(_chunk, _encoding, callback) { + setImmediate(callback, error); + }, + }); + const writer = Writable.toWeb(writable).getWriter(); + + await Promise.all([ + assert.rejects(writer.write(Buffer.from('hello')), isSameError(error)), + assert.rejects(writer.closed, isSameError(error)), + ]); +}); + +test('WritableStream aborts while a native write is pending', async () => { + const error = new Error('abort'); + let finishWrite; + let startWrite; + const writeStarted = new Promise((resolve) => { + startWrite = resolve; + }); + const writable = new Writable({ + write(_chunk, _encoding, callback) { + finishWrite = callback; + startWrite(); + }, + }); + const writer = Writable.toWeb(writable).getWriter(); + const writePromise = writer.write(Buffer.from('hello')); + await writeStarted; + + const writeRejected = assert.rejects(writePromise, isSameError(error)); + const closedRejected = assert.rejects(writer.closed, isSameError(error)); + await Promise.all([ + writer.abort(error), + writeRejected, + closedRejected, + ]); + + finishWrite(); + await new Promise(setImmediate); + assert.strictEqual(writable.destroyed, true); +}); + +test('WritableStream handles destruction while a write is pending', async () => { + const error = new Error('destroy'); + let startWrite; + const writeStarted = new Promise((resolve) => { + startWrite = resolve; + }); + const writable = new Writable({ + write(_chunk, _encoding, _callback) { + startWrite(); + }, + }); + const writer = Writable.toWeb(writable).getWriter(); + const writePromise = writer.write(Buffer.from('hello')); + await writeStarted; + + const writeRejected = assert.rejects(writePromise, isSameError(error)); + const closedRejected = assert.rejects(writer.closed, isSameError(error)); + writable.destroy(error); + await Promise.all([writeRejected, closedRejected]); +}); + test('Duplex-backed pair does NOT destroy on sync write throw', async () => { const error = new TypeError('invalid chunk'); const duplex = new Duplex({ diff --git a/test/parallel/test-webstreams-adapters-writable-buffer-sources.js b/test/parallel/test-webstreams-adapters-writable-buffer-sources.js index 995db97e747364..1a4a8eeb01f8ba 100644 --- a/test/parallel/test-webstreams-adapters-writable-buffer-sources.js +++ b/test/parallel/test-webstreams-adapters-writable-buffer-sources.js @@ -3,11 +3,46 @@ const common = require('../common'); const assert = require('assert'); const { Buffer } = require('buffer'); +const { ServerResponse } = require('http'); const { Duplex, Writable } = require('stream'); const { suite, test } = require('node:test'); const ctors = [ArrayBuffer, SharedArrayBuffer]; +function createServerResponse() { + const socket = new Writable({ + write(chunk, encoding, callback) { + callback(); + }, + }); + const response = new ServerResponse({ + method: 'GET', + httpVersionMajor: 1, + httpVersionMinor: 1, + }); + socket.on('error', common.mustNotCall()); + response.on('error', common.mustNotCall()); + response.assignSocket(socket); + return response; +} + +async function completesWithin(promise) { + let timer; + try { + return await Promise.race([ + promise, + new Promise((_, reject) => { + timer = setTimeout( + () => reject(new Error('write timed out')), + common.platformTimeout(1000), + ); + }), + ]); + } finally { + clearTimeout(timer); + } +} + suite('underlying Writable', () => { suite('in non-object mode', () => { for (const ctor of ctors) { @@ -26,6 +61,241 @@ suite('underlying Writable', () => { await writer.write(buffer); }); } + + for (const highWaterMark of [1, 16]) { + test(`waits for mutable chunks with highWaterMark ${highWaterMark}`, + async () => { + let finishWrite; + let consumed; + const writable = new Writable({ + highWaterMark, + write(chunk, encoding, callback) { + finishWrite = () => { + consumed = Buffer.from(chunk); + callback(); + }; + }, + }); + writable.on('error', common.mustNotCall()); + const writer = Writable.toWeb(writable).getWriter(); + const input = new Uint8Array([1, 2, 3, 4]); + let settled = false; + const writePromise = writer.write(input).then(() => { + settled = true; + }); + + await new Promise(setImmediate); + assert.strictEqual(settled, false); + finishWrite(); + await writePromise; + input.fill(9); + await writer.close(); + + assert.deepStrictEqual( + consumed, + Buffer.from([1, 2, 3, 4]), + ); + }); + } + + test('copies mutable chunks while the Writable is corked', async () => { + let consumed; + const writable = new Writable({ + write(chunk, encoding, callback) { + consumed = Buffer.from(chunk); + callback(); + }, + }); + writable.on('error', common.mustNotCall()); + writable.cork(); + const writer = Writable.toWeb(writable).getWriter(); + const input = new Uint8Array([1, 2, 3, 4]); + + await writer.write(input); + input.fill(9); + writable.uncork(); + await writer.close(); + + assert.deepStrictEqual(consumed, Buffer.from([1, 2, 3, 4])); + }); + + test('copies mutable chunks when write() is overridden', async () => { + let consumed; + let received; + let notifyConsumed; + const consumedPromise = new Promise((resolve) => { + notifyConsumed = resolve; + }); + const writable = new Writable({ + write(chunk, encoding, callback) { + callback(); + }, + }); + writable.on('error', common.mustNotCall()); + const writer = Writable.toWeb(writable).getWriter(); + writable.write = common.mustCall((chunk) => { + received = chunk; + setImmediate(() => { + consumed = Buffer.from(chunk); + notifyConsumed(); + }); + return true; + }); + const input = new Uint8Array([1, 2, 3, 4]); + + await writer.write(input); + input.fill(9); + await consumedPromise; + await writer.close(); + + assert.notStrictEqual(received.buffer, input.buffer); + assert.deepStrictEqual(consumed, Buffer.from([1, 2, 3, 4])); + }); + + test('does not trust a patched Writable.prototype.write', async () => { + const originalWrite = Writable.prototype.write; + let consumed; + let notifyConsumed; + const consumedPromise = new Promise((resolve) => { + notifyConsumed = resolve; + }); + const writable = new Writable({ + write(chunk, encoding, callback) { + setImmediate(() => { + consumed = Buffer.from(chunk); + callback(); + notifyConsumed(); + }); + }, + }); + writable.on('error', common.mustNotCall()); + let writer; + + Writable.prototype.write = function(chunk) { + return originalWrite.call(this, chunk); + }; + try { + writer = Writable.toWeb(writable).getWriter(); + const input = new Uint8Array([1, 2, 3, 4]); + await completesWithin(writer.write(input)); + input.fill(9); + await consumedPromise; + + assert.deepStrictEqual(consumed, Buffer.from([1, 2, 3, 4])); + } finally { + Writable.prototype.write = originalWrite; + } + await writer.close(); + }); + + test('reads write() once for classification and invocation', async () => { + const originalWrite = Writable.prototype.write; + let consumed; + const writable = new Writable({ + write(chunk, encoding, callback) { + setImmediate(() => { + consumed = Buffer.from(chunk); + callback(); + }); + }, + }); + writable.on('error', common.mustNotCall()); + const writer = Writable.toWeb(writable).getWriter(); + let writeAccesses = 0; + Object.defineProperty(writable, 'write', { + configurable: true, + get() { + writeAccesses++; + return writeAccesses === 1 ? + originalWrite : + function callbacklessWrite(chunk) { + return originalWrite.call(this, chunk); + }; + }, + }); + + const input = new Uint8Array([1, 2, 3, 4]); + await completesWithin(writer.write(input)); + input.fill(9); + + assert.strictEqual(writeAccesses, 1); + assert.deepStrictEqual(consumed, Buffer.from([1, 2, 3, 4])); + delete writable.write; + await writer.close(); + }); + + test('preserves cloned view brands and SharedArrayBuffer backing', + async () => { + const dataView = new DataView( + Uint8Array.from([0, 1, 2, 3, 4, 0]).buffer, + 1, + 4, + ); + const uint16Buffer = new ArrayBuffer(6); + const uint16 = new Uint16Array(uint16Buffer, 2, 2); + new Uint8Array(uint16Buffer, 2, 4).set([1, 2, 3, 4]); + const shared = new SharedArrayBuffer(6); + const sharedView = new Uint8Array(shared, 1, 4); + sharedView.set([1, 2, 3, 4]); + const inputs = [ + Buffer.from([1, 2, 3, 4]), + dataView, + uint16, + sharedView, + ]; + const expected = inputs.map((chunk) => ({ + brand: Buffer.isBuffer(chunk) ? + 'Buffer' : Object.prototype.toString.call(chunk), + bytes: Buffer.from(new Uint8Array( + chunk.buffer, + chunk.byteOffset, + chunk.byteLength, + )), + shared: chunk.buffer instanceof SharedArrayBuffer, + })); + const received = []; + const writable = new Writable({ + write(chunk, encoding, callback) { + callback(); + }, + }); + writable.on('error', common.mustNotCall()); + const writer = Writable.toWeb(writable).getWriter(); + writable.write = common.mustCall((chunk) => { + received.push(chunk); + return true; + }, inputs.length); + + for (const chunk of inputs) { + await writer.write(chunk); + new Uint8Array( + chunk.buffer, + chunk.byteOffset, + chunk.byteLength, + ).fill(9); + } + await writer.close(); + + for (let i = 0; i < received.length; i++) { + const actual = received[i]; + const actualBrand = Buffer.isBuffer(actual) ? + 'Buffer' : Object.prototype.toString.call(actual); + assert.strictEqual(actualBrand, expected[i].brand); + assert.notStrictEqual(actual.buffer, inputs[i].buffer); + assert.strictEqual( + actual.buffer instanceof SharedArrayBuffer, + expected[i].shared, + ); + assert.deepStrictEqual( + Buffer.from(new Uint8Array( + actual.buffer, + actual.byteOffset, + actual.byteLength, + )), + expected[i].bytes, + ); + } + }); }); suite('in object mode', () => { @@ -48,16 +318,71 @@ suite('underlying Writable', () => { }); }); +suite('underlying ServerResponse', () => { + test('rejects invalid view types before cloning', async () => { + const response = createServerResponse(); + const writer = Writable.toWeb(response).getWriter(); + + try { + await Promise.all([ + assert.rejects(writer.write(new DataView(new ArrayBuffer(4))), { + code: 'ERR_INVALID_ARG_TYPE', + }), + assert.rejects(writer.closed, { + code: 'ERR_INVALID_ARG_TYPE', + }), + ]); + } finally { + await new Promise((resolve) => response.end(resolve)); + } + }); + + test('preserves write() overrides', async () => { + const response = createServerResponse(); + const writer = Writable.toWeb(response).getWriter(); + const originalWrite = response.write; + let received; + response.write = common.mustCall((chunk) => { + received = chunk; + return true; + }); + const input = new DataView(Uint8Array.from([1, 2, 3, 4]).buffer); + + try { + await writer.write(input); + new Uint8Array(input.buffer).fill(9); + assert(received instanceof DataView); + assert.notStrictEqual(received.buffer, input.buffer); + assert.deepStrictEqual( + Buffer.from(received.buffer), + Buffer.from([1, 2, 3, 4]), + ); + } finally { + response.write = originalWrite; + await new Promise((resolve) => response.end(resolve)); + } + }); +}); + suite('underlying Duplex', () => { suite('in non-object mode', () => { for (const ctor of ctors) { - test(`converts ${ctor.name} chunks`, async () => { + test(`copies ${ctor.name} chunks`, async () => { const buffer = new ctor(4); + new Uint8Array(buffer).set([1, 2, 3, 4]); const duplex = new Duplex({ writableObjectMode: false, write: common.mustCall((chunk, encoding, callback) => { assert(Buffer.isBuffer(chunk)); - assert.strictEqual(chunk.buffer, buffer); + assert.notStrictEqual(chunk.buffer, buffer); + assert.strictEqual( + chunk.buffer instanceof SharedArrayBuffer, + buffer instanceof SharedArrayBuffer, + ); + assert.deepStrictEqual( + chunk, + Buffer.from([1, 2, 3, 4]), + ); callback(); }), read() { @@ -69,6 +394,37 @@ suite('underlying Duplex', () => { await writer.write(buffer); }); } + + test('copies mutable chunks without waiting for the write callback', + async () => { + let consumed; + let finishWrite; + const duplex = new Duplex({ + write(chunk, encoding, callback) { + finishWrite = () => { + consumed = Buffer.from(chunk); + callback(); + }; + }, + read() { + this.push(null); + }, + }); + duplex.on('error', common.mustNotCall()); + const writer = Duplex.toWeb(duplex).writable.getWriter(); + const input = new Uint8Array([1, 2, 3, 4]); + + await writer.write(input); + input.fill(9); + finishWrite(); + await new Promise(setImmediate); + + assert.deepStrictEqual( + consumed, + Buffer.from([1, 2, 3, 4]), + ); + await writer.close(); + }); }); suite('in object mode', () => { diff --git a/test/parallel/test-webstreams-compression-buffer-source.js b/test/parallel/test-webstreams-compression-buffer-source.js index 3304a8e64f3175..a81d4b674b3805 100644 --- a/test/parallel/test-webstreams-compression-buffer-source.js +++ b/test/parallel/test-webstreams-compression-buffer-source.js @@ -3,6 +3,7 @@ require('../common'); const assert = require('assert'); const test = require('node:test'); const { DecompressionStream, CompressionStream } = require('stream/web'); +const { gzipSync } = require('zlib'); // Minimal gzip-compressed bytes for "hello" const compressedGzip = new Uint8Array([ @@ -40,3 +41,31 @@ test('CompressionStream round-trip with ArrayBuffer input', async () => { const result = Buffer.concat(out.map((c) => Buffer.from(c))); assert.strictEqual(result.toString(), 'hello'); }); + +test('DecompressionStream writable completion is not coupled to readable ' + + 'backpressure', async () => { + const expected = Buffer.alloc(1024 * 1024, 0x61); + const compressed = gzipSync(expected); + const ds = new DecompressionStream('gzip'); + const writer = ds.writable.getWriter(); + let settled = false; + const writePromise = writer.write(compressed).then(() => { + settled = true; + }); + + await new Promise(setImmediate); + const settledBeforeRead = settled; + if (settledBeforeRead) { + compressed.fill(0); + } + + const outputPromise = Array.fromAsync(ds.readable); + await writePromise; + await writer.close(); + const output = Buffer.concat( + (await outputPromise).map((chunk) => Buffer.from(chunk)), + ); + + assert.strictEqual(settledBeforeRead, true); + assert.deepStrictEqual(output, expected); +});