From d4926c9d24bab0b7710091ec312754ad17c1d544 Mon Sep 17 00:00:00 2001 From: Nathan Renzoni <31897391+nrenzoni@users.noreply.github.com> Date: Tue, 29 Sep 2026 22:35:02 -0400 Subject: [PATCH] fix(runtime-sdk): preserve encoded ASGI response bodies --- packages/runtime-sdk/README.md | 11 +++ packages/runtime-sdk/src/workers/asgi.py | 11 ++- .../tests/workerd-test/asgi/asgi.wd-test | 15 ++++ .../workerd-test/asgi/tests/test_asgi.py | 38 +++++++++- .../tests/workerd-test/asgi/worker.py | 69 +++++++++++++++++++ 5 files changed, 141 insertions(+), 3 deletions(-) diff --git a/packages/runtime-sdk/README.md b/packages/runtime-sdk/README.md index 9bd57fb3..31cbe8a1 100644 --- a/packages/runtime-sdk/README.md +++ b/packages/runtime-sdk/README.md @@ -1,3 +1,14 @@ # workers-runtime-sdk Runtime SDK for Python Cloudflare Workers. + +## ASGI response encoding + +The ASGI adapter preserves application response bytes and `Content-Encoding`. +Applications and middleware are responsible for encoding their response bodies; +the adapter uses the Workers Fetch API's manual encoding mode to avoid +compressing an already-compressed response a second time. + +Applications must negotiate supported encodings with the request's +`Accept-Encoding` header. Manual encoding does not make a gzip response +appropriate for an identity-only client. diff --git a/packages/runtime-sdk/src/workers/asgi.py b/packages/runtime-sdk/src/workers/asgi.py index 06a647a0..35977c27 100644 --- a/packages/runtime-sdk/src/workers/asgi.py +++ b/packages/runtime-sdk/src/workers/asgi.py @@ -261,8 +261,12 @@ async def send(got): transform_stream = TransformStream.new() readable = transform_stream.readable writer = transform_stream.writable.getWriter() + # ASGI bytes already match Content-Encoding; do not encode twice. resp = Response.new( - readable, headers=_to_js_headers(headers), status=status + readable, + headers=_to_js_headers(headers), + status=status, + encodeBody="manual", ) result.set_result(resp) with acquire_js_buffer(body) as jsbytes: @@ -281,7 +285,10 @@ async def send(got): buf = px.getBuffer() px.destroy() resp = Response.new( - buf.data, headers=_to_js_headers(headers), status=status + buf.data, + headers=_to_js_headers(headers), + status=status, + encodeBody="manual", ) result.set_result(resp) finished_response.set() diff --git a/packages/runtime-sdk/tests/workerd-test/asgi/asgi.wd-test b/packages/runtime-sdk/tests/workerd-test/asgi/asgi.wd-test index 754f4e29..ab0d62b1 100644 --- a/packages/runtime-sdk/tests/workerd-test/asgi/asgi.wd-test +++ b/packages/runtime-sdk/tests/workerd-test/asgi/asgi.wd-test @@ -10,10 +10,25 @@ const unitTests :Workerd.Config = ( ], bindings = [ ( name = "SELF", service = "python-asgi" ), + ( name = "ASGI_HTTP", service = "asgi-http" ), ], compatibilityDate = "%COMPAT_DATE", compatibilityFlags = ["python_workers"], ) + ), + ( name = "asgi-http", + external = ( + address = "loopback:asgi-http", + http = (), + ) ) ], + sockets = [ + ( + name = "asgi-http", + address = "loopback:asgi-http", + http = (), + service = "python-asgi", + ), + ], ); diff --git a/packages/runtime-sdk/tests/workerd-test/asgi/tests/test_asgi.py b/packages/runtime-sdk/tests/workerd-test/asgi/tests/test_asgi.py index ca77eeaa..3ed6f2d8 100644 --- a/packages/runtime-sdk/tests/workerd-test/asgi/tests/test_asgi.py +++ b/packages/runtime-sdk/tests/workerd-test/asgi/tests/test_asgi.py @@ -5,7 +5,12 @@ import js import pytest from pyodide.ffi import to_js -from worker import STREAMING_CHUNK_SIZE, STREAMING_NUM_CHUNKS, example_hdr +from worker import ( + ENCODED_RESPONSE_PAYLOAD, + STREAMING_CHUNK_SIZE, + STREAMING_NUM_CHUNKS, + example_hdr, +) import asgi from workers import Request, env @@ -89,6 +94,37 @@ async def test_streaming(): assert all(b == i % 256 for b in chunk) +@pytest.mark.asyncio +@pytest.mark.parametrize("encoding", ["gzip", "identity"]) +@pytest.mark.parametrize("mode", ["buffered", "streaming"]) +async def test_already_encoded_response_crosses_http_unchanged(encoding, mode): + path = f"/encoded/{encoding}/{mode}" + # SELF.fetch does not serialize the response over HTTP, so it cannot catch + # workerd compressing already-gzipped ASGI bytes a second time. ASGI_HTTP + # instead connects to the worker's loopback HTTP socket. + response = await env.ASGI_HTTP.fetch( + f"http://example.com{path}", + headers={"accept-encoding": "gzip, identity"}, + ) + + assert response.status == 201 + assert response.headers["content-type"] == "application/json; charset=utf-8" + assert response.headers["content-encoding"] == encoding + assert response.headers["vary"] == "Accept-Encoding" + assert response.headers["x-asgi-response"] == "already-encoded" + + reader = response.body.getReader() + chunks = [] + while True: + result = await reader.read() + if result.done: + break + chunks.append(result.value.to_bytes()) + # Fetch decodes the HTTP content encoding once, just as a browser does. + # Double gzip therefore leaves a gzip member here instead of the JSON. + assert b"".join(chunks) == ENCODED_RESPONSE_PAYLOAD + + class _ListHandler(logging.Handler): """A logging handler that captures records into a list for assertions.""" diff --git a/packages/runtime-sdk/tests/workerd-test/asgi/worker.py b/packages/runtime-sdk/tests/workerd-test/asgi/worker.py index 68e94ca9..21738875 100644 --- a/packages/runtime-sdk/tests/workerd-test/asgi/worker.py +++ b/packages/runtime-sdk/tests/workerd-test/asgi/worker.py @@ -1,3 +1,4 @@ +import gzip import os from testlib.entrypoint import run_pytest @@ -159,6 +160,63 @@ async def __call__(self, scope, receive, send): ) +ENCODED_RESPONSE_PAYLOAD = ( + b'{"message":"ASGI middleware already encoded this response",' + b'"padding":"' + b"compression-regression-" * 128 + b'"}' +) + + +class EncodedResponseApp: + """Return middleware-produced bytes unchanged, buffered or streamed.""" + + def __init__(self, encoding, streaming): + self.encoding = encoding + self.streaming = streaming + self.body = ( + gzip.compress(ENCODED_RESPONSE_PAYLOAD, mtime=0) + if encoding == "gzip" + else ENCODED_RESPONSE_PAYLOAD + ) + + async def __call__(self, scope, receive, send): + if scope["type"] == "lifespan": + message = await receive() + if message["type"] == "lifespan.startup": + await send({"type": "lifespan.startup.complete"}) + return + + await receive() + headers = [ + (b"content-type", b"application/json; charset=utf-8"), + (b"content-encoding", self.encoding.encode()), + (b"vary", b"Accept-Encoding"), + (b"x-asgi-response", b"already-encoded"), + ] + if not self.streaming: + headers.append((b"content-length", str(len(self.body)).encode())) + await send( + { + "type": "http.response.start", + "status": 201, + "headers": headers, + } + ) + if self.streaming: + # Split inside the gzip header and before its footer, rather than + # emitting separate gzip members for the individual ASGI chunks. + chunks = (self.body[:7], self.body[7:-8], self.body[-8:], b"") + for index, chunk in enumerate(chunks): + await send( + { + "type": "http.response.body", + "body": chunk, + "more_body": index < len(chunks) - 1, + } + ) + else: + await send({"type": "http.response.body", "body": self.body}) + + # --------------------------------------------------------------------------- # App instances and constants # --------------------------------------------------------------------------- @@ -244,6 +302,13 @@ async def __call__(self, scope, receive, send): scope_echo_app = ScopeEchoApp() late_failure_stream_app = LateFailureStreamApp() multi_cookie_app = MultiCookieApp() +encoded_response_apps = { + f"/encoded/{encoding}/{mode}": EncodedResponseApp( + encoding, streaming=mode == "streaming" + ) + for encoding in ("gzip", "identity") + for mode in ("buffered", "streaming") +} example_hdr = {"Header1": "Value1", "Header2": "Value2"} @@ -255,6 +320,10 @@ async def fetch(self, request): url = URL.new(request.url) path = url.pathname + if path in encoded_response_apps: + return await asgi.fetch( + encoded_response_apps[path], request, self.env, self.ctx + ) if path == "/sse": return await asgi.fetch(sse_app, request, self.env, self.ctx) elif path == "/stream":