uws: fix O(n^2) BackPressure buffer compaction and reallocation on large writes

packages/bun-uws/src/AsyncSocket.h

@@ -183,8 +183,8 @@ 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 only; already-written bytes waiting for compaction are not backpressure. */
+        return getAsyncSocketData()->buffer.length();
     }
 
     /* Returns the text representation of an IPv4 or IPv6 address */
@@ -253,7 +253,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 +301,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 */

packages/bun-uws/src/AsyncSocketData.h

@@ -25,7 +25,7 @@ namespace uWS {
 
 struct BackPressure {
     std::string buffer;
-    unsigned int pendingRemoval = 0;
+    size_t pendingRemoval = 0;
     BackPressure(BackPressure &&other) {
         buffer = std::move(other.buffer);
         pendingRemoval = other.pendingRemoval;
@@ -34,12 +34,12 @@ struct BackPressure {
     void append(const char *data, size_t length) {
         buffer.append(data, length);
     }
-    void erase(unsigned int length) {
+    void erase(size_t length) {
         pendingRemoval += length;
-        /* Always erase a minimum of 1/32th the current backpressure */
-        if (pendingRemoval > (buffer.length() >> 5)) {
+        /* Compact once half the buffer is dead space so draining n bytes moves
+         * O(n) total (geometric series). clear() releases capacity on full drain. */
+        if (pendingRemoval > (buffer.length() >> 1)) {
             buffer.erase(0, pendingRemoval);
-            buffer.shrink_to_fit();
             pendingRemoval = 0;
         }
     }

packages/bun-uws/src/HttpResponse.h

@@ -596,6 +596,14 @@
 
         size_t length = data.length();
 
+        /* A write this large is split into several AsyncSocket::write calls below
+         * (chunk framing, INT_MAX slicing). Reserve once so those appends never
+         * realloc + copy a multi-GB backpressure buffer. */
+        if (length > 1024 * 1024) {
+            auto &bp = Super::getAsyncSocketData()->buffer;
+            bp.reserve(bp.length() + length + 32);
+        }
+
         // Special handling for extremely large data (greater than UINT_MAX bytes)
         // most clients expect a max of UINT_MAX, so we need to split the write into multiple writes
         if (length > UINT_MAX) {

packages/bun-uws/src/WebSocket.h

@@ -106,7 +106,8 @@ struct WebSocket : AsyncSocket<SSL> {
     }
 
     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

test/js/node/http/node-http-backpressure-max.test.ts

@@ -11,7 +11,7 @@ import http from "node:http";
 import type { AddressInfo } from "node:net";
 
 describe("backpressure", () => {
-  // Linux CI only have 8GB with is not enought because we will clone all or most of this 4GB into memory
+  // Linux CI only has 8GB which is not enough because we clone all or most of this 4GB into memory.
   it.skipIf(isCI && isLinux)(
     "should handle backpressure with the maximum allowed bytes",
     async () => {
@@ -45,9 +45,6 @@ describe("backpressure", () => {
 
       expect(totalBytes).toBe(payloadSize);
     },
-    // Moving 4 GiB through the server and the fetch reader takes ~60s on the
-    // slowest CI runners (darwin x64), which sat exactly at the old 60s
-    // limit and made the test flaky there.
-    120_000,
+    30_000,
   );
 });