GV

GeezerVison

The World Wide Web, restored to 800 pixels of glory.

Home|About|Gaming cabinet|Archives|How it works|Starter sites

On this site

  • Sign in
  • ActionsCI/CD and automation
  • CodespacesCloud development environments
  • All Features
  • Documentation
  • Training
  • Certifications
  • Build with Copilot
  • Customize Copilot
  • Integrations
  • Trending repos
  • GitHub Sponsors
  • Developer articles
  • Enterprises

This view

  • Original page
  • New address

GeezerVison is showing github.com — stream: improve Web Compression spec compliance · nodejs/node@4bdcaf2

File tree

Expand file treeCollapse file tree
Open diff view settings
Filter options
  • lib/internal/webstreams
    • adapters.js
    • compression.js
  • test
    • parallel
      • test-webstreams-adapters-sync-write-error.js
      • test-webstreams-compression-bad-chunks.js
      • test-webstreams-decompression-reject-trailing.js
    • wpt/status
      • compression.json
Expand file treeCollapse file tree
Open diff view settings
Collapse file

‎lib/internal/webstreams/adapters.js‎

Copy file name to clipboardExpand all lines: lib/internal/webstreams/adapters.js
+36-8Lines changed: 36 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,8 @@ const {
1111
SafePromiseAll,
1212
SafePromisePrototypeFinally,
1313
SafeSet,
14+
StringPrototypeStartsWith,
15+
Symbol,
1416
TypeError,
1517
TypedArrayPrototypeGetBuffer,
1618
TypedArrayPrototypeGetByteLength,
@@ -94,6 +96,9 @@ const { UV_EOF } = internalBinding('uv');
9496

9597
const encoder = new TextEncoder();
9698

99+
const kValidateChunk = Symbol('kValidateChunk');
100+
const kDestroyOnSyncError = Symbol('kDestroyOnSyncError');
101+
97102
// Collect all negative (error) ZLIB codes and Z_NEED_DICT
98103
const ZLIB_FAILURES = new SafeSet([
99104
...ArrayPrototypeFilter(
@@ -139,9 +144,10 @@ function handleKnownInternalErrors(cause) {
139144

140145
/**
141146
* @param {Writable} streamWritable
147+
* @param {object} [options]
142148
* @returns {WritableStream}
143149
*/
144-
function newWritableStreamFromStreamWritable(streamWritable) {
150+
function newWritableStreamFromStreamWritable(streamWritable, options = kEmptyObject) {
145151
// Not using the internal/streams/utils isWritableNodeStream utility
146152
// here because it will return false if streamWritable is a Duplex
147153
// whose writable option is false. For a Duplex that is not writable,
@@ -220,12 +226,26 @@ function newWritableStreamFromStreamWritable(streamWritable) {
220226
if (!streamWritable.writableObjectMode && isArrayBuffer(chunk)) {
221227
chunk = new Uint8Array(chunk);
222228
}
223-
if (streamWritable.writableNeedDrain || !streamWritable.write(chunk)) {
224-
backpressurePromise = PromiseWithResolvers();
225-
return SafePromisePrototypeFinally(
226-
backpressurePromise.promise, () => {
227-
backpressurePromise = undefined;
228-
});
229+
try {
230+
options[kValidateChunk]?.(chunk);
231+
if (streamWritable.writableNeedDrain || !streamWritable.write(chunk)) {
232+
backpressurePromise = PromiseWithResolvers();
233+
return SafePromisePrototypeFinally(
234+
backpressurePromise.promise, () => {
235+
backpressurePromise = undefined;
236+
});
237+
}
238+
} catch (error) {
239+
// When the kDestroyOnSyncError flag is set (e.g. for
240+
// CompressionStream), a sync throw must also destroy the
241+
// stream so the readable side is errored too. Without this
242+
// the readable side hangs forever. This replicates the
243+
// TransformStream semantics: error both sides on any throw
244+
// in the transform path.
245+
if (options[kDestroyOnSyncError]) {
246+
destroy(streamWritable, error);
247+
}
248+
throw error;
229249
}
230250
},
231251

@@ -662,9 +682,15 @@ function newReadableWritablePairFromDuplex(duplex, options = kEmptyObject) {
662682
return { readable, writable };
663683
}
664684

685+
const writableOptions = {
686+
__proto__: null,
687+
[kValidateChunk]: options[kValidateChunk],
688+
[kDestroyOnSyncError]: options[kDestroyOnSyncError],
689+
};
690+
665691
const writable =
666692
isWritable(duplex) ?
667-
newWritableStreamFromStreamWritable(duplex) :
693+
newWritableStreamFromStreamWritable(duplex, writableOptions) :
668694
new WritableStream();
669695

670696
if (!isWritable(duplex))
@@ -1064,4 +1090,6 @@ module.exports = {
10641090
newStreamDuplexFromReadableWritablePair,
10651091
newWritableStreamFromStreamBase,
10661092
newReadableStreamFromStreamBase,
1093+
kValidateChunk,
1094+
kDestroyOnSyncError,
10671095
};
Collapse file

‎lib/internal/webstreams/compression.js‎

Copy file name to clipboardExpand all lines: lib/internal/webstreams/compression.js
+38-11Lines changed: 38 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -7,15 +7,28 @@ const {
77

88
const {
99
newReadableWritablePairFromDuplex,
10+
kValidateChunk,
11+
kDestroyOnSyncError,
1012
} = require('internal/webstreams/adapters');
1113

1214
const { customInspect } = require('internal/webstreams/util');
1315

16+
const {
17+
isArrayBufferView,
18+
isSharedArrayBuffer,
19+
} = require('internal/util/types');
20+
1421
const {
1522
customInspectSymbol: kInspect,
1623
kEnumerableProperty,
1724
} = require('internal/util');
1825

26+
const {
27+
codes: {
28+
ERR_INVALID_ARG_TYPE,
29+
},
30+
} = require('internal/errors');
31+
1932
const { createEnumConverter } = require('internal/webidl');
2033

2134
let zlib;
@@ -24,6 +37,18 @@ function lazyZlib() {
2437
return zlib;
2538
}
2639

40+
// Per the Compression Streams spec, chunks must be BufferSource
41+
// (ArrayBuffer or ArrayBufferView not backed by SharedArrayBuffer).
42+
function validateBufferSourceChunk(chunk) {
43+
if (isArrayBufferView(chunk) && isSharedArrayBuffer(chunk.buffer)) {
44+
throw new ERR_INVALID_ARG_TYPE(
45+
'chunk',
46+
['ArrayBuffer', 'Buffer', 'TypedArray', 'DataView'],
47+
chunk,
48+
);
49+
}
50+
}
51+
2752
const formatConverter = createEnumConverter('CompressionFormat', [
2853
'deflate',
2954
'deflate-raw',
@@ -62,7 +87,10 @@ class CompressionStream {
6287
this.#handle = lazyZlib().createBrotliCompress();
6388
break;
6489
}
65-
this.#transform = newReadableWritablePairFromDuplex(this.#handle);
90+
this.#transform = newReadableWritablePairFromDuplex(this.#handle, {
91+
[kValidateChunk]: validateBufferSourceChunk,
92+
[kDestroyOnSyncError]: true,
93+
});
6694
}
6795

6896
/**
@@ -108,25 +136,24 @@ class DecompressionStream {
108136
});
109137
break;
110138
case 'deflate-raw':
111-
this.#handle = lazyZlib().createInflateRaw();
139+
this.#handle = lazyZlib().createInflateRaw({
140+
rejectGarbageAfterEnd: true,
141+
});
112142
break;
113143
case 'gzip':
114144
this.#handle = lazyZlib().createGunzip({
115145
rejectGarbageAfterEnd: true,
116146
});
117147
break;
118148
case 'brotli':
119-
this.#handle = lazyZlib().createBrotliDecompress();
149+
this.#handle = lazyZlib().createBrotliDecompress({
150+
rejectGarbageAfterEnd: true,
151+
});
120152
break;
121153
}
122-
this.#transform = newReadableWritablePairFromDuplex(this.#handle);
123-
124-
this.#handle.on('error', (err) => {
125-
if (this.#transform?.writable &&
126-
!this.#transform.writable.locked &&
127-
typeof this.#transform.writable.abort === 'function') {
128-
this.#transform.writable.abort(err);
129-
}
154+
this.#transform = newReadableWritablePairFromDuplex(this.#handle, {
155+
[kValidateChunk]: validateBufferSourceChunk,
156+
[kDestroyOnSyncError]: true,
130157
});
131158
}
132159

Collapse file

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

Copy file name to clipboard
+79Lines changed: 79 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,79 @@
1+
'use strict';
2+
// Flags: --no-warnings --expose-internals
3+
require('../common');
4+
const assert = require('assert');
5+
const test = require('node:test');
6+
const { Duplex, Writable } = require('stream');
7+
const {
8+
newWritableStreamFromStreamWritable,
9+
newReadableWritablePairFromDuplex,
10+
} = require('internal/webstreams/adapters');
11+
12+
// Verify that when the underlying Node.js stream throws synchronously from
13+
// write(), the writable web stream properly rejects but does not destroy
14+
// the stream (destroy-on-sync-throw is only used internally by
15+
// CompressionStream/DecompressionStream).
16+
17+
test('WritableStream from Node.js stream handles sync write throw', async () => {
18+
const error = new TypeError('invalid chunk');
19+
const writable = new Writable({
20+
write() {
21+
throw error;
22+
},
23+
});
24+
25+
const ws = newWritableStreamFromStreamWritable(writable);
26+
const writer = ws.getWriter();
27+
28+
await assert.rejects(writer.write('bad'), (err) => {
29+
assert.strictEqual(err, error);
30+
return true;
31+
});
32+
33+
// Standalone writable should not be destroyed on sync write error
34+
assert.strictEqual(writable.destroyed, false);
35+
});
36+
37+
test('Duplex-backed pair does NOT destroy on sync write throw', async () => {
38+
const error = new TypeError('invalid chunk');
39+
const duplex = new Duplex({
40+
read() {},
41+
write() {
42+
throw error;
43+
},
44+
});
45+
46+
const { writable, readable } = newReadableWritablePairFromDuplex(duplex);
47+
const writer = writable.getWriter();
48+
49+
await assert.rejects(writer.write('bad'), (err) => {
50+
assert.strictEqual(err, error);
51+
return true;
52+
});
53+
54+
// A plain Duplex should NOT be destroyed on sync write error
55+
assert.strictEqual(duplex.destroyed, false);
56+
57+
// The readable side should still be usable
58+
const reader = readable.getReader();
59+
reader.cancel();
60+
});
61+
62+
test('WritableStream from Node.js stream - valid writes still work', async () => {
63+
const chunks = [];
64+
const writable = new Writable({
65+
write(chunk, _encoding, cb) {
66+
chunks.push(chunk);
67+
cb();
68+
},
69+
});
70+
71+
const ws = newWritableStreamFromStreamWritable(writable);
72+
const writer = ws.getWriter();
73+
74+
await writer.write(Buffer.from('hello'));
75+
await writer.write(Buffer.from(' world'));
76+
await writer.close();
77+
78+
assert.strictEqual(Buffer.concat(chunks).toString(), 'hello world');
79+
});
Collapse file

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

Copy file name to clipboard
+57Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,57 @@
1+
'use strict';
2+
require('../common');
3+
const assert = require('assert');
4+
const test = require('node:test');
5+
const { CompressionStream, DecompressionStream } = require('stream/web');
6+
7+
// Verify that writing invalid (non-BufferSource) chunks to
8+
// CompressionStream and DecompressionStream properly rejects
9+
// on both the write and the read side, instead of hanging.
10+
11+
const badChunks = [
12+
{ name: 'undefined', value: undefined, code: 'ERR_INVALID_ARG_TYPE' },
13+
{ name: 'null', value: null, code: 'ERR_STREAM_NULL_VALUES' },
14+
{ name: 'number', value: 3.14, code: 'ERR_INVALID_ARG_TYPE' },
15+
{ name: 'object', value: {}, code: 'ERR_INVALID_ARG_TYPE' },
16+
{ name: 'array', value: [65], code: 'ERR_INVALID_ARG_TYPE' },
17+
{
18+
name: 'SharedArrayBuffer',
19+
value: new SharedArrayBuffer(1),
20+
code: 'ERR_INVALID_ARG_TYPE',
21+
},
22+
{
23+
name: 'Uint8Array backed by SharedArrayBuffer',
24+
value: new Uint8Array(new SharedArrayBuffer(1)),
25+
code: 'ERR_INVALID_ARG_TYPE',
26+
},
27+
];
28+
29+
for (const format of ['deflate', 'deflate-raw', 'gzip', 'brotli']) {
30+
for (const { name, value, code } of badChunks) {
31+
const expected = { name: 'TypeError', code };
32+
33+
test(`CompressionStream rejects bad chunk (${name}) for ${format}`, async () => {
34+
const cs = new CompressionStream(format);
35+
const writer = cs.writable.getWriter();
36+
const reader = cs.readable.getReader();
37+
38+
const writePromise = writer.write(value);
39+
const readPromise = reader.read();
40+
41+
await assert.rejects(writePromise, expected);
42+
await assert.rejects(readPromise, expected);
43+
});
44+
45+
test(`DecompressionStream rejects bad chunk (${name}) for ${format}`, async () => {
46+
const ds = new DecompressionStream(format);
47+
const writer = ds.writable.getWriter();
48+
const reader = ds.readable.getReader();
49+
50+
const writePromise = writer.write(value);
51+
const readPromise = reader.read();
52+
53+
await assert.rejects(writePromise, expected);
54+
await assert.rejects(readPromise, expected);
55+
});
56+
}
57+
}

0 commit comments

Comments
0 (0)
You are visitor 01402503 — GeezerVison does not store the pages you look up.
Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required.