char/avril

(wip) why'd you have to go and make things so complicaaaaated?

git clone https://git.t4t.associates/char/avril

Charlotte Somsnapshot: embed shared main and worker startup in the CLI688e96e

main
branchesmainall refs →
15.9 KiB335 linesraw
1mod common;
2
3use common::run;
4
5#[test]
6fn stream_transfers_preserve_backpressure_chunks_and_failure_propagation() {
7    let output = run(
8        r#"
9        let canceled;
10        const source = new ReadableStream({
11            start(controller) { controller.enqueue({value: 1}); controller.close(); },
12            cancel(reason) { canceled = reason; },
13        });
14        const movedReadable = structuredClone(source, {transfer: [source]});
15        if (!source.locked || movedReadable === source) throw Error('Readable transfer lock');
16        try { movedReadable.getReader({mode: 'byob'}); throw Error('Transferred stream retained byte mode'); }
17        catch (error) { if (!(error instanceof TypeError)) throw error; }
18        let pulls = 0;
19        const lazySource = new ReadableStream({pull(controller) { pulls++; controller.enqueue(1); }});
20        const lazyClone = structuredClone(lazySource, {transfer: [lazySource]});
21        if (pulls !== 0) throw Error('Readable transfer pulled before demand');
22        await lazyClone.getReader().read();
23        if (pulls === 0) throw Error('Readable transfer did not pull on demand');
24        const chunk = (await movedReadable.getReader().read()).value;
25        if (chunk.value !== 1) throw Error('Readable transfer chunk');
26        const cancelSource = new ReadableStream({cancel(reason) { canceled = reason; }});
27        const movedCancel = structuredClone(cancelSource, {transfer: [cancelSource]});
28        const cancelReason = {cancel: true};
29        await movedCancel.cancel(cancelReason);
30        if (canceled === cancelReason || !canceled.cancel) throw Error('Readable transfer cancel');
31        const cancelFailure = {cancelFailure: true};
32        const failedCancelSource = new ReadableStream({cancel() { throw cancelFailure; }});
33        const failedCancelClone = structuredClone(failedCancelSource, {transfer: [failedCancelSource]});
34        try { await failedCancelClone.cancel(); throw Error('Readable cancel rejection'); }
35        catch (error) { if (error === cancelFailure || !error.cancelFailure) throw error; }
36
37        const written = [];
38        const writable = new WritableStream({write(value) { written.push(value); }});
39
40        const movedWritable = structuredClone(writable, {transfer: [writable]});
41        if (!writable.locked || movedWritable === writable) throw Error('Writable transfer lock');
42        const writer = movedWritable.getWriter();
43        await writer.write({value: 2});
44        await writer.close();
45        if (written.length !== 1 || written[0].value !== 2) throw Error('Writable transfer writes');
46        const closedSource = new WritableStream(), closedWriter = closedSource.getWriter();
47        await closedWriter.close();
48        closedWriter.releaseLock();
49        const closedClone = structuredClone(closedSource, {transfer: [closedSource]});
50        try { await closedClone.getWriter().write(1); throw Error('Closed writable transfer'); }
51        catch (error) { if (!(error instanceof TypeError)) throw error; }
52        const writeReason = {writeFailure: true};
53        const failedWritable = new WritableStream({write() { throw writeReason; }});
54        const movedFailedWritable = structuredClone(failedWritable, {transfer: [failedWritable]});
55        try { await movedFailedWritable.getWriter().write(1); throw Error('Writable transfer error'); }
56        catch (error) { if (error === writeReason || !error.writeFailure) throw error; }
57        let cloneWriteAbort;
58        const uncloneableWriteSource = new WritableStream({abort(reason) { cloneWriteAbort = reason; }});
59        const uncloneableWriteClone = structuredClone(uncloneableWriteSource, {transfer: [uncloneableWriteSource]});
60        try { await uncloneableWriteClone.getWriter().write(() => {}); throw Error('Uncloneable writable chunk'); }
61        catch (error) { if (!(error instanceof DOMException) || error.name !== 'DataCloneError') throw error; }
62        if (!(cloneWriteAbort instanceof DOMException) || cloneWriteAbort.name !== 'DataCloneError') {
63            throw Error('Writable clone failure did not abort the source');
64        }
65
66        const uncommitted = new ReadableStream(), transferBuffer = new ArrayBuffer(1);
67        try {
68            structuredClone({stream: uncommitted, fail() {}}, {transfer: [uncommitted, transferBuffer]});
69            throw Error('Transfer serialization failure');
70        } catch (error) { if (!(error instanceof DOMException) || error.name !== 'DataCloneError') throw error; }
71        if (transferBuffer.detached || uncommitted.locked) throw Error('Uncommitted stream transfer changed source');
72        const recovered = structuredClone(uncommitted, {transfer: [uncommitted]});
73        if (recovered === uncommitted || !uncommitted.locked) throw Error('Failed transfer poisoned later clone');
74
75        const transform = new TransformStream({transform(value, controller) { controller.enqueue(value * 2); }});
76        const movedTransform = structuredClone(transform, {transfer: [transform]});
77        if (Object.hasOwn(movedTransform, 'readable') || Object.hasOwn(movedTransform, 'writable')) {
78            throw Error('Transform transfer replaced public accessors');
79        }
80        const reader = movedTransform.readable.getReader(), transformWriter = movedTransform.writable.getWriter();
81        const firstWrite = transformWriter.write(3);
82        if ((await reader.read()).value !== 6) throw Error('Transform transfer');
83        await firstWrite;
84        reader.releaseLock();
85        transformWriter.releaseLock();
86        const movedAgain = structuredClone(movedTransform, {transfer: [movedTransform]});
87        const secondReader = movedAgain.readable.getReader(), secondWriter = movedAgain.writable.getWriter();
88        const secondWrite = secondWriter.write(4);
89        if ((await secondReader.read()).value !== 8) throw Error('Retransferred transform');
90        await secondWrite;
91        await secondWriter.close();
92        if (!(await secondReader.read()).done) throw Error('Retransferred transform close');
93
94        const reason = {failure: true};
95        const failedSource = new ReadableStream({pull() {throw reason;}});
96        const failed = structuredClone(failedSource, {transfer: [failedSource]});
97        try { await failed.getReader().read(); throw Error('Readable transfer error'); }
98        catch (error) { if (error === reason || error.failure !== true) throw error; }
99        let cloneFailureCancellation;
100        const uncloneableChunkSource = new ReadableStream({
101            pull(controller) { controller.enqueue(() => {}); },
102            cancel(error) { cloneFailureCancellation = error; },
103        });
104        const uncloneableChunkClone = structuredClone(uncloneableChunkSource, {transfer: [uncloneableChunkSource]});
105        try { await uncloneableChunkClone.getReader().read(); throw Error('Uncloneable stream chunk'); }
106        catch (error) { if (!(error instanceof DOMException) || error.name !== 'DataCloneError') throw error; }
107        if (!(cloneFailureCancellation instanceof DOMException) || cloneFailureCancellation.name !== 'DataCloneError') {
108            throw Error('Chunk clone failure did not cancel the source');
109        }
110
111        const byteSource = new ReadableStream({
112            type: 'bytes',
113            start(controller) { controller.enqueue(new Uint8Array([1, 2, 3])); controller.close(); },
114        });
115        const byteClone = structuredClone(byteSource, {transfer: [byteSource]});
116        const byteReader = byteClone.getReader();
117        const byteResult = await byteReader.read();
118        if (byteResult.done || byteResult.value.join() !== '1,2,3') throw Error('Transferred byte stream read');
119        if (!(await byteReader.read()).done) throw Error('Transferred byte stream EOF');
120        console.log('stream transfer');
121        "#,
122        &[],
123    );
124    assert_eq!(output.stdout, b"stream transfer\n");
125}
126
127#[test]
128fn release_and_internal_errors_do_not_create_unhandled_rejections() {
129    let output = run(
130        r#"
131        for (const make of [
132            () => new ReadableStream(),
133            () => new ReadableStream({type:'bytes'}),
134            () => new ReadableStream({start(c) {c.error('source failure');}})
135        ]) {
136            const reader = make().getReader();
137            reader.releaseLock();
138        }
139        const byob = new ReadableStream({type:'bytes'}).getReader({mode:'byob'});
140        byob.releaseLock();
141        for (const make of [
142            () => new WritableStream(),
143            () => new WritableStream({start(c) {c.error('sink failure');}})
144        ]) {
145            const writer = make().getWriter();
146            writer.releaseLock();
147        }
148        await new Promise(resolve => setTimeout(resolve));
149        console.log('handled');
150        "#,
151        &[],
152    );
153    assert_eq!(output.stdout, b"handled\n");
154}
155
156#[test]
157fn pull_and_write_failures_preserve_reasons_and_observable_promises() {
158    let output = run(
159        r#"
160        const reason = {failure:true};
161        async function rejects(promise) {
162            try {await promise; throw Error('Expected rejection');}
163            catch (error) {if (error !== reason) throw error;}
164        }
165        for (const failure of [
166            () => {throw reason;},
167            () => Promise.reject(reason),
168            () => ({then(resolve, reject) {reject(reason);}})
169        ]) {
170            const reader = new ReadableStream({pull:failure}).getReader();
171            await Promise.all([rejects(reader.read()), rejects(reader.closed)]);
172            reader.releaseLock();
173            const writer = new WritableStream({write:failure}).getWriter();
174            await Promise.all([rejects(writer.write(1)), rejects(writer.closed)]);
175            writer.releaseLock();
176        }
177        console.log('reasons');
178        "#,
179        &[],
180    );
181    assert_eq!(output.stdout, b"reasons\n");
182}
183
184#[test]
185fn piping_propagates_source_and_sink_failures_without_leaking_internal_rejections() {
186    let output = run(
187        r#"
188        const reason = {failure:true};
189        let aborted, canceled;
190        const failedSource = new ReadableStream({pull() {throw reason;}});
191        try {
192            await failedSource.pipeTo(new WritableStream({abort(error) {aborted = error;}}));
193            throw Error('Successful failed source');
194        } catch (error) {if (error !== reason) throw error;}
195        if (aborted !== reason || failedSource.locked) throw Error('Source failure propagation');
196        const source = new ReadableStream({
197            pull(c) {c.enqueue(1);},
198            cancel(error) {canceled = error;}
199        });
200        const failedSink = new WritableStream({write() {return Promise.reject(reason);}});
201        try {await source.pipeTo(failedSink); throw Error('Successful failed sink');}
202        catch (error) {if (error !== reason) throw error;}
203        if (canceled !== reason || source.locked || failedSink.locked) throw Error('Sink failure propagation');
204        console.log('piping');
205        "#,
206        &[],
207    );
208    assert_eq!(output.stdout, b"piping\n");
209}
210
211#[test]
212fn public_operation_promises_and_observer_promises_keep_their_identities() {
213    let output = run(
214        r#"
215        const reader = new ReadableStream({start(c) {c.close();}}).getReader();
216        const reads = [reader.read(), reader.read()];
217        if (reads[0] === reads[1]) throw Error('Shared read results');
218        if ((await Promise.all(reads)).some(result => !result.done)) throw Error('Closed reads');
219        const cancels = [reader.cancel(), reader.cancel()];
220        if (cancels[0] === cancels[1]) throw Error('Shared closed cancel results');
221        await Promise.all(cancels);
222        reader.releaseLock();
223
224        const a = new ReadableStream().getReader(), b = new ReadableStream().getReader();
225        if (a.closed !== a.closed || a.closed === b.closed) throw Error('Reader closed identity');
226        const cancellations = [a.cancel(), b.cancel()];
227        if (cancellations[0] === cancellations[1]) throw Error('Shared cancel results');
228        await Promise.all(cancellations);
229        a.releaseLock(); b.releaseLock();
230
231        const writer = new WritableStream().getWriter();
232        const other = new WritableStream().getWriter();
233        if (writer.ready !== writer.ready || writer.ready === other.ready ||
234            writer.closed !== writer.closed || writer.closed === other.closed) {
235            throw Error('Writer observer identity');
236        }
237        const writes = [writer.write(1), writer.write(2)];
238        if (writes[0] === writes[1]) throw Error('Shared write results');
239        await Promise.all(writes);
240        await writer.close();
241        const aborts = [writer.abort(), writer.abort()];
242        if (aborts[0] === aborts[1]) throw Error('Shared closed abort results');
243        await Promise.all(aborts);
244        await other.close();
245        writer.releaseLock(); other.releaseLock();
246        console.log('identity');
247        "#,
248        &[],
249    );
250    assert_eq!(output.stdout, b"identity\n");
251}
252
253#[test]
254fn default_transforms_keep_backpressure_independent_between_streams() {
255    let output = run(
256        r#"
257        const a = new TransformStream(), b = new TransformStream();
258        const ar = a.readable.getReader(), br = b.readable.getReader();
259        const aw = a.writable.getWriter(), bw = b.writable.getWriter();
260        let aWritten = false, bWritten = false, bReady = false;
261        const aWrite = aw.write('a').then(() => {aWritten = true;});
262        const bWrite = bw.write('b').then(() => {bWritten = true;});
263        const aReady = aw.ready, bReadyPromise = bw.ready;
264        if (aReady === bReadyPromise) throw Error('Shared backpressure promise');
265        bReadyPromise.then(() => {bReady = true;});
266        await new Promise(resolve => setTimeout(resolve));
267        if (aWritten || bWritten || bReady) throw Error('Missing initial backpressure');
268        if ((await ar.read()).value !== 'a') throw Error('First transform');
269        await aWrite; await aReady;
270        if (!aWritten || bWritten || bReady) throw Error('Cross-stream backpressure');
271        if ((await br.read()).value !== 'b') throw Error('Second transform');
272        await bWrite; await bReadyPromise;
273        await Promise.all([aw.close(), bw.close()]);
274        if (!(await ar.read()).done || !(await br.read()).done) throw Error('Transform closure');
275        ar.releaseLock(); br.releaseLock(); aw.releaseLock(); bw.releaseLock();
276        console.log('backpressure');
277        "#,
278        &[],
279    );
280    assert_eq!(output.stdout, b"backpressure\n");
281}
282
283#[test]
284fn user_callback_thenables_are_assimilated_once_per_invocation() {
285    let output = run(
286        r#"
287        let accesses = 0, calls = 0;
288        const writer = new WritableStream({
289            write(value) {
290                calls++;
291                return {get then() {
292                    accesses++;
293                    return resolve => queueMicrotask(resolve);
294                }};
295            }
296        }).getWriter();
297        await Promise.all([writer.write(1), writer.write(2), writer.write(3)]);
298        await writer.close();
299        if (calls !== 3 || accesses !== 3) throw Error('Callback assimilation');
300        writer.releaseLock();
301        console.log('thenables');
302        "#,
303        &[],
304    );
305    assert_eq!(output.stdout, b"thenables\n");
306}
307
308#[test]
309fn internal_handling_does_not_suppress_unobserved_public_write_failures() {
310    let (_dir, child) = common::spawn(
311        "new WritableStream({write() {throw Error('public write failure');}}).getWriter().write(1);",
312    );
313    let output = common::wait(child);
314    assert!(!output.status.success());
315    assert!(String::from_utf8_lossy(&output.stderr).contains("public write failure"));
316}
317
318#[test]
319fn transferred_writable_ready_resolves_before_the_first_write() {
320    let out = run(
321        r#"
322        let value;
323        const source=new WritableStream({write(chunk){value=chunk;}});
324        const moved=structuredClone(source,{transfer:[source]});
325        const writer=moved.getWriter();
326        await writer.ready;
327        await writer.write('ready');
328        await writer.close();
329        if(value!=='ready')throw Error('Writable transfer readiness');
330        console.log('transfer ready');
331    "#,
332        &[],
333    );
334    assert_eq!(out.stdout, b"transfer ready\n");
335}