fix(frontend): stream dynamic response compression

projects/monolith/frontend/src/hooks.server.js

@@ -16,14 +16,11 @@
 // A note on `reroute`: it lives in the universal hooks.js (runs on both server
 // and client). `handle` is server-only and must live here.
 
-import { promisify } from "node:util";
+import { Readable, pipeline } from "node:stream";
 import zlib from "node:zlib";
 
-const gzip = promisify(zlib.gzip);
-const brotli = promisify(zlib.brotliCompress);
-
 // Below ~one MTU the framing + header overhead of compression outweighs the
-// savings, so small bodies (most API JSON, redirects, 304s) pass through.
+// savings, so bodies with a known small length pass through.
 const MIN_BYTES = 1400;
 
 // Only text-shaped payloads. Images/fonts/video are already compressed;
@@ -91,7 +88,7 @@ export async function handle({ event, resolve }) {
   }
 
   // HEAD must not have its body materialised; pass through untouched.
-  if (event.request.method === "HEAD") return response;
+  if (event.request.method === "HEAD" || !response.body) return response;
   // Respect anything upstream already encoded.
   if (response.headers.has("content-encoding")) return response;
 
@@ -108,36 +105,30 @@ export async function handle({ event, resolve }) {
   const declaredLen = Number(response.headers.get("content-length"));
   if (declaredLen && declaredLen < MIN_BYTES) return response;
 
-  // Buffer the body. Every browser-facing payload that reaches here is already
-  // buffered upstream (json(await res.json()) for proxies, an SSR string for
-  // pages), so this defeats no streaming we rely on -- and event-stream, the
-  // one thing we DO stream, was excluded above.
-  const original = new Uint8Array(await response.arrayBuffer());
-  if (original.byteLength < MIN_BYTES) {
-    // arrayBuffer() consumed the body; re-emit the small payload verbatim.
-    return new Response(original, {
-      status: response.status,
-      statusText: response.statusText,
-      headers: response.headers,
-    });
-  }
-
-  const compressed =
+  // Stream through the compressor with backpressure. Buffering every complete
+  // response here multiplies memory during concurrent SSR/API requests.
+  // Unknown-length bodies are compressed even when small so finding their size
+  // never requires retaining the full response.
+  const compressor =
     encoding === "br"
-      ? await brotli(original, {
-          params: {
-            // Quality 5 lands near gzip-9's ratio at a fraction of brotli-11's
-            // CPU. The async API runs on the libuv threadpool, so a big body
-            // never blocks the Node event loop / other in-flight requests.
-            [zlib.constants.BROTLI_PARAM_QUALITY]: 5,
-            [zlib.constants.BROTLI_PARAM_SIZE_HINT]: original.byteLength,
-          },
+      ? zlib.createBrotliCompress({
+          params: { [zlib.constants.BROTLI_PARAM_QUALITY]: 5 },
         })
-      : await gzip(original, { level: 6 });
+      : zlib.createGzip({ level: 6 });
+  const compressed = Readable.toWeb(compressor, {
+    strategy: {
+      highWaterMark: 64 * 1024,
+      size: (chunk) => chunk.byteLength,
+    },
+  });
+  // pipeline propagates source errors and destroys the source on downstream
+  // cancellation. The response reader observes errors through toWeb; consume
+  // the callback error so there is no unhandled background rejection.
+  pipeline(Readable.fromWeb(response.body), compressor, () => {});
 
   const headers = new Headers(response.headers);
   headers.set("content-encoding", encoding);
-  headers.set("content-length", String(compressed.byteLength));
+  headers.delete("content-length");
 
   // Caches (Cloudflare) must key on accept-encoding so a brotli body is never
   // handed to a gzip-only client. Add the token without clobbering an existing

projects/monolith/frontend/src/hooks.server.test.js

@@ -1,5 +1,5 @@
 import { describe, expect, it } from "vitest";
-import { gunzipSync } from "node:zlib";
+import { brotliDecompressSync, gunzipSync } from "node:zlib";
 import { handle, pickEncoding } from "./hooks.server.js";
 
 describe("response compression", () => {
@@ -153,7 +153,12 @@ describe("Activity index cache isolation", () => {
         "no-store",
       );
       expect(response.headers.has("etag")).toBe(false);
-      expect(await response.json()).toEqual({ daily: [] });
+      const body = Buffer.from(await response.arrayBuffer());
+      const decoded =
+        response.headers.get("content-encoding") === "gzip"
+          ? gunzipSync(body)
+          : body;
+      expect(JSON.parse(decoded.toString())).toEqual({ daily: [] });
     }
   });
 
@@ -232,3 +237,127 @@ describe("Grimoire invitation cache isolation", () => {
     },
   );
 });
+
+describe("streaming compression", () => {
+  it.each([
+    ["gzip", gunzipSync],
+    ["br", brotliDecompressSync],
+  ])(
+    "streams %s without consuming the whole upstream body",
+    async (encoding, decode) => {
+      let pulled = 0;
+      const chunk = new TextEncoder().encode("response ".repeat(32768));
+      const body = new ReadableStream({
+        pull(controller) {
+          if (pulled === 32) controller.close();
+          else {
+            pulled++;
+            controller.enqueue(chunk);
+          }
+        },
+      });
+      const response = await handle({
+        event: {
+          request: new Request("https://example.com", {
+            headers: { "accept-encoding": encoding },
+          }),
+        },
+        resolve: async () =>
+          new Response(body, {
+            headers: {
+              "content-type": "text/plain",
+              "content-length": String(32 * chunk.length),
+            },
+          }),
+      });
+      expect(pulled).toBeLessThan(32);
+      expect(response.headers.has("content-length")).toBe(false);
+      const decoded = decode(Buffer.from(await response.arrayBuffer()));
+      expect(decoded.length).toBe(32 * chunk.length);
+      expect(decoded.subarray(0, chunk.length)).toEqual(Buffer.from(chunk));
+    },
+  );
+
+  it("cancels upstream when the downstream reader disconnects", async () => {
+    let cancelled = false;
+    const body = new ReadableStream({
+      pull(controller) {
+        controller.enqueue(new Uint8Array(65536));
+      },
+      cancel() {
+        cancelled = true;
+      },
+    });
+    const response = await handle({
+      event: {
+        request: new Request("https://example.com", {
+          headers: { "accept-encoding": "gzip" },
+        }),
+      },
+      resolve: async () =>
+        new Response(body, { headers: { "content-type": "text/plain" } }),
+    });
+    await response.body.cancel();
+    await new Promise((resolve) => setImmediate(resolve));
+    expect(cancelled).toBe(true);
+  });
+
+  it("propagates an upstream failure to the response reader", async () => {
+    const body = new ReadableStream({
+      pull(controller) {
+        controller.error(new Error("upstream failed"));
+      },
+    });
+    const response = await handle({
+      event: {
+        request: new Request("https://example.com", {
+          headers: { "accept-encoding": "gzip" },
+        }),
+      },
+      resolve: async () =>
+        new Response(body, { headers: { "content-type": "text/plain" } }),
+    });
+    await expect(response.arrayBuffer()).rejects.toThrow("upstream failed");
+  });
+});
+
+describe("compression exclusions", () => {
+  it.each([
+    ["HEAD", "text/plain", {}],
+    ["GET", "text/event-stream", {}],
+    ["GET", "image/png", {}],
+    ["GET", "text/plain", { "content-encoding": "gzip" }],
+    ["GET", "text/plain", { "content-length": "5" }],
+  ])("passes through %s %s %j", async (method, type, headers) => {
+    const original = new Response("hello", {
+      headers: { "content-type": type, ...headers },
+    });
+    const response = await handle({
+      event: {
+        request: new Request("https://example.com", {
+          method,
+          headers: { "accept-encoding": "br" },
+        }),
+      },
+      resolve: async () => original,
+    });
+    expect(response).toBe(original);
+    expect(await response.text()).toBe("hello");
+  });
+
+  it("preserves a bodyless response", async () => {
+    const original = new Response(null, {
+      status: 204,
+      headers: { "content-type": "text/plain" },
+    });
+    const response = await handle({
+      event: {
+        request: new Request("https://example.com", {
+          headers: { "accept-encoding": "gzip" },
+        }),
+      },
+      resolve: async () => original,
+    });
+    expect(response).toBe(original);
+  });
+});