stream: improve Web Compression spec compliance

Pass rejectGarbageAfterEnd: true to createInflateRaw() and
createBrotliDecompress(), matching the behavior already in place
for deflate and gzip. The Compression Streams spec treats any
data following a valid compressed payload as an error.

When the underlying Node.js stream throws synchronously from
write() (e.g. zlib rejects an invalid chunk type), destroy the
stream so that the readable side is also errored. Without this,
the readable side hangs forever waiting for data that will never
arrive.

Introduce a kValidateChunk callback option in the webstreams
adapter layer. Compression streams use this to validate that
chunks are BufferSource instances not backed by SharedArrayBuffer,
replacing the previous monkey-patching of the underlying
handle's write method.

Unskip WPT compression bad-chunks tests which now run instead of
hang and mark the remaining expected failures.

Author: panva

lib/internal/webstreams/adapters.js

@@ -11,6 +11,8 @@ const {
   SafePromiseAll,
   SafePromisePrototypeFinally,
   SafeSet,
+  StringPrototypeStartsWith,
+  Symbol,
   TypeError,
   TypedArrayPrototypeGetBuffer,
   TypedArrayPrototypeGetByteLength,
@@ -94,6 +96,9 @@ 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(
@@ -139,9 +144,10 @@ function handleKnownInternalErrors(cause) {
 
 /**
  * @param {Writable} streamWritable
+ * @param {object} [options]
  * @returns {WritableStream}
  */
-function newWritableStreamFromStreamWritable(streamWritable) {
+function newWritableStreamFromStreamWritable(streamWritable, options = kEmptyObject) {
   // 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,
@@ -220,12 +226,26 @@ function newWritableStreamFromStreamWritable(streamWritable) {
       if (!streamWritable.writableObjectMode && isArrayBuffer(chunk)) {
         chunk = new Uint8Array(chunk);
       }
-      if (streamWritable.writableNeedDrain || !streamWritable.write(chunk)) {
-        backpressurePromise = PromiseWithResolvers();
-        return SafePromisePrototypeFinally(
-          backpressurePromise.promise, () => {
-            backpressurePromise = undefined;
-          });
+      try {
+        options[kValidateChunk]?.(chunk);
+        if (streamWritable.writableNeedDrain || !streamWritable.write(chunk)) {
+          backpressurePromise = PromiseWithResolvers();
+          return SafePromisePrototypeFinally(
+            backpressurePromise.promise, () => {
+              backpressurePromise = undefined;
+            });
+        }
+      } catch (error) {
+        // When used as one half of a duplex-backed
+        // ReadableWritablePair (e.g. CompressionStream), a sync
+        // throw must also destroy the underlying stream so the
+        // readable side is errored too. Without this the readable
+        // side hangs forever. Only opt-in callers get this
+        // behavior.
+        if (options[kDestroyOnSyncError]) {
+          destroy(streamWritable, error);
+        }
+        throw error;
       }
     },
 
@@ -662,9 +682,15 @@ function newReadableWritablePairFromDuplex(duplex, options = kEmptyObject) {
     return { readable, writable };
   }
 
+  const writableOptions = {
+    __proto__: null,
+    [kValidateChunk]: options[kValidateChunk],
+    [kDestroyOnSyncError]: true,
+  };
+
   const writable =
     isWritable(duplex) ?
-      newWritableStreamFromStreamWritable(duplex) :
+      newWritableStreamFromStreamWritable(duplex, writableOptions) :
       new WritableStream();
 
   if (!isWritable(duplex))
@@ -1064,4 +1090,5 @@ module.exports = {
   newStreamDuplexFromReadableWritablePair,
   newWritableStreamFromStreamBase,
   newReadableStreamFromStreamBase,
+  kValidateChunk,
 };

lib/internal/webstreams/compression.js

@@ -7,15 +7,27 @@ const {
 
 const {
   newReadableWritablePairFromDuplex,
+  kValidateChunk,
 } = require('internal/webstreams/adapters');
 
 const { customInspect } = require('internal/webstreams/util');
 
+const {
+  isArrayBufferView,
+  isSharedArrayBuffer,
+} = require('internal/util/types');
+
 const {
   customInspectSymbol: kInspect,
   kEnumerableProperty,
 } = require('internal/util');
 
+const {
+  codes: {
+    ERR_INVALID_ARG_TYPE,
+  },
+} = require('internal/errors');
+
 const { createEnumConverter } = require('internal/webidl');
 
 let zlib;
@@ -24,6 +36,18 @@ function lazyZlib() {
   return zlib;
 }
 
+// Per the Compression Streams spec, chunks must be BufferSource
+// (ArrayBuffer or ArrayBufferView not backed by SharedArrayBuffer).
+function validateBufferSourceChunk(chunk) {
+  if (isArrayBufferView(chunk) && isSharedArrayBuffer(chunk.buffer)) {
+    throw new ERR_INVALID_ARG_TYPE(
+      'chunk',
+      ['Buffer', 'TypedArray', 'DataView'],
+      chunk,
+    );
+  }
+}
+
 const formatConverter = createEnumConverter('CompressionFormat', [
   'deflate',
   'deflate-raw',
@@ -62,7 +86,9 @@ class CompressionStream {
         this.#handle = lazyZlib().createBrotliCompress();
         break;
     }
-    this.#transform = newReadableWritablePairFromDuplex(this.#handle);
+    this.#transform = newReadableWritablePairFromDuplex(this.#handle, {
+      [kValidateChunk]: validateBufferSourceChunk,
+    });
   }
 
   /**
@@ -108,25 +134,23 @@ class DecompressionStream {
         });
         break;
       case 'deflate-raw':
-        this.#handle = lazyZlib().createInflateRaw();
+        this.#handle = lazyZlib().createInflateRaw({
+          rejectGarbageAfterEnd: true,
+        });
         break;
       case 'gzip':
         this.#handle = lazyZlib().createGunzip({
           rejectGarbageAfterEnd: true,
         });
         break;
       case 'brotli':
-        this.#handle = lazyZlib().createBrotliDecompress();
+        this.#handle = lazyZlib().createBrotliDecompress({
+          rejectGarbageAfterEnd: true,
+        });
         break;
     }
-    this.#transform = newReadableWritablePairFromDuplex(this.#handle);
-
-    this.#handle.on('error', (err) => {
-      if (this.#transform?.writable &&
-        !this.#transform.writable.locked &&
-        typeof this.#transform.writable.abort === 'function') {
-        this.#transform.writable.abort(err);
-      }
+    this.#transform = newReadableWritablePairFromDuplex(this.#handle, {
+      [kValidateChunk]: validateBufferSourceChunk,
     });
   }
 

test/parallel/test-webstreams-adapters-sync-write-error.js

@@ -0,0 +1,77 @@
+'use strict';
+// Flags: --no-warnings --expose-internals
+require('../common');
+const assert = require('assert');
+const test = require('node:test');
+const { Duplex, Writable } = require('stream');
+const {
+  newWritableStreamFromStreamWritable,
+  newReadableWritablePairFromDuplex,
+} = require('internal/webstreams/adapters');
+
+// Verify that when the underlying Node.js stream throws synchronously from
+// write(), the writable web stream properly rejects.
+
+test('WritableStream from Node.js stream handles sync write throw', async () => {
+  const error = new TypeError('invalid chunk');
+  const writable = new Writable({
+    write() {
+      throw error;
+    },
+  });
+
+  const ws = newWritableStreamFromStreamWritable(writable);
+  const writer = ws.getWriter();
+
+  await assert.rejects(writer.write('bad'), (err) => {
+    assert.strictEqual(err, error);
+    return true;
+  });
+
+  // Standalone writable should not be destroyed on sync write error
+  assert.strictEqual(writable.destroyed, false);
+});
+
+test('Duplex-backed pair destroys on sync write throw', async () => {
+  const error = new TypeError('invalid chunk');
+  const duplex = new Duplex({
+    read() {},
+    write() {
+      throw error;
+    },
+  });
+
+  const { writable, readable } = newReadableWritablePairFromDuplex(duplex);
+  const writer = writable.getWriter();
+  const reader = readable.getReader();
+
+  await assert.rejects(writer.write('bad'), (err) => {
+    assert.strictEqual(err, error);
+    return true;
+  });
+
+  // The duplex must be destroyed so the readable side is also errored
+  assert.strictEqual(duplex.destroyed, true);
+
+  // The readable side should also be errored, not hang
+  await assert.rejects(reader.read());
+});
+
+test('WritableStream from Node.js stream - valid writes still work', async () => {
+  const chunks = [];
+  const writable = new Writable({
+    write(chunk, _encoding, cb) {
+      chunks.push(chunk);
+      cb();
+    },
+  });
+
+  const ws = newWritableStreamFromStreamWritable(writable);
+  const writer = ws.getWriter();
+
+  await writer.write(Buffer.from('hello'));
+  await writer.write(Buffer.from(' world'));
+  await writer.close();
+
+  assert.strictEqual(Buffer.concat(chunks).toString(), 'hello world');
+});

test/parallel/test-webstreams-compression-bad-chunks.js

@@ -0,0 +1,57 @@
+'use strict';
+require('../common');
+const assert = require('assert');
+const test = require('node:test');
+const { CompressionStream, DecompressionStream } = require('stream/web');
+
+// Verify that writing invalid (non-BufferSource) chunks to
+// CompressionStream and DecompressionStream properly rejects
+// on both the write and the read side, instead of hanging.
+
+const badChunks = [
+  { name: 'undefined', value: undefined, code: 'ERR_INVALID_ARG_TYPE' },
+  { name: 'null', value: null, code: 'ERR_STREAM_NULL_VALUES' },
+  { name: 'number', value: 3.14, code: 'ERR_INVALID_ARG_TYPE' },
+  { name: 'object', value: {}, code: 'ERR_INVALID_ARG_TYPE' },
+  { name: 'array', value: [65], code: 'ERR_INVALID_ARG_TYPE' },
+  {
+    name: 'SharedArrayBuffer',
+    value: new SharedArrayBuffer(1),
+    code: 'ERR_INVALID_ARG_TYPE',
+  },
+  {
+    name: 'Uint8Array backed by SharedArrayBuffer',
+    value: new Uint8Array(new SharedArrayBuffer(1)),
+    code: 'ERR_INVALID_ARG_TYPE',
+  },
+];
+
+for (const format of ['deflate', 'deflate-raw', 'gzip', 'brotli']) {
+  for (const { name, value, code } of badChunks) {
+    const expected = { name: 'TypeError', code };
+
+    test(`CompressionStream rejects bad chunk (${name}) for ${format}`, async () => {
+      const cs = new CompressionStream(format);
+      const writer = cs.writable.getWriter();
+      const reader = cs.readable.getReader();
+
+      const writePromise = writer.write(value);
+      const readPromise = reader.read();
+
+      await assert.rejects(writePromise, expected);
+      await assert.rejects(readPromise, expected);
+    });
+
+    test(`DecompressionStream rejects bad chunk (${name}) for ${format}`, async () => {
+      const ds = new DecompressionStream(format);
+      const writer = ds.writable.getWriter();
+      const reader = ds.readable.getReader();
+
+      const writePromise = writer.write(value);
+      const readPromise = reader.read();
+
+      await assert.rejects(writePromise, expected);
+      await assert.rejects(readPromise, expected);
+    });
+  }
+}