Read a stream body to its end when its queue filled before the read (#2456)

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-queued-before-read-untyped.mjs.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/fetch-body-queued-before-read-untyped.mjs"
+ ],
+ "diags": []
+}

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/fetch-body-queued-before-read.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/fetch-body-queued-before-read.ts"
+ ],
+ "diags": []
+}

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

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);

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<void> => new Promise((resolve) => setTimeout(resolve, 0));
+
+async function main(): Promise<void> {
+  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<Uint8Array>({
+      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<Uint8Array>({
+      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();

tests/harness/corpus-costs/fetch-body-queued-before-read-untyped.mjs.json

@@ -0,0 +1 @@
+{ "plain": 1900, "san": 43700 }

tests/harness/corpus-costs/fetch-body-queued-before-read.ts.json

@@ -0,0 +1 @@
+{ "plain": 1900, "san": 43700 }