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,8 @@ const { UV_EOF } = internalBinding('uv');
 
 const encoder = new TextEncoder();
 
+const kValidateChunk = Symbol('kValidateChunk');
+
 // Collect all negative (error) ZLIB codes and Z_NEED_DICT
 const ZLIB_FAILURES = new SafeSet([
   ...ArrayPrototypeFilter(
@@ -139,9 +143,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 +225,25 @@ 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;
-          });
+      // If the underlying Node.js stream throws synchronously from
+      // write() (e.g. zlib rejects an invalid chunk type), we must
+      // destroy the stream so that the readable side is also errored.
+      // Without this, the readable side would hang forever waiting
+      // for data that will never arrive. The same applies to any
+      // caller-provided chunk validation (e.g. web compression
+      // SharedArrayBuffer rejection).
+      try {
+        options[kValidateChunk]?.(chunk);
+        if (streamWritable.writableNeedDrain || !streamWritable.write(chunk)) {
+          backpressurePromise = PromiseWithResolvers();
+          return SafePromisePrototypeFinally(
+            backpressurePromise.promise, () => {
+              backpressurePromise = undefined;
+            });
+        }
+      } catch (error) {
+        destroy(streamWritable, error);
+        throw error;
       }
     },
 
@@ -662,9 +680,14 @@ function newReadableWritablePairFromDuplex(duplex, options = kEmptyObject) {
     return { readable, writable };
   }
 
+  const writableOptions = {
+    __proto__: null,
+    [kValidateChunk]: options[kValidateChunk],
+  };
+
   const writable =
     isWritable(duplex) ?
-      newWritableStreamFromStreamWritable(duplex) :
+      newWritableStreamFromStreamWritable(duplex, writableOptions) :
       new WritableStream();
 
   if (!isWritable(duplex))
@@ -1064,4 +1087,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,52 @@
+'use strict';
+// Flags: --no-warnings --expose-internals
+require('../common');
+const assert = require('assert');
+const test = require('node:test');
+const { Writable } = require('stream');
+const {
+  newWritableStreamFromStreamWritable,
+} = require('internal/webstreams/adapters');
+
+// Verify that when the underlying Node.js stream throws synchronously from
+// write(), the writable web stream properly rejects and the underlying
+// stream is destroyed.
+
+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;
+  });
+
+  // The underlying stream must be destroyed
+  assert.strictEqual(writable.destroyed, true);
+});
+
+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);
+    });
+  }
+}

test/parallel/test-webstreams-decompression-reject-trailing.js

@@ -0,0 +1,86 @@
+'use strict';
+require('../common');
+const assert = require('assert');
+const test = require('node:test');
+const { CompressionStream, DecompressionStream } = require('stream/web');
+
+// Verify that DecompressionStream rejects trailing data after a valid
+// compressed payload for all four supported formats (deflate, deflate-raw, gzip, brotli).
+
+const input = Buffer.from('hello');
+const trailingJunk = Buffer.from([0xDE, 0xAD]);
+
+async function compress(format, data) {
+  const cs = new CompressionStream(format);
+  const writer = cs.writable.getWriter();
+  writer.write(data);
+  writer.close();
+  const chunks = await Array.fromAsync(cs.readable);
+  return Buffer.concat(chunks.map((c) => Buffer.from(c)));
+}
+
+for (const format of ['deflate', 'deflate-raw', 'gzip', 'brotli']) {
+  test(`DecompressionStream rejects trailing junk for ${format}`, async () => {
+    const compressed = await compress(format, input);
+    const withJunk = Buffer.concat([compressed, trailingJunk]);
+
+    const ds = new DecompressionStream(format);
+    const writer = ds.writable.getWriter();
+    const reader = ds.readable.getReader();
+
+    writer.write(withJunk).catch(() => {});
+    writer.close().catch(() => {});
+
+    await assert.rejects(async () => {
+      const chunks = [];
+      while (true) {
+        const { done, value } = await reader.read();
+        if (done) break;
+        chunks.push(value);
+      }
+    }, (err) => {
+      assert(err instanceof Error, `Expected Error, got ${err?.constructor?.name}`);
+      return true;
+    });
+  });
+
+  test(`DecompressionStream accepts valid ${format} data without trailing junk`, async () => {
+    const compressed = await compress(format, input);
+
+    const ds = new DecompressionStream(format);
+    const writer = ds.writable.getWriter();
+
+    writer.write(compressed);
+    writer.close();
+
+    const chunks = await Array.fromAsync(ds.readable);
+    const result = Buffer.concat(chunks.map((c) => Buffer.from(c)));
+    assert.strictEqual(result.toString(), 'hello');
+  });
+}
+
+// Extra: Verify that trailing data is also rejected when passed as a separate
+// chunk after the valid compressed data has been fully written.
+for (const format of ['deflate', 'deflate-raw', 'gzip', 'brotli']) {
+  test(`DecompressionStream rejects trailing junk as separate chunk for ${format}`, async () => {
+    const compressed = await compress(format, input);
+
+    const ds = new DecompressionStream(format);
+    const writer = ds.writable.getWriter();
+    const reader = ds.readable.getReader();
+
+    writer.write(compressed).catch(() => {});
+    writer.write(trailingJunk).catch(() => {});
+    writer.close().catch(() => {});
+
+    await assert.rejects(async () => {
+      while (true) {
+        const { done } = await reader.read();
+        if (done) break;
+      }
+    }, (err) => {
+      assert(err instanceof Error, `Expected Error, got ${err?.constructor?.name}`);
+      return true;
+    });
+  });
+}

test/wpt/status/compression.json

@@ -1,9 +1,11 @@
 {
-  "compression-bad-chunks.any.js": {
-    "skip": "Execution \"hangs\", ArrayBuffer and TypedArray is not accepted and throws, instead of rejects during writer.write"
-  },
   "decompression-bad-chunks.any.js": {
-    "skip": "Execution \"hangs\", ArrayBuffer and TypedArray is not accepted and throws, instead of rejects during writer.write"
+    "fail": {
+      "expected": [
+        "chunk of type invalid deflate bytes should error the stream for brotli",
+        "chunk of type invalid gzip bytes should error the stream for brotli"
+      ]
+    }
   },
   "compression-with-detach.window.js": {
     "requires": ["crypto"]