From 28402b732919cca8e4486913cd3f62223533c0a9 Mon Sep 17 00:00:00 2001 From: Phillip Bronzel Date: Sat, 22 Aug 2026 12:23:26 +0200 Subject: [PATCH 1/2] fix: bun per-message websocket compression --- .../bun-websocket-outgoing-compression.md | 5 + packages/platform/bun/src/BunHttpServer.ts | 4 +- .../platform/bun/test/BunHttpServer.test.ts | 117 ++++++++++++++++++ 3 files changed, 124 insertions(+), 2 deletions(-) create mode 100644 .changeset/bun-websocket-outgoing-compression.md diff --git a/.changeset/bun-websocket-outgoing-compression.md b/.changeset/bun-websocket-outgoing-compression.md new file mode 100644 index 00000000000..bafbb905645 --- /dev/null +++ b/.changeset/bun-websocket-outgoing-compression.md @@ -0,0 +1,5 @@ +--- +"@effect/platform-bun": patch +--- + +Compress outgoing Bun WebSocket messages when per-message deflate is configured and negotiated. diff --git a/packages/platform/bun/src/BunHttpServer.ts b/packages/platform/bun/src/BunHttpServer.ts index 4ccfcf16b8b..4f705a1dd14 100644 --- a/packages/platform/bun/src/BunHttpServer.ts +++ b/packages/platform/bun/src/BunHttpServer.ts @@ -565,11 +565,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, true) } else if (Socket.isCloseEvent(chunk)) { ws.close(chunk.code, chunk.reason) } else { - ws.sendBinary(chunk) + ws.sendBinary(chunk, true) } return true diff --git a/packages/platform/bun/test/BunHttpServer.test.ts b/packages/platform/bun/test/BunHttpServer.test.ts index ca1e8d11283..28e926e3e8d 100644 --- a/packages/platform/bun/test/BunHttpServer.test.ts +++ b/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 + readonly headers: string +} + +const readWebSocketFrames = (port: number, perMessageDeflate: boolean) => + Effect.callback((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 = [] + 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) { + const server = yield* BunHttpServer.make({ + hostname: "127.0.0.1", + port: 0, + websocket: { perMessageDeflate: true } + }) + 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,31 @@ 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("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]) + })) }) From 5d9905629a38b5c49199e3008405a7376e356151 Mon Sep 17 00:00:00 2001 From: Phillip Bronzel Date: Mon, 24 Aug 2026 08:05:20 +0200 Subject: [PATCH 2/2] fix: add compressionThreshold --- .../bun-websocket-outgoing-compression.md | 4 +- packages/platform/bun/src/BunHttpServer.ts | 37 +++++++++++++++---- .../platform/bun/test/BunHttpServer.test.ts | 29 ++++++++++++++- 3 files changed, 59 insertions(+), 11 deletions(-) diff --git a/.changeset/bun-websocket-outgoing-compression.md b/.changeset/bun-websocket-outgoing-compression.md index bafbb905645..9be7aeaf63f 100644 --- a/.changeset/bun-websocket-outgoing-compression.md +++ b/.changeset/bun-websocket-outgoing-compression.md @@ -2,4 +2,6 @@ "@effect/platform-bun": patch --- -Compress outgoing Bun WebSocket messages when per-message deflate is configured and negotiated. +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. diff --git a/packages/platform/bun/src/BunHttpServer.ts b/packages/platform/bun/src/BunHttpServer.ts index 4f705a1dd14..4ce02e91b12 100644 --- a/packages/platform/bun/src/BunHttpServer.ts +++ b/packages/platform/bun/src/BunHttpServer.ts @@ -77,13 +77,27 @@ export type ServeOptions = * 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, - "open" | "message" | "close" | "drain" | "ping" | "pong" | "data" | "binaryType" -> +export type WebSocketOptions = + & Omit< + Bun.WebSocketHandler, + "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) => Response | Promise> = [ function(_request, _server) { @@ -110,7 +125,7 @@ export const make = Effect.fnUntraced( ...options as ServeOptions, 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 + private compressionThreshold: number public headersOverride?: Headers.Headers | undefined private remoteAddressOverride?: Option.Option | undefined @@ -362,6 +380,7 @@ class BunServerRequest extends Inspectable.Class implements ServerRequest.HttpSe resolve: (response: Response) => void, url: string, bunServer: BunServer, + compressionThreshold: number, headersOverride?: Headers.Headers, remoteAddressOverride?: Option.Option ) { @@ -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, true) + ws.sendText(chunk, chunk.length >= this.compressionThreshold) } else if (Socket.isCloseEvent(chunk)) { ws.close(chunk.code, chunk.reason) } else { - ws.sendBinary(chunk, true) + ws.sendBinary(chunk, chunk.byteLength >= this.compressionThreshold) } return true diff --git a/packages/platform/bun/test/BunHttpServer.test.ts b/packages/platform/bun/test/BunHttpServer.test.ts index 28e926e3e8d..bbbbdba54db 100644 --- a/packages/platform/bun/test/BunHttpServer.test.ts +++ b/packages/platform/bun/test/BunHttpServer.test.ts @@ -78,11 +78,11 @@ const readWebSocketFrames = (port: number, perMessageDeflate: boolean) => return Effect.sync(() => socket.destroy()) }) -const makeWebSocketServer = Effect.fnUntraced(function*(payload: string) { +const makeWebSocketServer = Effect.fnUntraced(function*(payload: string, compressionThreshold?: number) { const server = yield* BunHttpServer.make({ hostname: "127.0.0.1", port: 0, - websocket: { perMessageDeflate: true } + websocket: { perMessageDeflate: true, compressionThreshold } }) yield* server.serve(Effect.gen(function*() { const request = yield* HttpServerRequest.HttpServerRequest @@ -151,6 +151,31 @@ describe("BunHttpServer", () => { 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)