From 5a1bc2a6d52c627f5fda71b608e6537adf2ee90c Mon Sep 17 00:00:00 2001 From: Chris Tate Date: Tue, 27 Oct 2026 01:56:31 -0500 Subject: [PATCH] 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 | 12 ++-- .../8409-fetch-compressed-read-after-end.ts | 66 +++++++++++++++++++ ...09-fetch-compressed-read-after-end.ts.json | 1 + 3 files changed, 75 insertions(+), 4 deletions(-) create mode 100644 tests/corpus/8409-fetch-compressed-read-after-end.ts create mode 100644 tests/harness/corpus-costs/8409-fetch-compressed-read-after-end.ts.json diff --git a/packages/runtime/src/scr_fetch.c b/packages/runtime/src/scr_fetch.c index 652dfa8f1..189e97f84 100644 --- a/packages/runtime/src/scr_fetch.c +++ b/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)); } diff --git a/tests/corpus/8409-fetch-compressed-read-after-end.ts b/tests/corpus/8409-fetch-compressed-read-after-end.ts new file mode 100644 index 000000000..182c9405a --- /dev/null +++ b/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 => new Promise((resolve) => setTimeout(resolve, ms)); + +async function readSlowly(port: number, path: string): Promise { + 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 { + 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()); +}); diff --git a/tests/harness/corpus-costs/8409-fetch-compressed-read-after-end.ts.json b/tests/harness/corpus-costs/8409-fetch-compressed-read-after-end.ts.json new file mode 100644 index 000000000..e97925233 --- /dev/null +++ b/tests/harness/corpus-costs/8409-fetch-compressed-read-after-end.ts.json @@ -0,0 +1 @@ +{ "plain": 2800, "san": 42400 }