From 40beacd4e85865700fa8b109efe826abbe5e1a28 Mon Sep 17 00:00:00 2001 From: James M Snell Date: Wed, 28 Oct 2026 00:11:36 +0000 Subject: [PATCH] Encode string chunks as UTF-8 when pumping TypeScript streams Under typescript_implemented_streams, the C++ bridge's pump for a queued (JS-sourced) stream rejected string chunks with "This ReadableStream did not return bytes." The C++ implementation's pump has always written them as their UTF-8 encoding (valueToBytes() in queue.c++), and code such as server-sent-event handlers that enqueue strings without a TextEncoder relies on it. Every sink that sends a body on goes through this pump: a fetch() request body, iterable request bodies included, a Response the fetch handler returns, cache.put(), KV put(), HTMLRewriter's input and streamed replacement content, EventSource.from() and JS RPC. As the request body pump runs apart from the response, its failure did not reject fetch(): the request went out with its body cut short at the batch holding the string, and only the source's cancel() saw the error. Elsewhere the client got an empty reply, or an RPC receiver a premature disconnect. The pump now writes a string chunk with toUSVString(), exactly as valueToBytes() does: lone surrogates become U+FFFD and embedded NULs are kept. Other non-byte chunks still reject. Reading a body in JavaScript (text() and the like) still rejects strings in both implementations, as the spec has it. Tests pin the parity in each sink's suite: the readable suite (request, iterable and returned response bodies), htmlrewriter, cache, the cross-implementation RPC test, and EventSource, which gains a TypeScript-streams config. The readable suite's main cells pin fetch_iterable_type_support for the iterable body. The bridge unit test that pinned the rejection with a string now uses a number. typescript_implemented_streams is still experimental, with no enable date, so the change needs no compatibility flag. Fixes: https://github.com/cloudflare/workerd/issues/8041 --- src/tests/streams/cache/AGENTS.md | 1 + src/tests/streams/cache/cache-streams.js | 23 +++++ src/tests/streams/cache/main.js | 1 + src/tests/streams/htmlrewriter/AGENTS.md | 1 + src/tests/streams/htmlrewriter/main.js | 1 + .../streams/htmlrewriter/rewrite-streams.js | 35 +++++++ src/tests/streams/readable/AGENTS.md | 2 + src/tests/streams/readable/main.js | 13 ++- .../readable/readable-cpp-pedantic.wd-test | 1 + .../streams/readable/readable-cpp.wd-test | 3 + .../streams/readable/readable-modules.capnp | 1 + .../streams/readable/readable-ts.wd-test | 1 + src/tests/streams/readable/string-chunks.js | 92 +++++++++++++++++++ src/workerd/api/js-readable-stream-test.c++ | 31 ++++++- src/workerd/api/js-readable-stream.c++ | 9 +- src/workerd/api/tests/BUILD.bazel | 6 ++ src/workerd/api/tests/eventsource-test.js | 25 +++++ .../api/tests/eventsource-ts-test.wd-test | 20 ++++ .../tests/js-rpc-streams-crossflag-test.js | 29 ++++++ 19 files changed, 290 insertions(+), 5 deletions(-) create mode 100644 src/tests/streams/readable/string-chunks.js create mode 100644 src/workerd/api/tests/eventsource-ts-test.wd-test diff --git a/src/tests/streams/cache/AGENTS.md b/src/tests/streams/cache/AGENTS.md index 5b94adf883a..b88b222134b 100644 --- a/src/tests/streams/cache/AGENTS.md +++ b/src/tests/streams/cache/AGENTS.md @@ -21,6 +21,7 @@ the record back through its MOCK service binding (/last-put). | Test | Shape | | --- | --- | | `putJsValueStreamBody` | value stream of Uint8Array chunks, byte-exact at the backend | +| `putStringChunkBody` | a string chunk stored as its UTF-8 encoding beside a byte chunk, byte-exact at the backend | | `putJsByteStreamBody` | 64 KiB chunked byte stream | | `putFixedLengthStreamBody` | FixedLengthStream body; the declared length arrives as a concrete Content-Length in the serialized head | | `putIdentityStreamBody` | identity body fed by a concurrent writer | diff --git a/src/tests/streams/cache/cache-streams.js b/src/tests/streams/cache/cache-streams.js index 44b1bb8c4f1..f7c1f6c6a70 100644 --- a/src/tests/streams/cache/cache-streams.js +++ b/src/tests/streams/cache/cache-streams.js @@ -57,6 +57,29 @@ export const putJsValueStreamBody = { }, }; +// A string chunk is stored as its UTF-8 encoding beside the byte chunk +// after it: the pattern's first 100 bytes are ASCII, so the string of +// those characters encodes to exactly them. +export const putStringChunkBody = { + async test(ctrl, env) { + const ascii = String.fromCharCode(...patternChunk(0, 100)); + const rs = new ReadableStream({ + start(c) { + c.enqueue(ascii); + c.enqueue(patternChunk(100, 100)); + c.close(); + }, + }); + await caches.default.put( + 'https://example.com/string-chunk', + new Response(rs) + ); + const record = await lastPut(env); + strictEqual(record.byteLength, 200); + strictEqual(record.patternOk, true); + }, +}; + // A JS byte stream, chunked. export const putJsByteStreamBody = { async test(ctrl, env) { diff --git a/src/tests/streams/cache/main.js b/src/tests/streams/cache/main.js index 3a2e979efef..9f28de03462 100644 --- a/src/tests/streams/cache/main.js +++ b/src/tests/streams/cache/main.js @@ -7,6 +7,7 @@ export { putJsValueStreamBody, + putStringChunkBody, putJsByteStreamBody, putFixedLengthStreamBody, putIdentityStreamBody, diff --git a/src/tests/streams/htmlrewriter/AGENTS.md b/src/tests/streams/htmlrewriter/AGENTS.md index c35437e1acc..a9d3b8a9d74 100644 --- a/src/tests/streams/htmlrewriter/AGENTS.md +++ b/src/tests/streams/htmlrewriter/AGENTS.md @@ -19,6 +19,7 @@ under the MODERN `fixup-transform-stream-backpressure`. | `handlerAcrossChunkBoundaries` | chunks split MID-TAG; the parser reassembles and the handler mutates both elements | | `rewrittenBodyIsReadableStream` | output body drained incrementally via a reader | | `contentFromReadableStream` | element.replace(ReadableStream) — streamed replacement content | +| `stringChunksAreUtf8` | string chunks in the transformed body and in streamed replacement content read as UTF-8 beside byte chunks (parity) | | `identityStreamBody` | identity body fed by a concurrent writer | | `cancelReachesSourceAfterNextChunk` | cancel remains pending at the source until another chunk wakes the pump, then reaches the source | | `erroringSourceRejectsConsumption` | source error surfaces from .text() | diff --git a/src/tests/streams/htmlrewriter/main.js b/src/tests/streams/htmlrewriter/main.js index ef092450df4..eade15936d8 100644 --- a/src/tests/streams/htmlrewriter/main.js +++ b/src/tests/streams/htmlrewriter/main.js @@ -11,6 +11,7 @@ export { handlerAcrossChunkBoundaries, rewrittenBodyIsReadableStream, contentFromReadableStream, + stringChunksAreUtf8, identityStreamBody, cancelReachesSourceAfterNextChunk, erroringSourceRejectsConsumption, diff --git a/src/tests/streams/htmlrewriter/rewrite-streams.js b/src/tests/streams/htmlrewriter/rewrite-streams.js index 5a90aa891df..431e58ba4d1 100644 --- a/src/tests/streams/htmlrewriter/rewrite-streams.js +++ b/src/tests/streams/htmlrewriter/rewrite-streams.js @@ -125,6 +125,41 @@ export const contentFromReadableStream = { }, }; +// String chunks, in the transformed body and in streamed replacement +// content, are read as their UTF-8 encoding beside the byte chunks +// around them (parity). +export const stringChunksAreUtf8 = { + async test() { + const mixed = (first, second) => + new ReadableStream({ + start(c) { + c.enqueue(enc.encode(first)); + c.enqueue(second); + c.close(); + }, + }); + + strictEqual( + await new HTMLRewriter() + .transform(new Response(mixed('

caf', '\u00e9

'))) + .text(), + '

caf\u00e9

' + ); + + strictEqual( + await new HTMLRewriter() + .on('div', { + element(element) { + element.replace(mixed('new', ''), { html: true }); + }, + }) + .transform(new Response('
old
')) + .text(), + 'new' + ); + }, +}; + // An IdentityTransformStream body fed by a concurrent writer. export const identityStreamBody = { async test() { diff --git a/src/tests/streams/readable/AGENTS.md b/src/tests/streams/readable/AGENTS.md index 5a6eda0a3b9..b961d94883a 100644 --- a/src/tests/streams/readable/AGENTS.md +++ b/src/tests/streams/readable/AGENTS.md @@ -106,6 +106,7 @@ C++ implementation; `draining-reader.js` asserts both sides. | `streams_enable_constructors` (2022-11-30) | yes (the suite's subject) | `readable-cpp-legacy` cell: ctor and .from() throw plain Errors naming the flag; ReadableStreamDefaultController not exposed; native-backed streams (Response bodies) fully usable incl. tee() | | `transformstream_enable_standard_constructor` | yes (tee-reentrancy drives tee through TransformStreams) | transform suite's legacy cells | | `capture_async_api_throws`, `workers_api_getters_setters_on_prototype`, `set_tostring_tag`, `fixup-transform-stream-backpressure`, `writable_stream_spec_compliant_writer` | yes | other suites' legacy cells | +| `fetch_iterable_type_support` (2026-02-19) | yes (string-chunks.js sends an async generator as a request body) | not exercised here: iterable bodies are fetch's surface, not this suite's | | `pedantic_wpt` (dateless opt-in) | `readable-cpp-pedantic` cell | zero observable deltas on this suite's probed surface (cpp == pedantic in every probe) | ## Module map @@ -138,6 +139,7 @@ C++ implementation; `draining-reader.js` asserts both sides. | `data-volumes.js` | value-stream volume axes: 4096-chunk counts, a 1 MiB single string chunk, 8 MiB aggregate (128 × 64 KiB), and 1 MiB through tee on both branches — every chunk index-encoded | | `pollution.js` | prototype pollution neither implementation observes: a patched array iterator or replaced Number global leaves bodies intact; a patched controller error() still errors the stream; omitted dictionaries read nothing from Object.prototype, nor does from() for the source and strategy it builds; @@asyncIterator shape (TS: the values() function object, per WebIDL; C++: a separate function) and for-await ignoring a patched values() | | `unhandled-rejection.js` | a reader's closed promise is marked handled where it is made (on a readable stream, on an errored stream; replaced at release from a closed stream): rejected with nothing looking at it, it reports no unhandledrejection (parity) | +| `string-chunks.js` | string chunks in a body the runtime sends on (a fetch request body, in one batch and one chunk per pull; an async-generator request body; a response the fetch handler returns, served by SELF at `/strings`) go out as their UTF-8 encoding beside the byte chunks, lone surrogates replaced with U+FFFD and embedded NULs kept (parity; a body read in JavaScript rejects them, `integration-body.js`) | Consumed sources: streams-async-iterator-test.js (deleted), streams-tee-edge-cases-test.js (value half), streams-test.js (from/ diff --git a/src/tests/streams/readable/main.js b/src/tests/streams/readable/main.js index eb6c46b3592..0fc83fec7f0 100644 --- a/src/tests/streams/readable/main.js +++ b/src/tests/streams/readable/main.js @@ -9,9 +9,11 @@ // tests: it echoes the request body back, letting tests round-trip a // ReadableStream through the fetch machinery in both directions, and // reports the Content-Length the request carried. /gzip and /short serve -// expected-length.js's compressed and short responses instead. +// expected-length.js's compressed and short responses, and /strings +// string-chunks.js's response, instead. import { compressedResponse, shortResponse } from 'expected-length'; +import { stringsResponse } from 'string-chunks'; export default { async fetch(request) { @@ -22,6 +24,9 @@ export default { if (url.pathname === '/short') { return shortResponse(url); } + if (url.pathname === '/strings') { + return stringsResponse(); + } return new Response(request.body, { headers: { 'content-type': 'application/octet-stream', @@ -307,3 +312,9 @@ export { } from 'pollution'; export { readerClosedIsMarkedHandled } from 'unhandled-rejection'; + +export { + stringChunksInRequestBody, + stringChunksInIterableRequestBody, + stringChunksInReturnedResponse, +} from 'string-chunks'; diff --git a/src/tests/streams/readable/readable-cpp-pedantic.wd-test b/src/tests/streams/readable/readable-cpp-pedantic.wd-test index 628f77cbcdb..a8958362906 100644 --- a/src/tests/streams/readable/readable-cpp-pedantic.wd-test +++ b/src/tests/streams/readable/readable-cpp-pedantic.wd-test @@ -21,6 +21,7 @@ const unitTests :Workerd.Config = ( "set_tostring_tag", "writable_stream_spec_compliant_writer", "pedantic_wpt", + "fetch_iterable_type_support", "enable_weak_ref", ], bindings = [ diff --git a/src/tests/streams/readable/readable-cpp.wd-test b/src/tests/streams/readable/readable-cpp.wd-test index f9ad9936abf..3007b71e638 100644 --- a/src/tests/streams/readable/readable-cpp.wd-test +++ b/src/tests/streams/readable/readable-cpp.wd-test @@ -20,6 +20,8 @@ using ReadableModules = import "readable-modules.capnp"; # - set_tostring_tag: Symbol.toStringTag observed by api-surface tests. # - writable_stream_spec_compliant_writer: writer behavior in the # integration tests' pipe interactions. +# - fetch_iterable_type_support: string-chunks.js sends an async +# generator as a request body. const unitTests :Workerd.Config = ( services = [ ( name = "streams-readable-cpp", @@ -34,6 +36,7 @@ const unitTests :Workerd.Config = ( "workers_api_getters_setters_on_prototype", "set_tostring_tag", "writable_stream_spec_compliant_writer", + "fetch_iterable_type_support", "enable_weak_ref", ], bindings = [ diff --git a/src/tests/streams/readable/readable-modules.capnp b/src/tests/streams/readable/readable-modules.capnp index 9f9e6966b7b..6bf1695edab 100644 --- a/src/tests/streams/readable/readable-modules.capnp +++ b/src/tests/streams/readable/readable-modules.capnp @@ -35,4 +35,5 @@ const modules :List(Workerd.Worker.Module) = [ (name = "transfer", esModule = embed "transfer.js"), (name = "pollution", esModule = embed "pollution.js"), (name = "unhandled-rejection", esModule = embed "unhandled-rejection.js"), + (name = "string-chunks", esModule = embed "string-chunks.js"), ]; diff --git a/src/tests/streams/readable/readable-ts.wd-test b/src/tests/streams/readable/readable-ts.wd-test index abb3f17f74c..40883201fe8 100644 --- a/src/tests/streams/readable/readable-ts.wd-test +++ b/src/tests/streams/readable/readable-ts.wd-test @@ -15,6 +15,7 @@ const unitTests :Workerd.Config = ( "typescript_implemented_streams", "expose_draining_reader", "experimental", + "fetch_iterable_type_support", "enable_weak_ref", ], bindings = [ diff --git a/src/tests/streams/readable/string-chunks.js b/src/tests/streams/readable/string-chunks.js new file mode 100644 index 00000000000..b3ba17d3418 --- /dev/null +++ b/src/tests/streams/readable/string-chunks.js @@ -0,0 +1,92 @@ +// Copyright (c) 2026 Cloudflare, Inc. +// Licensed under the Apache 2.0 license found in the LICENSE file or at: +// https://opensource.org/licenses/Apache-2.0 + +// String chunks in a body the runtime sends on: a fetch request body, an +// iterable request body, and a response the fetch handler returns, +// observed through the SELF service. +// +// PARITY: a string chunk is sent as its UTF-8 encoding, lone surrogates +// replaced with U+FFFD and embedded NULs kept, alongside the byte chunks +// around it. Strings are not valid body chunks (a body read in JavaScript, +// text() and the like, rejects them in both; integration-body.js), but +// the C++ implementation has always encoded them when it sends a body on, +// and code such as server-sent-event handlers relies on it. + +import { deepStrictEqual, strictEqual } from 'node:assert'; + +const enc = new TextEncoder(); + +// Bytes and strings in one stream, with the UTF-8 the runtime sends for +// them. +const MIXED_CHUNKS = [enc.encode('ab'), 'cd', '', '\uD800', 'x\0y']; +const MIXED_BYTES = [ + ...enc.encode('abcd'), + 0xef, + 0xbf, + 0xbd, + ...enc.encode('x\0y'), +]; + +function streamOf(chunks) { + return new ReadableStream({ + start(c) { + for (const chunk of chunks) c.enqueue(chunk); + c.close(); + }, + }); +} + +async function echoedBytes(env, body) { + const resp = await env.SELF.fetch('http://example.org', { + method: 'POST', + body, + }); + return [...new Uint8Array(await resp.arrayBuffer())]; +} + +// Served by main.js at /strings: an event stream whose second event is a +// string chunk. +export function stringsResponse() { + return new Response( + streamOf([enc.encode('data: one\n\n'), 'data: two\n\n']), + { headers: { 'content-type': 'text/event-stream' } } + ); +} + +export const stringChunksInRequestBody = { + async test(ctrl, env) { + deepStrictEqual( + await echoedBytes(env, streamOf(MIXED_CHUNKS)), + MIXED_BYTES + ); + + // One chunk per pull, so the strings reach the pump in batches of + // their own. + let i = 0; + const pulled = new ReadableStream({ + pull(c) { + if (i < MIXED_CHUNKS.length) c.enqueue(MIXED_CHUNKS[i++]); + else c.close(); + }, + }); + deepStrictEqual(await echoedBytes(env, pulled), MIXED_BYTES); + }, +}; + +export const stringChunksInIterableRequestBody = { + async test(ctrl, env) { + async function* body() { + yield enc.encode('ab'); + yield 'cd'; + } + deepStrictEqual(await echoedBytes(env, body()), [...enc.encode('abcd')]); + }, +}; + +export const stringChunksInReturnedResponse = { + async test(ctrl, env) { + const resp = await env.SELF.fetch('http://example.org/strings'); + strictEqual(await resp.text(), 'data: one\n\ndata: two\n\n'); + }, +}; diff --git a/src/workerd/api/js-readable-stream-test.c++ b/src/workerd/api/js-readable-stream-test.c++ index 08325bb7f11..02c6ba4aa42 100644 --- a/src/workerd/api/js-readable-stream-test.c++ +++ b/src/workerd/api/js-readable-stream-test.c++ @@ -1909,7 +1909,7 @@ KJ_TEST("JsReadableStream pumpTo rejects when a queued stream produces non-byte js.v8Context(), [](jsg::Lock& js, const v8::FunctionCallbackInfo& info) { auto controller = KJ_ASSERT_NONNULL(jsg::JsValue(info[0]).tryCast()); auto enqueue = KJ_ASSERT_NONNULL(controller.get(js, "enqueue"_kj).tryCast()); - enqueue.call(js, controller, js.str("not bytes"_kj)); + enqueue.call(js, controller, js.num(42)); }))); auto stream = makeTsStream(js, underlying); @@ -2080,6 +2080,35 @@ KJ_TEST("JsReadableStream pumpTo of a queued TypeScript stream dropped in IoCont KJ_EXPECT(!ended); } +KJ_TEST("JsReadableStream pumpTo writes a queued stream's string chunks as UTF-8") { + auto fixture = makeTsStreamsFixture(); + kj::Vector collected; + bool ended = false; + fixture.runInIoContext([&](const TestFixture::Environment& env) -> kj::Promise { + auto& js = env.js; + + // Bytes, then strings, in one batch, as the legacy pump's valueToBytes() writes them. + auto underlying = js.obj(); + underlying.set(js, "start"_kj, + jsg::JsValue(js.wrapSimpleFunction( + js.v8Context(), [](jsg::Lock& js, const v8::FunctionCallbackInfo& info) { + auto controller = KJ_ASSERT_NONNULL(jsg::JsValue(info[0]).tryCast()); + auto enqueue = KJ_ASSERT_NONNULL(controller.get(js, "enqueue"_kj).tryCast()); + enqueue.call(js, controller, jsg::JsUint8Array::create(js, "ab"_kjb)); + enqueue.call(js, controller, js.str("cd"_kj)); + enqueue.call(js, controller, js.str("\xc3\xa9"_kj)); + auto close = KJ_ASSERT_NONNULL(controller.get(js, "close"_kj).tryCast()); + close.call(js, controller); + }))); + + auto stream = makeTsStream(js, underlying); + return stream.pumpTo(js, kj::heap(collected, ended), EndStream::YES) + .then([](DeferredProxy proxy) { return kj::mv(proxy.proxyTask); }); + }); + KJ_EXPECT(collected.asPtr() == "abcd\xc3\xa9"_kjb); + KJ_EXPECT(ended); +} + KJ_TEST("JsReadableStream text() over the limit cancels a queued TypeScript stream") { auto fixture = makeTsStreamsFixture(); kj::Maybe cancelReason; diff --git a/src/workerd/api/js-readable-stream.c++ b/src/workerd/api/js-readable-stream.c++ index 25654ab8b86..62abfb33e8b 100644 --- a/src/workerd/api/js-readable-stream.c++ +++ b/src/workerd/api/js-readable-stream.c++ @@ -545,9 +545,10 @@ jsg::Promise queuedPumpStep(jsg::Lock& js, kj::Rc state, "ReadableStreamDrainingReader.read() promise did not resolve to an object " "with a chunks array."); - // Only ArrayBuffer/ArrayBufferView chunks are usable as bytes. Everything else - // (including strings) rejects, matching the legacy pump's historical behavior and - // error text exactly. + // ArrayBuffer/ArrayBufferView chunks are written as their bytes, and strings as their + // UTF-8 encoding (lone surrogates replaced with U+FFFD), as the legacy pump's + // valueToBytes() writes them. Everything else rejects. JS body consumption (text() and + // the like) rejects strings in both implementations; only pumps encode them. auto pieces = kj::Vector>(chunks.size()); for (uint32_t i = 0; i < chunks.size(); i++) { auto chunk = chunks.get(js, i); @@ -555,6 +556,8 @@ jsg::Promise queuedPumpStep(jsg::Lock& js, kj::Rc state, pieces.add(jsg::JsBufferSource(view).copy()); } else KJ_IF_SOME(buffer, JSG_TRY_CAST_ARRAYBUFFER(chunk)) { pieces.add(jsg::JsBufferSource(buffer).copy()); + } else KJ_IF_SOME(str, chunk.tryCast()) { + pieces.add(webstreams::stringToBytes(str.toUSVString(js))); } else { JSG_FAIL_REQUIRE(TypeError, "This ReadableStream did not return bytes."); } diff --git a/src/workerd/api/tests/BUILD.bazel b/src/workerd/api/tests/BUILD.bazel index 27c17237d72..38d2f8331bc 100644 --- a/src/workerd/api/tests/BUILD.bazel +++ b/src/workerd/api/tests/BUILD.bazel @@ -753,6 +753,12 @@ wd_test( data = ["eventsource-test.js"], ) +wd_test( + src = "eventsource-ts-test.wd-test", + args = ["--experimental"], + data = ["eventsource-test.js"], +) + wd_test( src = "legacy-dispatch-exceptions-test.wd-test", args = ["--experimental"], diff --git a/src/workerd/api/tests/eventsource-test.js b/src/workerd/api/tests/eventsource-test.js index 820d8dacb05..65ae08456f0 100644 --- a/src/workerd/api/tests/eventsource-test.js +++ b/src/workerd/api/tests/eventsource-test.js @@ -357,6 +357,31 @@ export const eventSourceFromTest = { }, }; +// String chunks are read as their UTF-8 encoding, beside byte chunks. +export const eventSourceFromStringChunksTest = { + async test() { + const rs = new ReadableStream({ + start(c) { + c.enqueue(new TextEncoder().encode('data: first\n\n')); + c.enqueue('data: caf\u00e9\n\n'); + c.close(); + }, + }); + const { promise, resolve } = Promise.withResolvers(); + const eventsource = EventSource.from(rs); + const events = []; + eventsource.onmessage = (event) => { + events.push(event.data); + if (events.length === 2) { + eventsource.close(); + resolve(); + } + }; + await promise; + deepStrictEqual(events, ['first', 'caf\u00e9']); + }, +}; + export const eventSourceFromWithBOMTest = { async test() { const enc = new TextEncoder(); diff --git a/src/workerd/api/tests/eventsource-ts-test.wd-test b/src/workerd/api/tests/eventsource-ts-test.wd-test new file mode 100644 index 00000000000..ab8135e79b5 --- /dev/null +++ b/src/workerd/api/tests/eventsource-ts-test.wd-test @@ -0,0 +1,20 @@ +using Workerd = import "/workerd/workerd.capnp"; + +# eventsource-test.js under the TypeScript streams implementation: an +# EventSource reads its stream through the C++ bridge, which handles the +# two implementations' streams differently. +const unitTests :Workerd.Config = ( + services = [ + ( name = "eventsource-ts-test", + worker = ( + modules = [ + (name = "worker", esModule = embed "eventsource-test.js") + ], + compatibilityFlags = ["nodejs_compat", "experimental", "streams_enable_constructors", "spec_compliant_dispatch_exceptions", "typescript_implemented_streams"], + bindings = [ + (name = "subrequest", service = "eventsource-ts-test") + ] + ) + ), + ], +); diff --git a/src/workerd/api/tests/js-rpc-streams-crossflag-test.js b/src/workerd/api/tests/js-rpc-streams-crossflag-test.js index eaf02acf4dd..2c9d177f8d8 100644 --- a/src/workerd/api/tests/js-rpc-streams-crossflag-test.js +++ b/src/workerd/api/tests/js-rpc-streams-crossflag-test.js @@ -37,6 +37,17 @@ function makeConstructorBackedEchoPair() { return { readable, writable }; } +// Bytes, then a string chunk, which the sender writes as its UTF-8 encoding. +function stringChunkReadable() { + return new ReadableStream({ + start(c) { + c.enqueue(enc.encode('ab')); + c.enqueue('cd\u00e9'); + c.close(); + }, + }); +} + async function isCrossImplementationRunner(env) { if (env.PEER === undefined) return false; assert.notStrictEqual(usingTsImpl, await env.PEER.usesTypeScriptStreams()); @@ -92,6 +103,10 @@ export class Peer extends WorkerEntrypoint { return inspectLength(stream, this.env.HTTP); } + makeStringChunkReadable() { + return stringChunkReadable(); + } + // Returns a peer-created readable. makeReadable() { return new ReadableStream({ @@ -234,6 +249,20 @@ export default { }, }; +// The sender writes a string chunk as its UTF-8 encoding, whichever implementation sends it and +// whichever receives it (the two configs swap them). +export const stringChunksCrossAsUtf8 = { + async test(controller, env) { + if (!(await isCrossImplementationRunner(env))) return; + assert.strictEqual( + await env.PEER.readFrom(stringChunkReadable()), + 'abcd\u00e9' + ); + const returned = await env.PEER.makeStringChunkReadable(); + assert.strictEqual(await new Response(returned).text(), 'abcd\u00e9'); + }, +}; + export const readableLength = { async test(controller, env) { if (!(await isCrossImplementationRunner(env))) return;