From 598c8032b7bb4c0671b4cc3aa73b40b4925f5db7 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Mon, 20 Jul 2026 12:28:49 +0000 Subject: [PATCH 1/7] uws: cursor-based BackPressure buffer (erase is a pointer bump, append compacts instead of reallocating) The std::string-backed BackPressure front-erased and shrink_to_fit() every time pendingRemoval crossed 1/32 of the buffer, so draining N bytes paid ~32 passes of memmove(remaining) + realloc(remaining) + memcpy(remaining), and each realloc briefly held old + new allocations. Replace it with a single malloc'd slab and head/tail cursors: - erase(n) bumps head; on full drain resets to 0,0 and frees. - append()/resize() reuse the drained head gap via one memmove before growing, and use realloc() when head==0 so the allocator can extend in place. - getBufferedAmount() reports unsent bytes (length()); memoryCost() reports allocation footprint (totalLength()). Adds integrity tests that push large direct and corked-frame sends through a backpressured ServerWebSocket and sha1-compare the drained bytes. --- packages/bun-uws/src/AsyncSocket.h | 9 +- packages/bun-uws/src/AsyncSocketData.h | 127 +++++++--- packages/bun-uws/src/WebSocket.h | 3 +- ...bsocket-server-backpressure-buffer.test.ts | 237 ++++++++++++++++++ 4 files changed, 337 insertions(+), 39 deletions(-) create mode 100644 test/js/bun/websocket/websocket-server-backpressure-buffer.test.ts diff --git a/packages/bun-uws/src/AsyncSocket.h b/packages/bun-uws/src/AsyncSocket.h index ffc6d3c77312..b8a3db918195 100644 --- a/packages/bun-uws/src/AsyncSocket.h +++ b/packages/bun-uws/src/AsyncSocket.h @@ -183,8 +183,9 @@ struct AsyncSocket { /* Returns the user space backpressure. */ size_t getBufferedAmount() { - /* We return the actual amount of bytes in backbuffer, including pendingRemoval */ - return getAsyncSocketData()->buffer.totalLength(); + /* Unsent bytes; already-written bytes sitting behind the head cursor + * are not backpressure (maxBackpressure checks, drain progress). */ + return getAsyncSocketData()->buffer.length(); } /* Returns the text representation of an IPv4 or IPv6 address */ @@ -253,7 +254,7 @@ struct AsyncSocket { /* Check if we couldn't write the entire buffer */ if ((unsigned int) written < buffer_len) { /* Remove the successfully written data from the buffer */ - asyncSocketData->buffer.erase((unsigned int) written); + asyncSocketData->buffer.erase((size_t) written); /* If we wrote less than we attempted, the socket buffer is likely full * likely is used as an optimization hint to the compiler @@ -301,7 +302,7 @@ struct AsyncSocket { /* On failure return, otherwise continue down the function */ if ((unsigned int) written < buffer_len) { /* Update buffering (todo: we can do better here if we keep track of what happens to this guy later on) */ - asyncSocketData->buffer.erase((unsigned int) written); + asyncSocketData->buffer.erase((size_t) written); if (optionally) { /* Thankfully we can exit early here */ diff --git a/packages/bun-uws/src/AsyncSocketData.h b/packages/bun-uws/src/AsyncSocketData.h index c36c30b5848e..9142b5f0908f 100644 --- a/packages/bun-uws/src/AsyncSocketData.h +++ b/packages/bun-uws/src/AsyncSocketData.h @@ -19,53 +19,112 @@ #ifndef UWS_ASYNCSOCKETDATA_H #define UWS_ASYNCSOCKETDATA_H -#include +#include +#include +#include namespace uWS { +/* Contiguous write-behind buffer with a moving head cursor. erase() is a + * pointer bump; append()/resize() reuse the drained head gap via one memmove + * before ever growing. The previous std::string shape front-erased + shrank to + * fit every ~1/32 drained, so a drain of N bytes moved ~60N bytes and briefly + * held 2x the live data during each realloc. */ struct BackPressure { - std::string buffer; - unsigned int pendingRemoval = 0; - BackPressure(BackPressure &&other) { - buffer = std::move(other.buffer); - pendingRemoval = other.pendingRemoval; - } BackPressure() = default; - void append(const char *data, size_t length) { - buffer.append(data, length); + BackPressure(BackPressure &&other) noexcept + : buf(other.buf), head(other.head), tail(other.tail), cap(other.cap) { + other.buf = nullptr; + other.head = other.tail = other.cap = 0; } - void erase(unsigned int length) { - pendingRemoval += length; - /* Always erase a minimum of 1/32th the current backpressure */ - if (pendingRemoval > (buffer.length() >> 5)) { - buffer.erase(0, pendingRemoval); - buffer.shrink_to_fit(); - pendingRemoval = 0; - } + BackPressure(const BackPressure &) = delete; + BackPressure &operator=(const BackPressure &) = delete; + ~BackPressure() { std::free(buf); } + + /* Unsent bytes. data() points at length() contiguous bytes. */ + size_t length() const { return tail - head; } + size_t size() const { return length(); } + const char *data() const { return buf + head; } + /* Allocation footprint for memoryCost / GC reporting. */ + size_t totalLength() const { return cap; } + + void append(const char *src, size_t n) { + if (!n) return; + ensureTailRoom(n); + std::memcpy(buf + tail, src, n); + tail += n; } - size_t length() { - return buffer.length() - pendingRemoval; + + void erase(size_t n) { + head += n; + if (head >= tail) { + /* Fully drained: next append writes at offset 0 with no memmove. */ + head = tail = 0; + release(); + } } + void clear() { - pendingRemoval = 0; - buffer.clear(); - buffer.shrink_to_fit(); - } - void reserve(size_t length) { - buffer.reserve(length + pendingRemoval); + head = tail = 0; + release(); } - void resize(size_t length) { - buffer.resize(length + pendingRemoval); + + /* Make room for at least n live bytes without later realloc. */ + void reserve(size_t n) { + if (n > length()) ensureTailRoom(n - length()); } - const char *data() { - return buffer.data() + pendingRemoval; + + /* Grow to n live bytes; caller writes into data() + old length(). */ + void resize(size_t n) { + size_t live = length(); + if (n > live) { + ensureTailRoom(n - live); + tail += n - live; + } else { + tail = head + n; + } } - size_t size() { - return length(); + +private: + static constexpr size_t MIN_CAPACITY = 4096; + + char *buf = nullptr; + size_t head = 0; + size_t tail = 0; + size_t cap = 0; + + /* Ensure [tail, tail+n) is writable. Prefers compacting into the drained + * head gap over growing so steady-state producer/consumer never reallocs. */ + void ensureTailRoom(size_t n) { + if (tail + n <= cap) return; + + size_t live = tail - head; + if (head && live + n <= cap) { + std::memmove(buf, buf + head, live); + head = 0; + tail = live; + return; + } + + size_t newCap = std::max(std::max(cap * 2, live + n), MIN_CAPACITY); + if (head == 0) { + /* realloc may extend in place (mimalloc, glibc mremap). */ + buf = (char *) std::realloc(buf, newCap); + } else { + char *nb = (char *) std::malloc(newCap); + if (live) std::memcpy(nb, buf + head, live); + std::free(buf); + buf = nb; + head = 0; + tail = live; + } + cap = newCap; } - /* The total length, incuding pending removal */ - size_t totalLength() { - return buffer.length(); + + void release() { + std::free(buf); + buf = nullptr; + cap = 0; } }; diff --git a/packages/bun-uws/src/WebSocket.h b/packages/bun-uws/src/WebSocket.h index c1e8b5e1f160..03231d2bec4f 100644 --- a/packages/bun-uws/src/WebSocket.h +++ b/packages/bun-uws/src/WebSocket.h @@ -106,7 +106,8 @@ struct WebSocket : AsyncSocket { } size_t memoryCost() { - return getBufferedAmount() + sizeof(WebSocket); + /* Allocation footprint for reportExtraMemoryAllocated, not unsent bytes. */ + return Super::getAsyncSocketData()->buffer.totalLength() + sizeof(WebSocket); } /* Sending fragmented messages puts a bit of effort on the user; you must not interleave regular sends diff --git a/test/js/bun/websocket/websocket-server-backpressure-buffer.test.ts b/test/js/bun/websocket/websocket-server-backpressure-buffer.test.ts new file mode 100644 index 000000000000..225ec5567cc5 --- /dev/null +++ b/test/js/bun/websocket/websocket-server-backpressure-buffer.test.ts @@ -0,0 +1,237 @@ +import { describe, expect, it } from "bun:test"; +import crypto from "node:crypto"; +import net from "node:net"; + +// Drives the uws BackPressure buffer through its append / erase / resize paths +// and verifies the bytes that reach the client exactly match what was sent. + +function patternBuffer(len: number, seed: number): Buffer { + const b = Buffer.allocUnsafe(len); + let x = seed | 1; + for (let i = 0; i < len; i++) { + x = (x * 1103515245 + 12345) >>> 0; + b[i] = x >>> 24; + } + return b; +} + +// Resolves once the raw socket has completed the WS upgrade and paused, so the +// server's outgoing writes land in the BackPressure buffer. Returns the paused +// socket and any frame bytes that arrived after the handshake headers. +async function pausedClient(port: number): Promise<{ sock: net.Socket; initial: Buffer }> { + const sock = net.connect(port, "127.0.0.1"); + sock.on("error", () => {}); + const { promise, resolve, reject } = Promise.withResolvers(); + let buf = Buffer.alloc(0); + const onData = (d: Buffer) => { + buf = buf.length ? Buffer.concat([buf, d]) : d; + const i = buf.indexOf("\r\n\r\n"); + if (i < 0) return; + sock.pause(); + sock.off("data", onData); + if (!buf.subarray(0, i).toString("latin1").includes(" 101 ")) { + reject(new Error("upgrade failed: " + buf.subarray(0, i))); + return; + } + resolve(buf.subarray(i + 4)); + }; + sock.on("data", onData); + sock.on("connect", () => { + sock.write( + "GET / HTTP/1.1\r\n" + + `Host: 127.0.0.1:${port}\r\n` + + "Upgrade: websocket\r\n" + + "Connection: Upgrade\r\n" + + "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n" + + "Sec-WebSocket-Version: 13\r\n\r\n", + ); + }); + const initial = await promise; + return { sock, initial }; +} + +describe("BackPressure buffer", () => { + // >16KB sends take the direct us_socket_write2 path and then append() the + // unwritten tail into BackPressure; drain exercises erase() as a pure + // head-cursor bump. + it("delivers a large direct send byte-for-byte while draining", async () => { + const SIZE = 32 * 1024 * 1024; + const payload = patternBuffer(SIZE, 0xabcd); + const expectedHash = crypto.createHash("sha1").update(payload).digest("hex"); + + let bufferedAfterSend = 0; + let drainSawDecrease = true; + let prev = Infinity; + let serverWs: import("bun").ServerWebSocket | undefined; + const sentSignal = Promise.withResolvers(); + const drained = Promise.withResolvers(); + await using server = Bun.serve({ + port: 0, + fetch(req, s) { + if (s.upgrade(req)) return; + return new Response("no", { status: 500 }); + }, + websocket: { + maxBackpressure: SIZE * 2, + idleTimeout: 0, + open(ws) { + serverWs = ws; + ws.sendBinary(payload); + bufferedAfterSend = ws.getBufferedAmount(); + sentSignal.resolve(); + }, + drain(ws) { + const b = ws.getBufferedAmount(); + if (b > prev) drainSawDecrease = false; + prev = b; + if (b === 0) drained.resolve(); + }, + message() {}, + close() { + drained.resolve(); + }, + }, + }); + + const { sock, initial } = await pausedClient(server.port); + await sentSignal.promise; + + // 32MB cannot fit in the kernel send buffer, so a non-empty remainder must + // have been copied into the BackPressure buffer. + expect(bufferedAfterSend).toBeGreaterThan(0); + expect(bufferedAfterSend).toBeLessThanOrEqual(SIZE + 10); + + // Drain: read until we have the full 10-byte header + SIZE payload bytes. + const target = 10 + SIZE; + const hash = crypto.createHash("sha1"); + let received = 0; + const consume = (chunk: Buffer) => { + let off = 0; + while (off < chunk.length && received < target) { + if (received < 10) { + const skip = Math.min(10 - received, chunk.length - off); + received += skip; + off += skip; + continue; + } + const take = Math.min(target - received, chunk.length - off); + hash.update(chunk.subarray(off, off + take)); + received += take; + off += take; + } + }; + consume(initial); + const allReceived = Promise.withResolvers(); + if (received >= target) allReceived.resolve(); + sock.on("data", chunk => { + consume(chunk); + if (received >= target) allReceived.resolve(); + }); + sock.on("close", () => allReceived.resolve()); + sock.resume(); + + await allReceived.promise; + await drained.promise; + sock.destroy(); + + expect(drainSawDecrease).toBe(true); + expect(received).toBe(target); + expect(hash.digest("hex")).toBe(expectedHash); + }, 30_000); + + // Small (<16KB) sends take getSendBuffer(): once the cork buffer fills they + // hit BackPressure.resize() + the in-place tail write, then write(nullptr,0) + // drains (erase). Keeping the window full forces append()/resize() to reuse + // the drained head gap via a compact instead of reallocating. + it("delivers many corked frames while appending into a partly-drained buffer", async () => { + const FRAME = 4096; + const COUNT = 4096; // 16MB of payload through the cork->backpressure path + const WINDOW = 1 * 1024 * 1024; + const headerLen = 4; // server frame, 16-bit extended length, no mask + + const expected = crypto.createHash("sha1"); + const frames: Buffer[] = []; + for (let i = 0; i < COUNT; i++) { + const p = patternBuffer(FRAME, i); + frames.push(p); + expected.update(p); + } + const expectedHash = expected.digest("hex"); + + let sent = 0; + let sawBufferedAboveWindow = false; + const drained = Promise.withResolvers(); + const fill = (ws: import("bun").ServerWebSocket) => { + while (sent < COUNT) { + ws.sendBinary(frames[sent]); + sent++; + if (ws.getBufferedAmount() >= WINDOW) { + sawBufferedAboveWindow = true; + return; + } + } + if (ws.getBufferedAmount() === 0) drained.resolve(); + }; + await using server = Bun.serve({ + port: 0, + fetch(req, s) { + if (s.upgrade(req)) return; + return new Response("no", { status: 500 }); + }, + websocket: { + maxBackpressure: WINDOW * 4, + idleTimeout: 0, + open: fill, + drain: fill, + message() {}, + close() { + drained.resolve(); + }, + }, + }); + + const { sock, initial } = await pausedClient(server.port); + + const perFrame = headerLen + FRAME; + const target = COUNT * perFrame; + const hash = crypto.createHash("sha1"); + let received = 0; + let frameOff = 0; + const consume = (chunk: Buffer) => { + let off = 0; + while (off < chunk.length && received < target) { + if (frameOff < headerLen) { + const skip = Math.min(headerLen - frameOff, chunk.length - off); + frameOff += skip; + received += skip; + off += skip; + continue; + } + const take = Math.min(perFrame - frameOff, chunk.length - off); + hash.update(chunk.subarray(off, off + take)); + frameOff += take; + received += take; + off += take; + if (frameOff === perFrame) frameOff = 0; + } + }; + consume(initial); + const allReceived = Promise.withResolvers(); + if (received >= target) allReceived.resolve(); + sock.on("data", chunk => { + consume(chunk); + if (received >= target) allReceived.resolve(); + }); + sock.on("close", () => allReceived.resolve()); + sock.resume(); + + await allReceived.promise; + await drained.promise; + sock.destroy(); + + expect(sawBufferedAboveWindow).toBe(true); + expect(sent).toBe(COUNT); + expect(received).toBe(target); + expect(hash.digest("hex")).toBe(expectedHash); + }, 30_000); +}); From 4b0e43263b02f019e3229a0e73f158afab880c06 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Mon, 20 Jul 2026 12:42:07 +0000 Subject: [PATCH 2/7] review: abort on malloc/realloc failure, wire handshake failure to reject, drop unused local --- packages/bun-uws/src/AsyncSocketData.h | 9 ++++++--- .../websocket-server-backpressure-buffer.test.ts | 7 ++++--- 2 files changed, 10 insertions(+), 6 deletions(-) diff --git a/packages/bun-uws/src/AsyncSocketData.h b/packages/bun-uws/src/AsyncSocketData.h index 9142b5f0908f..b76f5117eb9f 100644 --- a/packages/bun-uws/src/AsyncSocketData.h +++ b/packages/bun-uws/src/AsyncSocketData.h @@ -107,17 +107,20 @@ struct BackPressure { } size_t newCap = std::max(std::max(cap * 2, live + n), MIN_CAPACITY); + char *nb; if (head == 0) { /* realloc may extend in place (mimalloc, glibc mremap). */ - buf = (char *) std::realloc(buf, newCap); + nb = (char *) std::realloc(buf, newCap); + if (!nb) std::abort(); } else { - char *nb = (char *) std::malloc(newCap); + nb = (char *) std::malloc(newCap); + if (!nb) std::abort(); if (live) std::memcpy(nb, buf + head, live); std::free(buf); - buf = nb; head = 0; tail = live; } + buf = nb; cap = newCap; } diff --git a/test/js/bun/websocket/websocket-server-backpressure-buffer.test.ts b/test/js/bun/websocket/websocket-server-backpressure-buffer.test.ts index 225ec5567cc5..b8a33bf0184f 100644 --- a/test/js/bun/websocket/websocket-server-backpressure-buffer.test.ts +++ b/test/js/bun/websocket/websocket-server-backpressure-buffer.test.ts @@ -20,8 +20,9 @@ function patternBuffer(len: number, seed: number): Buffer { // socket and any frame bytes that arrived after the handshake headers. async function pausedClient(port: number): Promise<{ sock: net.Socket; initial: Buffer }> { const sock = net.connect(port, "127.0.0.1"); - sock.on("error", () => {}); const { promise, resolve, reject } = Promise.withResolvers(); + sock.on("error", reject); + sock.once("close", () => reject(new Error("socket closed before upgrade completed"))); let buf = Buffer.alloc(0); const onData = (d: Buffer) => { buf = buf.length ? Buffer.concat([buf, d]) : d; @@ -47,6 +48,8 @@ async function pausedClient(port: number): Promise<{ sock: net.Socket; initial: ); }); const initial = await promise; + sock.off("error", reject); + sock.on("error", () => {}); return { sock, initial }; } @@ -62,7 +65,6 @@ describe("BackPressure buffer", () => { let bufferedAfterSend = 0; let drainSawDecrease = true; let prev = Infinity; - let serverWs: import("bun").ServerWebSocket | undefined; const sentSignal = Promise.withResolvers(); const drained = Promise.withResolvers(); await using server = Bun.serve({ @@ -75,7 +77,6 @@ describe("BackPressure buffer", () => { maxBackpressure: SIZE * 2, idleTimeout: 0, open(ws) { - serverWs = ws; ws.sendBinary(payload); bufferedAfterSend = ws.getBufferedAmount(); sentSignal.resolve(); From 4112b87a55456f69c640f639e482e695b8da2034 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Mon, 20 Jul 2026 12:56:35 +0000 Subject: [PATCH 3/7] review: trim comments to 3 lines, guard size arithmetic against overflow --- packages/bun-uws/src/AsyncSocketData.h | 12 +++++++----- .../websocket-server-backpressure-buffer.test.ts | 7 +++---- 2 files changed, 10 insertions(+), 9 deletions(-) diff --git a/packages/bun-uws/src/AsyncSocketData.h b/packages/bun-uws/src/AsyncSocketData.h index b76f5117eb9f..92eacd501fbc 100644 --- a/packages/bun-uws/src/AsyncSocketData.h +++ b/packages/bun-uws/src/AsyncSocketData.h @@ -20,16 +20,15 @@ #define UWS_ASYNCSOCKETDATA_H #include +#include #include #include namespace uWS { -/* Contiguous write-behind buffer with a moving head cursor. erase() is a - * pointer bump; append()/resize() reuse the drained head gap via one memmove - * before ever growing. The previous std::string shape front-erased + shrank to - * fit every ~1/32 drained, so a drain of N bytes moved ~60N bytes and briefly - * held 2x the live data during each realloc. */ +/* Contiguous buffer with a moving head cursor: erase() bumps head, + * append()/resize() compact into the drained head gap before growing, so + * draining never memmoves or reallocates. */ struct BackPressure { BackPressure() = default; BackPressure(BackPressure &&other) noexcept @@ -96,6 +95,9 @@ struct BackPressure { /* Ensure [tail, tail+n) is writable. Prefers compacting into the drained * head gap over growing so steady-state producer/consumer never reallocs. */ void ensureTailRoom(size_t n) { + /* tail + n and live + n cannot wrap past this; cap * 2 wrapping is + * harmless because live + n then wins the max(). */ + if (n > SIZE_MAX - tail) std::abort(); if (tail + n <= cap) return; size_t live = tail - head; diff --git a/test/js/bun/websocket/websocket-server-backpressure-buffer.test.ts b/test/js/bun/websocket/websocket-server-backpressure-buffer.test.ts index b8a33bf0184f..63ce3e32f302 100644 --- a/test/js/bun/websocket/websocket-server-backpressure-buffer.test.ts +++ b/test/js/bun/websocket/websocket-server-backpressure-buffer.test.ts @@ -140,10 +140,9 @@ describe("BackPressure buffer", () => { expect(hash.digest("hex")).toBe(expectedHash); }, 30_000); - // Small (<16KB) sends take getSendBuffer(): once the cork buffer fills they - // hit BackPressure.resize() + the in-place tail write, then write(nullptr,0) - // drains (erase). Keeping the window full forces append()/resize() to reuse - // the drained head gap via a compact instead of reallocating. + // Small (<16KB) sends go through getSendBuffer(): cork overflow hits + // BackPressure.resize() then erase(); keeping the window full makes + // append() compact into the drained head gap instead of reallocating. it("delivers many corked frames while appending into a partly-drained buffer", async () => { const FRAME = 4096; const COUNT = 4096; // 16MB of payload through the cork->backpressure path From 101e1a1f280d10610cdfa8d98ce71c23c3e2022b Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Mon, 20 Jul 2026 13:05:08 +0000 Subject: [PATCH 4/7] test: shrink payloads so default timeout suffices, drop per-test timeouts --- .../websocket-server-backpressure-buffer.test.ts | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/test/js/bun/websocket/websocket-server-backpressure-buffer.test.ts b/test/js/bun/websocket/websocket-server-backpressure-buffer.test.ts index 63ce3e32f302..4cb4d84cdaa1 100644 --- a/test/js/bun/websocket/websocket-server-backpressure-buffer.test.ts +++ b/test/js/bun/websocket/websocket-server-backpressure-buffer.test.ts @@ -58,7 +58,7 @@ describe("BackPressure buffer", () => { // unwritten tail into BackPressure; drain exercises erase() as a pure // head-cursor bump. it("delivers a large direct send byte-for-byte while draining", async () => { - const SIZE = 32 * 1024 * 1024; + const SIZE = 8 * 1024 * 1024; const payload = patternBuffer(SIZE, 0xabcd); const expectedHash = crypto.createHash("sha1").update(payload).digest("hex"); @@ -97,7 +97,7 @@ describe("BackPressure buffer", () => { const { sock, initial } = await pausedClient(server.port); await sentSignal.promise; - // 32MB cannot fit in the kernel send buffer, so a non-empty remainder must + // 8MB cannot fit in the kernel send buffer, so a non-empty remainder must // have been copied into the BackPressure buffer. expect(bufferedAfterSend).toBeGreaterThan(0); expect(bufferedAfterSend).toBeLessThanOrEqual(SIZE + 10); @@ -138,14 +138,14 @@ describe("BackPressure buffer", () => { expect(drainSawDecrease).toBe(true); expect(received).toBe(target); expect(hash.digest("hex")).toBe(expectedHash); - }, 30_000); + }); // Small (<16KB) sends go through getSendBuffer(): cork overflow hits // BackPressure.resize() then erase(); keeping the window full makes // append() compact into the drained head gap instead of reallocating. it("delivers many corked frames while appending into a partly-drained buffer", async () => { const FRAME = 4096; - const COUNT = 4096; // 16MB of payload through the cork->backpressure path + const COUNT = 2048; // 8MB: exceeds Linux tcp_wmem max (4MB) so the window fills const WINDOW = 1 * 1024 * 1024; const headerLen = 4; // server frame, 16-bit extended length, no mask @@ -233,5 +233,5 @@ describe("BackPressure buffer", () => { expect(sent).toBe(COUNT); expect(received).toBe(target); expect(hash.digest("hex")).toBe(expectedHash); - }, 30_000); + }); }); From cafb81de6e627be7910b0eb1855c350e5d91b579 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Mon, 20 Jul 2026 13:30:12 +0000 Subject: [PATCH 5/7] test: skip direct-send backpressure test on Windows Winsock's loopback send() accepts the full payload (100MB+ with the client paused) into kernel buffers so the WebSocket fast path never falls back to the uws BackPressure buffer on Windows. The cork-overflow test still exercises the buffer on that platform. Also defer the send until after the client has paused so the assertion that something was buffered is deterministic on POSIX. --- ...bsocket-server-backpressure-buffer.test.ts | 22 ++++++++++--------- 1 file changed, 12 insertions(+), 10 deletions(-) diff --git a/test/js/bun/websocket/websocket-server-backpressure-buffer.test.ts b/test/js/bun/websocket/websocket-server-backpressure-buffer.test.ts index 4cb4d84cdaa1..80c2e8c2f83a 100644 --- a/test/js/bun/websocket/websocket-server-backpressure-buffer.test.ts +++ b/test/js/bun/websocket/websocket-server-backpressure-buffer.test.ts @@ -1,4 +1,5 @@ import { describe, expect, it } from "bun:test"; +import { isWindows } from "harness"; import crypto from "node:crypto"; import net from "node:net"; @@ -56,8 +57,10 @@ async function pausedClient(port: number): Promise<{ sock: net.Socket; initial: describe("BackPressure buffer", () => { // >16KB sends take the direct us_socket_write2 path and then append() the // unwritten tail into BackPressure; drain exercises erase() as a pure - // head-cursor bump. - it("delivers a large direct send byte-for-byte while draining", async () => { + // head-cursor bump. Winsock's loopback send() accepts the entire payload + // (100MB+ observed with the client paused), so this path never reaches + // BackPressure on Windows; the cork-overflow test below covers that platform. + it.skipIf(isWindows)("delivers a large direct send byte-for-byte while draining", async () => { const SIZE = 8 * 1024 * 1024; const payload = patternBuffer(SIZE, 0xabcd); const expectedHash = crypto.createHash("sha1").update(payload).digest("hex"); @@ -65,7 +68,7 @@ describe("BackPressure buffer", () => { let bufferedAfterSend = 0; let drainSawDecrease = true; let prev = Infinity; - const sentSignal = Promise.withResolvers(); + const opened = Promise.withResolvers>(); const drained = Promise.withResolvers(); await using server = Bun.serve({ port: 0, @@ -77,9 +80,7 @@ describe("BackPressure buffer", () => { maxBackpressure: SIZE * 2, idleTimeout: 0, open(ws) { - ws.sendBinary(payload); - bufferedAfterSend = ws.getBufferedAmount(); - sentSignal.resolve(); + opened.resolve(ws); }, drain(ws) { const b = ws.getBufferedAmount(); @@ -95,10 +96,11 @@ describe("BackPressure buffer", () => { }); const { sock, initial } = await pausedClient(server.port); - await sentSignal.promise; - - // 8MB cannot fit in the kernel send buffer, so a non-empty remainder must - // have been copied into the BackPressure buffer. + const ws = await opened.promise; + // Send only after the client has paused its read side so the kernel send + // buffer is the only sink; a non-empty remainder lands in BackPressure. + ws.sendBinary(payload); + bufferedAfterSend = ws.getBufferedAmount(); expect(bufferedAfterSend).toBeGreaterThan(0); expect(bufferedAfterSend).toBeLessThanOrEqual(SIZE + 10); From 43c62e8f9c4a6005ee2be37b8453a8ade52c11a5 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Mon, 20 Jul 2026 14:03:22 +0000 Subject: [PATCH 6/7] test: trim Windows-skip comment to 3 lines --- .../websocket-server-backpressure-buffer.test.ts | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/test/js/bun/websocket/websocket-server-backpressure-buffer.test.ts b/test/js/bun/websocket/websocket-server-backpressure-buffer.test.ts index 80c2e8c2f83a..5641a63d5f90 100644 --- a/test/js/bun/websocket/websocket-server-backpressure-buffer.test.ts +++ b/test/js/bun/websocket/websocket-server-backpressure-buffer.test.ts @@ -55,11 +55,9 @@ async function pausedClient(port: number): Promise<{ sock: net.Socket; initial: } describe("BackPressure buffer", () => { - // >16KB sends take the direct us_socket_write2 path and then append() the - // unwritten tail into BackPressure; drain exercises erase() as a pure - // head-cursor bump. Winsock's loopback send() accepts the entire payload - // (100MB+ observed with the client paused), so this path never reaches - // BackPressure on Windows; the cork-overflow test below covers that platform. + // >16KB sends take the direct write2 path and append() the unwritten tail; + // drain exercises erase() as a head-cursor bump. Skipped on Windows: Winsock + // loopback accepts the full payload so BackPressure is never reached there. it.skipIf(isWindows)("delivers a large direct send byte-for-byte while draining", async () => { const SIZE = 8 * 1024 * 1024; const payload = patternBuffer(SIZE, 0xabcd); From 8db78aa00b3d46f2a0c4408d10ae139ec89458fa Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Mon, 20 Jul 2026 21:19:27 +0000 Subject: [PATCH 7/7] uws: route BackPressure allocation through us_malloc/us_realloc/us_free These resolve to mimalloc in non-ASAN builds and to libc under ASAN, matching every other allocation in bun-usockets/bun-uws. --- packages/bun-uws/src/AsyncSocketData.h | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/packages/bun-uws/src/AsyncSocketData.h b/packages/bun-uws/src/AsyncSocketData.h index 92eacd501fbc..b38061ee3636 100644 --- a/packages/bun-uws/src/AsyncSocketData.h +++ b/packages/bun-uws/src/AsyncSocketData.h @@ -19,6 +19,8 @@ #ifndef UWS_ASYNCSOCKETDATA_H #define UWS_ASYNCSOCKETDATA_H +#include "libusockets.h" + #include #include #include @@ -38,7 +40,7 @@ struct BackPressure { } BackPressure(const BackPressure &) = delete; BackPressure &operator=(const BackPressure &) = delete; - ~BackPressure() { std::free(buf); } + ~BackPressure() { us_free(buf); } /* Unsent bytes. data() points at length() contiguous bytes. */ size_t length() const { return tail - head; } @@ -111,14 +113,14 @@ struct BackPressure { size_t newCap = std::max(std::max(cap * 2, live + n), MIN_CAPACITY); char *nb; if (head == 0) { - /* realloc may extend in place (mimalloc, glibc mremap). */ - nb = (char *) std::realloc(buf, newCap); + /* mi_realloc may extend in place. */ + nb = (char *) us_realloc(buf, newCap); if (!nb) std::abort(); } else { - nb = (char *) std::malloc(newCap); + nb = (char *) us_malloc(newCap); if (!nb) std::abort(); if (live) std::memcpy(nb, buf + head, live); - std::free(buf); + us_free(buf); head = 0; tail = live; } @@ -127,7 +129,7 @@ struct BackPressure { } void release() { - std::free(buf); + us_free(buf); buf = nullptr; cap = 0; }