Drain a compressed fetch body that is still buffered when the response ends

Since the transfer drops its socket at the response's 'end', resuming a
read no longer reached the decoder: sf_response_resume_if_ready returned
before sf_inflate_resume once the transfer held no paused socket. A
compressed response whose whole body arrived before it was read then kept
its remaining decoded output buffered, and later reads stayed pending. The
buffered input and decoder output now drain on demand after 'end' as well
as while the socket is paused.

packages/runtime/src/scr_fetch.c

@@ -1129,15 +1129,19 @@ static void sf_response_pause_if_full(SfStream *s) {
 
 static void sf_response_resume_if_ready(SfStream *s) {
   SfTransfer *t = s->response_owner;
-  if (!t || t->done || !t->response_paused || !t->response_socket ||
-      s->queued >= 1 || s->close_requested || s->closed || s->error) {
+  if (!t || t->done || s->queued >= 1 || s->close_requested || s->closed ||
+      s->error || t->inflate_draining) {
     return;
   }
-  if (t->inflate_draining) return;
-  if (t->inflating && (t->inflate_pending || t->inflate_needs_drain)) {
+  /* Compressed input or decoder output still buffered drains on demand:
+   * while the socket is paused, and after 'end', when the transfer no
+   * longer holds the socket and no more data arrives. */
+  if (t->inflating && (t->inflate_pending || t->inflate_needs_drain) &&
+      (t->response_paused || t->response_ended)) {
     sf_inflate_resume(t);
     return;
   }
+  if (!t->response_paused || !t->response_socket) return;
   t->response_paused = false;
   scr_net_sock_release(scr_net_sock_resume(t->response_socket));
 }

tests/corpus/8409-fetch-compressed-read-after-end.ts

@@ -0,0 +1,66 @@
+// A compressed fetch() response whose whole body arrived before it was read:
+// the decoded body is larger than one read, so decoded output is still
+// buffered when the response ends, and every later read drains it. Reads
+// after a pause (the end was processed meanwhile) return the rest of the
+// body, as in Node, for gzip and deflate, through a reader and through
+// text(), and the next request on the same connection works.
+import { createServer, type IncomingMessage, type ServerResponse } from "node:http";
+import type { AddressInfo } from "node:net";
+import { deflateSync, gzipSync } from "node:zlib";
+
+const body = Array.from({ length: 20000 }, (_, i) => `line ${i % 97} of the body\n`).join("");
+const gzipped = gzipSync(body);
+const deflated = deflateSync(body);
+
+const server = createServer((req: IncomingMessage, res: ServerResponse) => {
+  const url = req.url ?? "/";
+  if (url === "/gzip" || url === "/gzip-text") {
+    res.writeHead(200, { "content-encoding": "gzip", "content-length": gzipped.length });
+    res.end(gzipped);
+    return;
+  }
+  if (url === "/deflate") {
+    res.writeHead(200, { "content-encoding": "deflate", "content-length": deflated.length });
+    res.end(deflated);
+    return;
+  }
+  res.end(`plain ${url}`);
+});
+
+const pause = (ms: number): Promise<void> => new Promise((resolve) => setTimeout(resolve, ms));
+
+async function readSlowly(port: number, path: string): Promise<string> {
+  const res = await fetch(`http://127.0.0.1:${port}${path}`);
+  const reader = res.body!.getReader();
+  const decoder = new TextDecoder();
+  let text = "";
+  let reads = 0;
+  for (;;) {
+    // Let the response end before most reads.
+    await pause(reads < 3 ? 30 : 0);
+    const { done, value } = await reader.read();
+    if (done) break;
+    reads++;
+    text += decoder.decode(value, { stream: true });
+  }
+  text += decoder.decode();
+  return `${path} ${res.status} ${text.length} ${text === body}`;
+}
+
+async function main(port: number): Promise<void> {
+  console.log(await readSlowly(port, "/gzip"));
+  console.log(await readSlowly(port, "/deflate"));
+  const res = await fetch(`http://127.0.0.1:${port}/gzip-text`);
+  await pause(50);
+  const text = await res.text();
+  console.log("text after end", text.length, text === body);
+  const plain = await fetch(`http://127.0.0.1:${port}/after`);
+  console.log(await plain.text());
+}
+
+server.listen(0, "127.0.0.1", () => {
+  const { port } = server.address() as AddressInfo;
+  main(port)
+    .catch((error: unknown) => console.log("failed", String(error)))
+    .finally(() => server.close());
+});

tests/harness/corpus-costs/8409-fetch-compressed-read-after-end.ts.json

@@ -0,0 +1 @@
+{ "plain": 2800, "san": 42400 }