Fix Bun per-message websocket compression

.changeset/bun-websocket-outgoing-compression.md

@@ -0,0 +1,7 @@
+---
+"@effect/platform-bun": patch
+---
+
+Compress outgoing Bun WebSocket messages when per-message deflate is configured and negotiated. Messages
+smaller than 1 KiB are left uncompressed, matching the default threshold used by Node's `ws` server.
+The threshold is configurable via the new `websocket.compressionThreshold` server option.

packages/platform/bun/src/BunHttpServer.ts

@@ -77,13 +77,27 @@ export type ServeOptions<R extends string> =
  * through, e.g.
  * `BunHttpServer.layer({ port: 3000, websocket: { perMessageDeflate: true } })`.
  *
+ * The `compressionThreshold` option controls the minimum message size in bytes
+ * that is compressed when per-message deflate is negotiated. It defaults to
+ * 1024, matching the default threshold of Node's `ws` server.
+ *
  * @category options
  * @since 4.0.0
  */
-export type WebSocketOptions = Omit<
-  Bun.WebSocketHandler<WebSocketContext>,
-  "open" | "message" | "close" | "drain" | "ping" | "pong" | "data" | "binaryType"
->
+export type WebSocketOptions =
+  & Omit<
+    Bun.WebSocketHandler<WebSocketContext>,
+    "open" | "message" | "close" | "drain" | "ping" | "pong" | "data" | "binaryType"
+  >
+  & {
+    /**
+     * The minimum message size in bytes that is compressed when per-message
+     * deflate is negotiated.
+     *
+     * @default 1024
+     */
+    readonly compressionThreshold?: number | undefined
+  }
 
 /**
  * Creates a scoped Bun `HttpServer` from `Bun.serve` options, stopping the server on scope finalization with optional graceful shutdown settings.
@@ -100,6 +114,7 @@ export const make = Effect.fnUntraced(
     }
   ) {
     const scope = yield* Effect.scope
+    const { compressionThreshold = MIN_COMPRESSIBLE_SIZE, ...websocket } = options.websocket ?? {}
     const handlerStack: Array<(request: Request, server: BunServer<WebSocketContext>) => Response | Promise<Response>> =
       [
         function(_request, _server) {
@@ -110,7 +125,7 @@ export const make = Effect.fnUntraced(
       ...options as ServeOptions<R>,
       fetch: handlerStack[0],
       websocket: {
-        ...options.websocket,
+        ...websocket,
         open(ws) {
           Deferred.doneUnsafe(ws.data.deferred, Exit.succeed(ws))
         },
@@ -161,7 +176,7 @@ export const make = Effect.fnUntraced(
             const context = Context.add(
               services,
               ServerRequest.HttpServerRequest,
-              new BunServerRequest(request, resolve, removeHost(request.url), server)
+              new BunServerRequest(request, resolve, removeHost(request.url), server, compressionThreshold)
             )
             const fiber = Fiber.runIn(Effect.runForkWith(context)(httpEffect), scope)
             request.signal.addEventListener("abort", () => {
@@ -183,6 +198,8 @@ export const make = Effect.fnUntraced(
   }
 )
 
+const MIN_COMPRESSIBLE_SIZE = 1024
+
 const makeResponse = (
   request: ServerRequest.HttpServerRequest,
   response: ServerResponse.HttpServerResponse,
@@ -354,6 +371,7 @@ class BunServerRequest extends Inspectable.Class implements ServerRequest.HttpSe
   public resolve: (response: Response) => void
   readonly url: string
   private bunServer: BunServer<WebSocketContext>
+  private compressionThreshold: number
   public headersOverride?: Headers.Headers | undefined
   private remoteAddressOverride?: Option.Option<string> | undefined
 
@@ -362,6 +380,7 @@ class BunServerRequest extends Inspectable.Class implements ServerRequest.HttpSe
     resolve: (response: Response) => void,
     url: string,
     bunServer: BunServer<WebSocketContext>,
+    compressionThreshold: number,
     headersOverride?: Headers.Headers,
     remoteAddressOverride?: Option.Option<string>
   ) {
@@ -372,6 +391,7 @@ class BunServerRequest extends Inspectable.Class implements ServerRequest.HttpSe
     this.resolve = resolve
     this.url = url
     this.bunServer = bunServer
+    this.compressionThreshold = compressionThreshold
     this.headersOverride = headersOverride
     this.remoteAddressOverride = remoteAddressOverride
   }
@@ -394,6 +414,7 @@ class BunServerRequest extends Inspectable.Class implements ServerRequest.HttpSe
       this.resolve,
       options.url ?? this.url,
       this.bunServer,
+      this.compressionThreshold,
       options.headers ?? this.headersOverride,
       "remoteAddress" in options ? options.remoteAddress : this.remoteAddressOverride
     )
@@ -565,11 +586,11 @@ class BunServerRequest extends Inspectable.Class implements ServerRequest.HttpSe
         const write = (chunk: Uint8Array | string | Socket.CloseEvent) =>
           Effect.sync(() => {
             if (typeof chunk === "string") {
-              ws.sendText(chunk)
+              ws.sendText(chunk, chunk.length >= this.compressionThreshold)
             } else if (Socket.isCloseEvent(chunk)) {
               ws.close(chunk.code, chunk.reason)
             } else {
-              ws.sendBinary(chunk)
+              ws.sendBinary(chunk, chunk.byteLength >= this.compressionThreshold)
             }
 
             return true

packages/platform/bun/test/BunHttpServer.test.ts

@@ -4,11 +4,101 @@ import * as Effect from "effect/Effect"
 import * as Exit from "effect/Exit"
 import * as Scope from "effect/Scope"
 import * as HttpServer from "effect/unstable/http/HttpServer"
+import * as HttpServerRequest from "effect/unstable/http/HttpServerRequest"
 import * as HttpServerResponse from "effect/unstable/http/HttpServerResponse"
+import * as Net from "node:net"
 
 const fetchText = (url: string) =>
   Effect.promise(() => fetch(url, { headers: { connection: "close" } }).then((response) => response.text()))
 
+interface WebSocketFrame {
+  readonly opcode: number
+  readonly payload: Uint8Array
+  readonly payloadLength: number
+  readonly rsv1: boolean
+}
+
+interface WebSocketFrames {
+  readonly frames: ReadonlyArray<WebSocketFrame>
+  readonly headers: string
+}
+
+const readWebSocketFrames = (port: number, perMessageDeflate: boolean) =>
+  Effect.callback<WebSocketFrames, Error>((resume) => {
+    const socket = Net.createConnection({ host: "127.0.0.1", port })
+    let received = Buffer.alloc(0)
+    let result: WebSocketFrames | undefined
+
+    socket.on("connect", () => {
+      socket.write([
+        "GET / HTTP/1.1",
+        `Host: 127.0.0.1:${port}`,
+        "Connection: Upgrade",
+        "Upgrade: websocket",
+        "Sec-WebSocket-Version: 13",
+        "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==",
+        ...(perMessageDeflate ? ["Sec-WebSocket-Extensions: permessage-deflate"] : []),
+        "",
+        ""
+      ].join("\r\n"))
+    })
+    socket.on("error", (error) => resume(Effect.fail(error)))
+    socket.on("close", () => {
+      if (result) resume(Effect.succeed(result))
+    })
+    socket.on("data", (chunk) => {
+      if (result) return
+      received = Buffer.concat([received, typeof chunk === "string" ? Buffer.from(chunk) : chunk])
+      const headerEnd = received.indexOf("\r\n\r\n")
+      if (headerEnd === -1) return
+
+      const headers = received.subarray(0, headerEnd).toString()
+      const frames: Array<WebSocketFrame> = []
+      let offset = headerEnd + 4
+      while (frames.length < 2) {
+        if (received.length < offset + 2) return
+        const first = received[offset]
+        const length = received[offset + 1] & 0x7f
+        const headerLength = length === 126 ? 4 : 2
+        if (received.length < offset + headerLength) return
+        const payloadLength = length === 126 ? received.readUInt16BE(offset + 2) : length
+        if (received.length < offset + headerLength + payloadLength) return
+        frames.push({
+          opcode: first & 0x0f,
+          payload: Uint8Array.from(received.subarray(offset + headerLength, offset + headerLength + payloadLength)),
+          payloadLength,
+          rsv1: (first & 0x40) !== 0
+        })
+        offset += headerLength + payloadLength
+      }
+      result = { frames, headers }
+      socket.write(Buffer.from([0x88, 0x82, 0, 0, 0, 0, 0x03, 0xe8]))
+    })
+
+    return Effect.sync(() => socket.destroy())
+  })
+
+const makeWebSocketServer = Effect.fnUntraced(function*(payload: string, compressionThreshold?: number) {
+  const server = yield* BunHttpServer.make({
+    hostname: "127.0.0.1",
+    port: 0,
+    websocket: { perMessageDeflate: true, compressionThreshold }
+  })
+  yield* server.serve(Effect.gen(function*() {
+    const request = yield* HttpServerRequest.HttpServerRequest
+    const socket = yield* request.upgrade
+    const write = yield* socket.writer
+    yield* socket.runRaw(() => undefined, {
+      onOpen: Effect.gen(function*() {
+        yield* Effect.orDie(write(payload))
+        yield* Effect.orDie(write(new TextEncoder().encode(payload)))
+      })
+    })
+    return HttpServerResponse.empty()
+  }))
+  return server
+})
+
 describe("BunHttpServer", () => {
   it.effect("closing an older serve scope keeps the newer handler active", () =>
     Effect.gen(function*() {
@@ -47,4 +137,56 @@ describe("BunHttpServer", () => {
       yield* Scope.close(secondScope, Exit.void)
       assert.strictEqual(yield* fetchText(url), "first")
     }))
+
+  it.effect("compresses outgoing WebSocket messages when per-message deflate is negotiated", () =>
+    Effect.gen(function*() {
+      const payload = "a".repeat(4_096)
+      const server = yield* makeWebSocketServer(payload)
+      const port = (server.address as HttpServer.TcpAddress).port
+      const { frames, headers } = yield* readWebSocketFrames(port, true)
+
+      assert.match(headers, /^sec-websocket-extensions:.*permessage-deflate/im)
+      assert.deepStrictEqual(frames.map((frame) => frame.opcode), [1, 2])
+      assert.isTrue(frames.every((frame) => frame.rsv1))
+      assert.isTrue(frames.every((frame) => frame.payloadLength < payload.length))
+    }))
+
+  it.effect("leaves small WebSocket messages uncompressed even when per-message deflate is negotiated", () =>
+    Effect.gen(function*() {
+      const payload = "a".repeat(64)
+      const server = yield* makeWebSocketServer(payload)
+      const port = (server.address as HttpServer.TcpAddress).port
+      const { frames, headers } = yield* readWebSocketFrames(port, true)
+
+      assert.match(headers, /^sec-websocket-extensions:.*permessage-deflate/im)
+      assert.deepStrictEqual(frames.map((frame) => frame.opcode), [1, 2])
+      assert.isFalse(frames.some((frame) => frame.rsv1))
+      assert.isTrue(frames.every((frame) => frame.payloadLength === payload.length))
+    }))
+
+  it.effect("compresses small WebSocket messages when below a custom compressionThreshold", () =>
+    Effect.gen(function*() {
+      const payload = "a".repeat(64)
+      const server = yield* makeWebSocketServer(payload, 32)
+      const port = (server.address as HttpServer.TcpAddress).port
+      const { frames } = yield* readWebSocketFrames(port, true)
+
+      assert.deepStrictEqual(frames.map((frame) => frame.opcode), [1, 2])
+      assert.isTrue(frames.every((frame) => frame.rsv1))
+      assert.isTrue(frames.every((frame) => frame.payloadLength < payload.length))
+    }))
+
+  it.effect("supports WebSocket clients without per-message deflate", () =>
+    Effect.gen(function*() {
+      const payload = "a".repeat(4_096)
+      const server = yield* makeWebSocketServer(payload)
+      const port = (server.address as HttpServer.TcpAddress).port
+      const { frames, headers } = yield* readWebSocketFrames(port, false)
+
+      assert.notMatch(headers, /^sec-websocket-extensions:.*permessage-deflate/im)
+      assert.deepStrictEqual(frames.map((frame) => frame.opcode), [1, 2])
+      assert.isFalse(frames.some((frame) => frame.rsv1))
+      assert.isTrue(frames.every((frame) => frame.payloadLength === payload.length))
+      assert.deepStrictEqual(frames.map((frame) => new TextDecoder().decode(frame.payload)), [payload, payload])
+    }))
 })