From 30d611e9da00ba70f3678e4663413a61ae91ea68 Mon Sep 17 00:00:00 2001 From: Chris Tate Date: Wed, 28 Oct 2026 00:23:51 -0500 Subject: [PATCH] Read a stream body to its end when its queue filled before the read text(), json(), bytes() and the other body readers of a Response or Request whose body is a ReadableStream asked the stream's source for more chunks after draining its queue only when the stream had a zero high-water mark. When the queue had filled before the read began (a timer ran after the Response was made), the reader took the queued chunk and then waited for a close that no pull would bring, and the read never settled. The reader now pulls whenever it empties the queue, so it reads to the end, as Node does. This applies to streams from ReadableStream.from, from an underlying source, and to JavaScript ReadableStreams. Adds corpus programs fetch-body-queued-before-read.ts and fetch-body-queued-before-read-untyped.mjs, with their corpus costs and preflight/order baselines. --- ...h-body-queued-before-read-untyped.mjs.json | 6 +++ .../fetch-body-queued-before-read.ts.json | 6 +++ packages/runtime/src/scr_fetch.c | 4 +- .../fetch-body-queued-before-read-untyped.mjs | 13 ++++++ tests/corpus/fetch-body-queued-before-read.ts | 45 +++++++++++++++++++ ...h-body-queued-before-read-untyped.mjs.json | 1 + .../fetch-body-queued-before-read.ts.json | 1 + 7 files changed, 75 insertions(+), 1 deletion(-) create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-queued-before-read-untyped.mjs.json create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-queued-before-read.ts.json create mode 100644 tests/corpus/fetch-body-queued-before-read-untyped.mjs create mode 100644 tests/corpus/fetch-body-queued-before-read.ts create mode 100644 tests/harness/corpus-costs/fetch-body-queued-before-read-untyped.mjs.json create mode 100644 tests/harness/corpus-costs/fetch-body-queued-before-read.ts.json diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-queued-before-read-untyped.mjs.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-queued-before-read-untyped.mjs.json new file mode 100644 index 0000000000..7287db88ae --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-queued-before-read-untyped.mjs.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/fetch-body-queued-before-read-untyped.mjs" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-queued-before-read.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-queued-before-read.ts.json new file mode 100644 index 0000000000..601f77bc86 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-queued-before-read.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/fetch-body-queued-before-read.ts" + ], + "diags": [] +} diff --git a/packages/runtime/src/scr_fetch.c b/packages/runtime/src/scr_fetch.c index c946b74be1..00dcc9b41d 100644 --- a/packages/runtime/src/scr_fetch.c +++ b/packages/runtime/src/scr_fetch.c @@ -2014,7 +2014,9 @@ static ScrPromise *sf_stream_collect(SfStream *s, int mode) { c->mode = mode; s->collector = c; sf_stream_drain(s); - if (s->hwm_zero && !s->head) sf_stream_pull(s); + /* The collector reads to the end: a queue it emptied asks the source + * for the next chunk, also when the queue had filled before. */ + if (!s->head) sf_stream_pull(s); return p; } diff --git a/tests/corpus/fetch-body-queued-before-read-untyped.mjs b/tests/corpus/fetch-body-queued-before-read-untyped.mjs new file mode 100644 index 0000000000..ca1e422c62 --- /dev/null +++ b/tests/corpus/fetch-body-queued-before-read-untyped.mjs @@ -0,0 +1,13 @@ +// The same from JavaScript: a body from a JavaScript ReadableStream whose +// first chunk is queued before text() reads it. +const encoder = new TextEncoder(); +const tick = () => new Promise((resolve) => setTimeout(resolve, 0)); + +const response = new Response(new ReadableStream({ start(c) { c.enqueue(encoder.encode("late ")); c.enqueue(encoder.encode("read")); c.close(); } })); +await tick(); +console.log("text", await response.text(), response.bodyUsed); + +let pulls = 0; +const pulled = new Response(new ReadableStream({ pull(c) { pulls++; c.enqueue(encoder.encode(String(pulls))); if (pulls === 4) c.close(); } })); +await tick(); +console.log("pull", await pulled.text(), pulls); diff --git a/tests/corpus/fetch-body-queued-before-read.ts b/tests/corpus/fetch-body-queued-before-read.ts new file mode 100644 index 0000000000..5dfd89b62e --- /dev/null +++ b/tests/corpus/fetch-body-queued-before-read.ts @@ -0,0 +1,45 @@ +// A Response or Request body from a ReadableStream whose queue filled +// before the body is read: text(), bytes() and json() read the rest of the +// stream to its close after a timer has run. +const encoder = new TextEncoder(); +const tick = (): Promise => new Promise((resolve) => setTimeout(resolve, 0)); + +async function main(): Promise { + const fromArray = new Response(ReadableStream.from([encoder.encode("from "), encoder.encode("array")])); + await tick(); + console.log("from", await fromArray.text(), fromArray.bodyUsed); + + let pulls = 0; + const pulled = new Response( + new ReadableStream({ + pull(controller) { + pulls++; + controller.enqueue(encoder.encode(`p${pulls}`)); + if (pulls === 3) controller.close(); + }, + }), + ); + await tick(); + console.log("pull", await pulled.text(), pulls); + + const started = new Request("http://example.com/", { + method: "POST", + body: new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode('{"ok":')); + controller.enqueue(encoder.encode("true}")); + controller.close(); + }, + }), + duplex: "half", + } as RequestInit); + await tick(); + console.log("json", JSON.stringify(await started.json())); + + const bytes = new Response(ReadableStream.from([new Uint8Array([1, 2]), new Uint8Array([3])])); + await tick(); + await tick(); + console.log("bytes", Array.from(await bytes.bytes()).join(",")); +} + +await main(); diff --git a/tests/harness/corpus-costs/fetch-body-queued-before-read-untyped.mjs.json b/tests/harness/corpus-costs/fetch-body-queued-before-read-untyped.mjs.json new file mode 100644 index 0000000000..b1bf2ddf8a --- /dev/null +++ b/tests/harness/corpus-costs/fetch-body-queued-before-read-untyped.mjs.json @@ -0,0 +1 @@ +{ "plain": 1900, "san": 43700 } diff --git a/tests/harness/corpus-costs/fetch-body-queued-before-read.ts.json b/tests/harness/corpus-costs/fetch-body-queued-before-read.ts.json new file mode 100644 index 0000000000..b1bf2ddf8a --- /dev/null +++ b/tests/harness/corpus-costs/fetch-body-queued-before-read.ts.json @@ -0,0 +1 @@ +{ "plain": 1900, "san": 43700 }