FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

stream: improve Web Compression spec compliance · nodejs/node@4bdcaf2 · GitHub

/ node Public

Commit 4bdcaf2

Browse files
panva
authored and
aduh95
committed
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. PR-URL: #62107 Reviewed-By: Yagiz Nizipli <yagiz@nizipli.com> Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: Mattias Buelens <mattias@buelens.com> Reviewed-By: René <contact.9a5d6388@renegade334.me.uk>
1 parent 82770cb commit 4bdcaf2

6 files changed

Lines changed: 302 additions & 23 deletions

File tree

‎lib/internal/webstreams/adapters.js‎

Lines 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
};

‎lib/internal/webstreams/compression.js‎

Lines 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

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

Lines 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+
});

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

Lines 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)
Back | FazBrowse Home | New Git URL