src/compression.py

@@ -25,7 +25,7 @@ def _quality(accept_encoding: str, coding: str) -> float:
 class NegotiatedGZipMiddleware:
     """Compress substantial responses only when the client permits gzip."""
 
-    def __init__(self, app: ASGIApp, minimum_size: int = 500, compresslevel: int = 9):
+    def __init__(self, app: ASGIApp, minimum_size: int = 500, compresslevel: int = 6):
         self.app = app
         self.minimum_size = minimum_size
         self.compresslevel = compresslevel
@@ -36,7 +36,9 @@ class NegotiatedGZipMiddleware:
             return
 
         accepted = Headers(scope=scope).get("Accept-Encoding", "")
-        if _quality(accepted, "gzip") > 0:
+        path = scope.get("path", "").rsplit("/", 1)[-1]
+        immutable_runtime = path.startswith("runtime-") and path.endswith(".js")
+        if _quality(accepted, "gzip") > 0 and not immutable_runtime:
             responder: ASGIApp = GZipResponder(
                 self.app, self.minimum_size, compresslevel=self.compresslevel
             )

src/frontend_bundle.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import gzip
 import hashlib
 import re
 from dataclasses import dataclass
@@ -16,6 +17,7 @@ WORKER_RUNTIME_SOURCE = "static/background-issue-sync.js"
 class FrontendBuild:
     dashboard_html: str
     runtime_bytes: bytes
+    runtime_gzip_bytes: bytes
     runtime_digest: str
     runtime_name: str
     page_sources: tuple[str, ...]
@@ -38,6 +40,7 @@ def build_frontend(frontend_dir: Path) -> FrontendBuild:
         raise ValueError("dashboard entry document has no page scripts")
 
     runtime = _bundle(frontend_dir, sources)
+    runtime_gzip = gzip.compress(runtime, compresslevel=6, mtime=0)
     digest = hashlib.sha256(runtime).hexdigest()[:16]
     runtime_name = f"runtime-{digest}.js"
     dashboard_html = SCRIPT_TAG.sub("", source_html)
@@ -63,6 +66,7 @@ def build_frontend(frontend_dir: Path) -> FrontendBuild:
     return FrontendBuild(
         dashboard_html=dashboard_html,
         runtime_bytes=runtime,
+        runtime_gzip_bytes=runtime_gzip,
         runtime_digest=digest,
         runtime_name=runtime_name,
         page_sources=sources,

src/views.py

@@ -1,9 +1,10 @@
 from pathlib import Path
 
-from fastapi import APIRouter, HTTPException
+from fastapi import APIRouter, HTTPException, Request
 from fastapi.responses import FileResponse, HTMLResponse, Response
 
 from src.frontend_bundle import build_frontend
+from src.compression import _quality
 
 router = APIRouter()
 DASHBOARD_FILE = Path(__file__).resolve().parent.parent / "frontend" / "index.html"
@@ -51,11 +52,20 @@ async def service_worker() -> Response:
 
 
 @router.get("/runtime-{digest}.js")
-async def runtime_bundle(digest: str) -> Response:
+async def runtime_bundle(digest: str, request: Request) -> Response:
     if digest != FRONTEND_BUILD.runtime_digest:
         raise HTTPException(status_code=404, detail="Runtime revision not found")
+    accepts_gzip = _quality(request.headers.get("Accept-Encoding", ""), "gzip") > 0
     return Response(
-        FRONTEND_BUILD.runtime_bytes,
+        (
+            FRONTEND_BUILD.runtime_gzip_bytes
+            if accepts_gzip
+            else FRONTEND_BUILD.runtime_bytes
+        ),
         media_type="application/javascript",
-        headers={"Cache-Control": "public, max-age=31536000, immutable"},
+        headers={
+            "Cache-Control": "public, max-age=31536000, immutable",
+            "Vary": "Accept-Encoding",
+            **({"Content-Encoding": "gzip"} if accepts_gzip else {}),
+        },
     )

tests/test_frontend_bundle.py

@@ -5,6 +5,7 @@ from pathlib import Path
 
 import httpx
 import pytest
+from starlette.requests import Request
 
 from src import main
 from src.frontend_bundle import build_frontend
@@ -21,6 +22,8 @@ def test_page_runtime_is_one_deterministic_content_addressed_bundle(tmp_path):
 
     assert first.runtime_name == second.runtime_name
     assert first.runtime_bytes == second.runtime_bytes
+    assert first.runtime_gzip_bytes == second.runtime_gzip_bytes
+    assert gzip.decompress(first.runtime_gzip_bytes) == first.runtime_bytes
     assert PAGE_SCRIPT.findall(first.dashboard_html) == []
     assert first.dashboard_html.count("<script src=") == 1
     assert f'<script src="{first.runtime_name}"></script>' in first.dashboard_html
@@ -52,11 +55,46 @@ async def test_runtime_is_transferred_as_gzip_when_the_client_accepts_it():
     assert response.status_code == 200
     assert response.headers["content-encoding"] == "gzip"
     assert response.headers["vary"] == "Accept-Encoding"
-    assert int(response.headers["content-length"]) <= 100 * 1024
+    assert int(response.headers["content-length"]) == len(main.FRONTEND_BUILD.runtime_gzip_bytes)
     assert response.content == main.FRONTEND_BUILD.runtime_bytes
     assert response.headers["cache-control"] == "public, max-age=31536000, immutable"
 
 
+@pytest.mark.anyio
+async def test_runtime_route_returns_the_precomputed_gzip_representation():
+    build = build_frontend(FRONTEND)
+    request = Request(
+        {"type": "http", "method": "GET", "path": f"/{build.runtime_name}",
+         "headers": [(b"accept-encoding", b"gzip")]}
+    )
+
+    response = await runtime_bundle(build.runtime_digest, request)
+
+    assert response.body == build.runtime_gzip_bytes
+    assert response.headers["content-encoding"] == "gzip"
+    assert response.headers["content-length"] == str(len(build.runtime_gzip_bytes))
+    assert response.headers["vary"] == "Accept-Encoding"
+
+
+@pytest.mark.anyio
+async def test_runtime_requests_bypass_dynamic_gzip_compression(monkeypatch):
+    class UnexpectedDynamicCompression:
+        def __init__(self, *args, **kwargs):
+            raise AssertionError("immutable runtime entered dynamic compression")
+
+    monkeypatch.setattr("src.compression.GZipResponder", UnexpectedDynamicCompression)
+    transport = httpx.ASGITransport(app=main.app)
+
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        response = await client.get(
+            f"/{main.FRONTEND_BUILD.runtime_name}",
+            headers={"Accept-Encoding": "gzip"},
+        )
+
+    assert response.status_code == 200
+    assert response.headers["content-encoding"] == "gzip"
+
+
 @pytest.mark.anyio
 @pytest.mark.parametrize("accept_encoding", ["identity", "gzip;q=0, identity;q=1"])
 async def test_runtime_stays_unencoded_when_the_client_declines_gzip(accept_encoding):
@@ -108,9 +146,13 @@ def test_offline_shell_precaches_exact_runtime_without_superseded_page_modules()
 @pytest.mark.anyio
 async def test_runtime_asset_is_immutable_while_html_and_worker_revalidate():
     build = build_frontend(FRONTEND)
+    request = Request(
+        {"type": "http", "method": "GET", "path": f"/{build.runtime_name}",
+         "headers": [(b"accept-encoding", b"identity")]}
+    )
 
     html = await dashboard()
-    bundle = await runtime_bundle(build.runtime_digest)
+    bundle = await runtime_bundle(build.runtime_digest, request)
     worker = await service_worker()
 
     assert build.runtime_name in html