perf(hub): gzip the SSE stream without delaying delivery (#1231)

* perf(hub): gzip the SSE stream without delaying delivery

SSE payloads are plain JSON that repeat the same field names on every
event, so they compress well - measured 72-77% on real captured traffic
from a hub with 15 active sessions.

Compression could not simply be turned on, though. Hono's compress()
middleware bails out whenever Transfer-Encoding is set, which streamSSE
always sets, so mounting it is a no-op. Wrapping the body in a
CompressionStream does compress, but it buffers until the stream ends -
measured on a 10-event stream, every event arrived at once when the
stream closed. On a connection that stays open for hours that means
events never arrive at all.

So drive zlib directly and issue a Z_SYNC_FLUSH after each chunk. That
costs about one percentage point of ratio and keeps delivery immediate:
verified in a real Chromium EventSource, first event at 13ms and each
subsequent event at its own 500ms tick, with no error events.

Clients that do not send Accept-Encoding: gzip keep the uncompressed
stream. No event payload or timing changes.

* fix(hub): cancel through the reader, gate reads on demand, honour q=0

Three defects in the first version of the SSE gzip wrapper:

Cancelling the source directly threw. The wrapper holds a reader for the
whole life of the connection, and cancelling a locked stream is invalid -
in Bun it throws TypeError: Invalid state: ReadableStream is locked
synchronously out of the cancel callback. Since SSE clients disconnect
mid-stream as a matter of course, this fired on essentially every
disconnect, and the upstream cancel never ran. Cancel through the reader
instead, which is allowed to.

Reads were not gated on downstream demand. Only zlib's own buffer was
consulted, and SSE compresses well enough that a slow client can be
megabytes behind while the compressed queue still looks nearly empty: a
test with a non-reading consumer pulled 1752 chunks before stalling.
Reading now waits for desiredSize to go positive, resumed from pull().

Accept-Encoding was matched with a substring test, so "gzip;q=0" - which
means the client refuses gzip - was read as acceptance. Parse the q-value.

Re-verified that none of this costs the property the change exists for:
in a real Chromium EventSource the first event still arrives at 13ms and
each one after it on its own 500ms tick, with no error events.

hub/src/web/routes/events.ts

@@ -7,6 +7,7 @@ import type { SyncEngine } from '../../sync/syncEngine'
 import type { VisibilityState } from '../../visibility/visibilityTracker'
 import type { VisibilityTracker } from '../../visibility/visibilityTracker'
 import type { WebAppEnv } from '../middleware/auth'
+import { compressSseResponse } from '../sseCompression'
 import { requireSession } from './guards'
 
 function parseOptionalId(value: string | undefined): string | null {
@@ -77,7 +78,7 @@ export function createEventsRoutes(
             }
         }
 
-        return streamSSE(c, async (stream) => {
+        const response = streamSSE(c, async (stream) => {
             manager.subscribe({
                 id: subscriptionId,
                 namespace,
@@ -117,6 +118,8 @@ export function createEventsRoutes(
 
             manager.unsubscribe(subscriptionId)
         })
+
+        return compressSseResponse(response, c.req.header('Accept-Encoding'))
     })
 
     app.post('/visibility', async (c) => {

hub/src/web/sseCompression.test.ts

@@ -0,0 +1,211 @@
+import { describe, expect, it } from 'bun:test'
+import zlib from 'node:zlib'
+import { compressSseResponse } from './sseCompression'
+
+function gunzip(data: Uint8Array): string {
+    return zlib.gunzipSync(Buffer.from(data)).toString('utf8')
+}
+
+function sseResponse(chunks: string[]): Response {
+    const encoder = new TextEncoder()
+    const body = new ReadableStream<Uint8Array>({
+        async start(controller) {
+            for (const chunk of chunks) {
+                controller.enqueue(encoder.encode(chunk))
+            }
+            controller.close()
+        }
+    })
+    return new Response(body, {
+        headers: {
+            'Content-Type': 'text/event-stream',
+            'Cache-Control': 'no-cache'
+        }
+    })
+}
+
+async function readAll(response: Response): Promise<Uint8Array> {
+    const reader = response.body!.getReader()
+    const parts: Uint8Array[] = []
+    for (;;) {
+        const { done, value } = await reader.read()
+        if (done) {
+            break
+        }
+        parts.push(value)
+    }
+    const total = parts.reduce((sum, part) => sum + part.length, 0)
+    const merged = new Uint8Array(total)
+    let offset = 0
+    for (const part of parts) {
+        merged.set(part, offset)
+        offset += part.length
+    }
+    return merged
+}
+
+describe('compressSseResponse', () => {
+    it('leaves the response untouched when the client does not accept gzip', async () => {
+        const original = sseResponse(['data: {"a":1}\n\n'])
+        const result = compressSseResponse(original, undefined)
+
+        expect(result).toBe(original)
+        expect(result.headers.get('Content-Encoding')).toBeNull()
+    })
+
+    it('leaves the response untouched when Accept-Encoding lacks gzip', async () => {
+        const original = sseResponse(['data: {"a":1}\n\n'])
+        const result = compressSseResponse(original, 'br, deflate')
+
+        expect(result).toBe(original)
+    })
+
+    it('gzips the stream and preserves the exact payload', async () => {
+        const payload = 'data: {"type":"heartbeat","data":{"timestamp":1}}\n\n'
+        const result = compressSseResponse(sseResponse([payload]), 'gzip, deflate')
+
+        expect(result.headers.get('Content-Encoding')).toBe('gzip')
+        expect(result.headers.get('Content-Type')).toBe('text/event-stream')
+        expect(result.headers.get('Cache-Control')).toBe('no-cache')
+
+        expect(gunzip(await readAll(result))).toBe(payload)
+    })
+
+    it('flushes every event immediately instead of buffering until the stream ends', async () => {
+        // A SSE connection stays open for hours. If the compressor buffers,
+        // events never reach the client. Each written event must produce
+        // decompressible output before the stream closes.
+        const encoder = new TextEncoder()
+        let emit!: (value: string) => void
+        let finish!: () => void
+        const body = new ReadableStream<Uint8Array>({
+            start(controller) {
+                emit = (value) => controller.enqueue(encoder.encode(value))
+                finish = () => controller.close()
+            }
+        })
+        const source = new Response(body, { headers: { 'Content-Type': 'text/event-stream' } })
+        const result = compressSseResponse(source, 'gzip')
+        const reader = result.body!.getReader()
+        const inflate = zlib.createGunzip()
+        const seen: string[] = []
+        inflate.on('data', (chunk: Buffer) => seen.push(chunk.toString('utf8')))
+
+        emit('data: {"seq":1}\n\n')
+        const first = await reader.read()
+        expect(first.done).toBe(false)
+        inflate.write(Buffer.from(first.value!))
+        await new Promise<void>((resolve) => inflate.flush(() => resolve()))
+        expect(seen.join('')).toBe('data: {"seq":1}\n\n')
+
+        emit('data: {"seq":2}\n\n')
+        const second = await reader.read()
+        expect(second.done).toBe(false)
+        inflate.write(Buffer.from(second.value!))
+        await new Promise<void>((resolve) => inflate.flush(() => resolve()))
+        expect(seen.join('')).toBe('data: {"seq":1}\n\ndata: {"seq":2}\n\n')
+
+        finish()
+    })
+
+    it('passes through a response without a body', () => {
+        const original = new Response(null, { status: 204 })
+        expect(compressSseResponse(original, 'gzip')).toBe(original)
+    })
+})
+
+describe('compressSseResponse cleanup', () => {
+    it('cancels the upstream stream when the client goes away', async () => {
+        // SSE clients disconnect mid-stream all the time; the source must be
+        // told, or the subscription behind it leaks.
+        let cancelledWith: unknown = Symbol('never')
+        const body = new ReadableStream<Uint8Array>({
+            start(controller) {
+                controller.enqueue(new TextEncoder().encode('data: {"a":1}\n\n'))
+            },
+            cancel(reason) {
+                cancelledWith = reason
+            }
+        })
+        const result = compressSseResponse(
+            new Response(body, { headers: { 'Content-Type': 'text/event-stream' } }),
+            'gzip'
+        )
+        const reader = result.body!.getReader()
+        await reader.read()
+        await reader.cancel('client gone')
+        await new Promise((resolve) => setTimeout(resolve, 20))
+
+        expect(cancelledWith).toBe('client gone')
+    })
+
+    it('does not reject when the client cancels', async () => {
+        const body = new ReadableStream<Uint8Array>({
+            start(controller) {
+                controller.enqueue(new TextEncoder().encode('data: {"a":1}\n\n'))
+            }
+        })
+        const result = compressSseResponse(
+            new Response(body, { headers: { 'Content-Type': 'text/event-stream' } }),
+            'gzip'
+        )
+        const rejections: unknown[] = []
+        const onRejection = (reason: unknown) => rejections.push(reason)
+        process.on('unhandledRejection', onRejection)
+
+        const reader = result.body!.getReader()
+        await reader.read()
+        await reader.cancel('client gone')
+        await new Promise((resolve) => setTimeout(resolve, 50))
+        process.off('unhandledRejection', onRejection)
+
+        expect(rejections).toEqual([])
+    })
+})
+
+describe('compressSseResponse content negotiation', () => {
+    it('honours an explicit q=0 refusal', () => {
+        const original = sseResponse(['data: {"a":1}\n\n'])
+        expect(compressSseResponse(original, 'gzip;q=0, deflate')).toBe(original)
+    })
+
+    it('still compresses when a q-value is present but non-zero', () => {
+        const result = compressSseResponse(sseResponse(['data: {"a":1}\n\n']), 'gzip;q=0.5')
+        expect(result.headers.get('Content-Encoding')).toBe('gzip')
+    })
+
+    it('compresses for a wildcard accept', () => {
+        const result = compressSseResponse(sseResponse(['data: {"a":1}\n\n']), '*')
+        expect(result.headers.get('Content-Encoding')).toBe('gzip')
+    })
+})
+
+describe('compressSseResponse backpressure', () => {
+    it('stops pulling from the source while the consumer is not reading', async () => {
+        // A slow client must not make the hub buffer without bound.
+        let produced = 0
+        const encoder = new TextEncoder()
+        const body = new ReadableStream<Uint8Array>({
+            pull(controller) {
+                produced += 1
+                controller.enqueue(encoder.encode(`data: {"seq":${produced},"pad":"${'x'.repeat(4000)}"}\n\n`))
+            }
+        })
+        const result = compressSseResponse(
+            new Response(body, { headers: { 'Content-Type': 'text/event-stream' } }),
+            'gzip'
+        )
+        const reader = result.body!.getReader()
+        await reader.read()
+        // Consumer goes quiet; production must not run away.
+        await new Promise((resolve) => setTimeout(resolve, 120))
+        const idle = produced
+
+        await reader.read()
+        await new Promise((resolve) => setTimeout(resolve, 60))
+
+        expect(idle).toBeLessThan(200)
+        expect(produced).toBeGreaterThanOrEqual(idle)
+        await reader.cancel('done')
+    })
+})

hub/src/web/sseCompression.ts

@@ -0,0 +1,128 @@
+import zlib from 'node:zlib'
+
+/**
+ * True when the client is willing to receive gzip.
+ *
+ * `Accept-Encoding: gzip;q=0` means the opposite of what a substring match
+ * would suggest, so parse the q-value rather than looking for the word.
+ */
+function acceptsGzip(acceptEncoding: string | undefined): boolean {
+    if (!acceptEncoding) {
+        return false
+    }
+    for (const part of acceptEncoding.split(',')) {
+        const [rawName, ...params] = part.split(';')
+        const name = rawName?.trim().toLowerCase()
+        if (name !== 'gzip' && name !== '*') {
+            continue
+        }
+        const q = params
+            .map((param) => param.trim().toLowerCase())
+            .find((param) => param.startsWith('q='))
+        if (q && Number(q.slice(2)) === 0) {
+            return false
+        }
+        return true
+    }
+    return false
+}
+
+/**
+ * Wraps an SSE response in a gzip stream when the client accepts it.
+ *
+ * SSE payloads are plain JSON with the same field names repeated on every
+ * event, so they compress extremely well (~75% on real traffic). The catch is
+ * that the standard compressors buffer: `CompressionStream` and Hono's
+ * `compress()` middleware only emit once the stream ends, which for a
+ * connection that stays open for hours means events never arrive. We therefore
+ * drive zlib directly and issue a Z_SYNC_FLUSH after every chunk, which costs
+ * about one percentage point of ratio and keeps delivery immediate.
+ *
+ * (Hono's `compress()` would skip this response anyway - it bails out when
+ * `Transfer-Encoding` is set, and `streamSSE` always sets it.)
+ */
+export function compressSseResponse(response: Response, acceptEncoding: string | undefined): Response {
+    if (!acceptsGzip(acceptEncoding) || !response.body) {
+        return response
+    }
+
+    const gzip = zlib.createGzip({ flush: zlib.constants.Z_SYNC_FLUSH })
+    const source = response.body
+    // Acquired synchronously so `cancel` can always reach it. Cancelling
+    // `source` directly would throw: it is locked for as long as we hold a
+    // reader, which is the whole lifetime of the connection.
+    const reader = source.getReader()
+    let resumeRead: (() => void) | null = null
+
+    const compressed = new ReadableStream<Uint8Array>({
+        start(controller) {
+            // Gate reading on downstream demand. Checking only zlib's own
+            // buffer is not enough: SSE compresses so well that a slow client
+            // can be megabytes behind while the compressed queue still looks
+            // nearly empty.
+            const awaitDemand = (): Promise<void> => {
+                if ((controller.desiredSize ?? 1) > 0) {
+                    return Promise.resolve()
+                }
+                return new Promise<void>((resolve) => {
+                    resumeRead = resolve
+                })
+            }
+
+            gzip.on('data', (chunk: Buffer) => {
+                controller.enqueue(new Uint8Array(chunk))
+                if ((controller.desiredSize ?? 1) <= 0) {
+                    gzip.pause()
+                }
+            })
+            gzip.on('end', () => {
+                controller.close()
+            })
+            gzip.on('error', (error) => {
+                controller.error(error)
+            })
+
+            void (async () => {
+                try {
+                    for (;;) {
+                        await awaitDemand()
+                        const { done, value } = await reader.read()
+                        if (done) {
+                            break
+                        }
+                        if (!gzip.write(Buffer.from(value))) {
+                            await new Promise<void>((resolve) => gzip.once('drain', resolve))
+                        }
+                        gzip.flush(zlib.constants.Z_SYNC_FLUSH)
+                    }
+                } catch (error) {
+                    gzip.destroy(error as Error)
+                    return
+                }
+                gzip.end()
+            })()
+        },
+        pull() {
+            gzip.resume()
+            const resume = resumeRead
+            resumeRead = null
+            resume?.()
+        },
+        cancel(reason) {
+            gzip.destroy()
+            resumeRead?.()
+            resumeRead = null
+            return reader.cancel(reason)
+        }
+    })
+
+    const headers = new Headers(response.headers)
+    headers.set('Content-Encoding', 'gzip')
+    headers.delete('Content-Length')
+
+    return new Response(compressed, {
+        status: response.status,
+        statusText: response.statusText,
+        headers
+    })
+}