Fix 5 bugs found via wrk stress testing the HTTP proxy

1. tcp_write infinite loop: When the output byte queue enters an error
   state (buffer full), tcp_write would loop forever because
   byte_queue_write_buf returns {NULL,0} without advancing. Break out
   of the loop when the write buffer is empty.

2. Proxy ignores async return values: toastyfs_async_get/put/delete
   can fail (return -1) but the proxy unconditionally set the operation
   to STARTED, causing it to wait forever for a result that never comes.
   Now checks return values and responds with 500 on failure.

3. Use-after-free on closed connections: When a client disconnects while
   its request is queued as PENDING, the HTTP_Conn is freed but the proxy
   still holds stale pointers. Added http_response_builder_is_valid() to
   detect and discard operations for dead connections.

4. Wrong HTTP status codes: The proxy returned 500 for all errors. Now
   returns 507 (Insufficient Storage) for FULL, 404 for NOT_FOUND, and
   503 (Service Unavailable) when the operation queue is full.

5. TCP connection leak (root cause of proxy hang): When a peer closes a
   connection, process_conn_events closes the socket and sets HUP, but
   tcp_conn_free_maybe immediately frees the TCP_Conn before
   tcp_next_event can deliver the HUP event. The HTTP server never learns
   about the disconnect, so HTTP_Conn entries accumulate until max_conns
   is reached and the proxy stops accepting new connections. Fixed by
   only calling tcp_conn_free_maybe when the CLOSED flag is already set
   (meaning the user called tcp_close), preserving the connection for
   HUP event delivery otherwise.

Also adds wrk-based stress test suite (stress-test.sh + lua scripts).

https://claude.ai/code/session_01EHoWWnVgyCFxah5WaNV4kS

lib/http_server.c

@@ -272,6 +272,11 @@ builder_to_conn(HTTP_ResponseBuilder builder)
     return conn;
 }
 
+bool http_response_builder_is_valid(HTTP_ResponseBuilder builder)
+{
+    return builder_to_conn(builder) != NULL;
+}
+
 void http_response_builder_status(HTTP_ResponseBuilder builder, int status)
 {
     HTTP_Conn *conn = builder_to_conn(builder);

lib/http_server.h

@@ -60,6 +60,7 @@ typedef struct {
 bool http_server_next_request(HTTP_Server *server,
     HTTP_Request **request, HTTP_ResponseBuilder *builder);
 
+bool http_response_builder_is_valid(HTTP_ResponseBuilder builder);
 void http_response_builder_status(HTTP_ResponseBuilder builder, int status);
 void http_response_builder_header(HTTP_ResponseBuilder builder, string header);
 void http_response_builder_content(HTTP_ResponseBuilder builder, string content);

lib/tcp.c

@@ -39,7 +39,14 @@ typedef struct {
     // ID of the general state this structure is in
     TCP_ConnState state;
 
-    // Information about the socket
+    // Information about the socket:
+    //   - TCP_CONN_FLAG_CLOSED
+    //       Whether the user is holding a handle to this struct.
+    //       It's first set when the TCP_EVENT_NEW is passed to
+    //       the user, and it's unset when the user calls tcp_close.
+    //   - TCP_CONN_FLAG_SECURE
+    //       Whether the connection was establushed via the
+    //       encrypted interface or not
     int flags;
 
     // Events associated to this connection that the user
@@ -57,11 +64,6 @@ typedef struct {
     // Underlying socket
     SOCKET fd;
 
-    // Whether the user is holding a handle to this struct.
-    // It's first set when the TCP_EVENT_NEW is passed to
-    // the user, and it's unset when the user calls tcp_close.
-    bool handled;
-
     // The socket should be closing as soon as the buffered
     // output data has been flushed. When this is set, no more
     // data can be buffered from the network.
@@ -347,7 +349,6 @@ static void tcp_conn_init(TCP *tcp, TCP_Conn *conn, bool secure, TCP_ConnState s
     conn->state = state;
     conn->flags = 0;
     conn->events = 0;
-    conn->handled = false;
     conn->closing = false;
     conn->fd = fd;
     conn->num_addrs = 0;
@@ -506,7 +507,7 @@ static bool tcp_conn_is_buffering(TCP_Conn *conn)
 
 static bool tcp_conn_free_maybe(TCP_Conn *conn)
 {
-    if (!conn->handled && conn->fd == INVALID_SOCKET) {
+    if (!(conn->flags & TCP_CONN_FLAG_CLOSED) && conn->fd == INVALID_SOCKET) {
         tcp_conn_free(conn);
         return true;
     } else {
@@ -1064,6 +1065,9 @@ void tcp_write(TCP_Handle handle, string data)
         byte_queue_write_setmincap(&conn->output, data.len);
         string buf = tcp_write_buf(handle);
 
+        if (buf.len == 0)
+            break; // Output buffer full or in error state
+
         int num = MIN(buf.len, data.len);
         memcpy(buf.ptr, data.ptr, num);
 
@@ -1101,8 +1105,10 @@ void tcp_close(TCP_Handle handle)
     if (conn == NULL)
         return;
 
+    // Only free immediately if the user already called tcp_close
+    // (CLOSED flag set). Otherwise, keep the connection alive so
+    // tcp_next_event can deliver the HUP event to the user.
     conn->flags |= TCP_CONN_FLAG_CLOSED;
-    conn->handled = false;
     tcp_conn_invalidate_handles(conn);
     if (tcp_conn_free_maybe(conn)) {
         tcp->num_conns--;

src/http_proxy.c

@@ -101,6 +101,7 @@ int http_proxy_tick(void *state, void **ctxs,
     struct pollfd *pdata, int pcap, int *pnum, int *timeout)
 {
     HTTPProxy *proxy = state;
+
     http_server_process_events(&proxy->http_server,
         ctxs  + proxy->num_polled_by_toasty,
         pdata + proxy->num_polled_by_toasty,
@@ -123,37 +124,42 @@ int http_proxy_tick(void *state, void **ctxs,
         ProxyOper *oper = &proxy->opers[i];
         assert(oper->state == PROXY_OPER_STARTED);
 
-        HTTP_Request *request = oper->request;
         HTTP_ResponseBuilder builder = oper->builder;
 
         switch (result.type) {
         case TOASTYFS_RESULT_PUT:
-            assert(request->method == CHTTP_METHOD_PUT);
             if (result.error == TOASTYFS_ERROR_VOID) {
                 http_response_builder_status(builder, 201);
                 http_response_builder_submit(builder);
+            } else if (result.error == TOASTYFS_ERROR_FULL) {
+                http_response_builder_status(builder, 507);
+                http_response_builder_submit(builder);
             } else {
                 http_response_builder_status(builder, 500);
                 http_response_builder_submit(builder);
             }
             break;
         case TOASTYFS_RESULT_GET:
-            assert(request->method == CHTTP_METHOD_GET);
             if (result.error == TOASTYFS_ERROR_VOID) {
                 http_response_builder_status(builder, 200);
                 http_response_builder_content(builder, (string) { result.data, result.size });
                 http_response_builder_submit(builder);
                 free(result.data);
+            } else if (result.error == TOASTYFS_ERROR_NOT_FOUND) {
+                http_response_builder_status(builder, 404);
+                http_response_builder_submit(builder);
             } else {
                 http_response_builder_status(builder, 500);
                 http_response_builder_submit(builder);
             }
             break;
         case TOASTYFS_RESULT_DELETE:
-            assert(request->method == CHTTP_METHOD_DELETE);
             if (result.error == TOASTYFS_ERROR_VOID) {
                 http_response_builder_status(builder, 204);
                 http_response_builder_submit(builder);
+            } else if (result.error == TOASTYFS_ERROR_NOT_FOUND) {
+                http_response_builder_status(builder, 404);
+                http_response_builder_submit(builder);
             } else {
                 http_response_builder_status(builder, 500);
                 http_response_builder_submit(builder);
@@ -166,6 +172,17 @@ int http_proxy_tick(void *state, void **ctxs,
         oper->state = PROXY_OPER_FREE;
     }
 
+    // Discard pending operations whose connection has been closed.
+    // When a client disconnects, the HTTP_Conn is freed and the builder
+    // becomes invalid.  The request pointers (url, body) reference the
+    // now-freed TCP read buffer, so we must not dereference them.
+    for (int i = 0; i < proxy->max_opers; i++) {
+        if (proxy->opers[i].state == PROXY_OPER_PENDING
+            && !http_response_builder_is_valid(proxy->opers[i].builder)) {
+            proxy->opers[i].state = PROXY_OPER_FREE;
+        }
+    }
+
     // Buffer operation requests
     for (;;) {
         HTTP_Request *request;
@@ -189,7 +206,7 @@ int http_proxy_tick(void *state, void **ctxs,
 
         if (i == proxy->max_opers) {
             // Queue is full
-            http_response_builder_status(builder, 500);
+            http_response_builder_status(builder, 503);
             http_response_builder_submit(builder);
             continue;
         }
@@ -223,28 +240,39 @@ int http_proxy_tick(void *state, void **ctxs,
             if (i < proxy->max_opers) {
                 // Found pending operation
                 ProxyOper *oper = &proxy->opers[i];
-
-                // Begin the operation based on the request method
                 HTTP_Request *request = oper->request;
+                int ret = -1;
+
                 switch (request->method) {
                 case CHTTP_METHOD_GET:
-                    toastyfs_async_get(proxy->toastyfs, request->url.path.ptr, request->url.path.len);
+                    ret = toastyfs_async_get(proxy->toastyfs,
+                        request->url.path.ptr, request->url.path.len);
                     break;
                 case CHTTP_METHOD_PUT:
-                    toastyfs_async_put(proxy->toastyfs, request->url.path.ptr, request->url.path.len,
+                    ret = toastyfs_async_put(proxy->toastyfs,
+                        request->url.path.ptr, request->url.path.len,
                         request->body.ptr, request->body.len);
                     break;
                 case CHTTP_METHOD_DELETE:
-                    toastyfs_async_delete(proxy->toastyfs, request->url.path.ptr, request->url.path.len);
+                    ret = toastyfs_async_delete(proxy->toastyfs,
+                        request->url.path.ptr, request->url.path.len);
                     break;
                 default:
                     UNREACHABLE;
                 }
 
+                if (ret < 0) {
+                    // Async operation failed to start -- respond with error
+                    // and free the slot so the proxy doesn't get stuck.
+                    http_response_builder_status(oper->builder, 500);
+                    http_response_builder_submit(oper->builder);
+                    oper->state = PROXY_OPER_FREE;
+                } else {
                     oper->state = PROXY_OPER_STARTED;
                 }
             }
         }
+    }
 
     // Register events that need to be monitored
     {

stress-get.lua

@@ -0,0 +1,10 @@
+-- wrk script for GET stress test
+-- Reads previously written keys
+
+counter = 0
+
+request = function()
+    counter = counter + 1
+    local key = "/stress-key-" .. (counter % 100 + 1)
+    return wrk.format("GET", key, nil, nil)
+end

stress-mixed.lua

@@ -0,0 +1,22 @@
+-- wrk script for mixed workload stress test
+-- Mix of PUT, GET, and DELETE operations
+
+counter = 0
+
+request = function()
+    counter = counter + 1
+    local op = counter % 10
+    local key = "/mixed-key-" .. (counter % 200 + 1)
+
+    if op < 5 then
+        -- 50% PUTs
+        local body = "val-" .. counter .. "-" .. string.rep("y", 64)
+        return wrk.format("PUT", key, nil, body)
+    elseif op < 9 then
+        -- 40% GETs
+        return wrk.format("GET", key, nil, nil)
+    else
+        -- 10% DELETEs
+        return wrk.format("DELETE", key, nil, nil)
+    end
+end

stress-put-report.lua

@@ -0,0 +1,27 @@
+-- wrk script for PUT stress test with response tracking
+
+counter = 0
+responses = {}
+
+request = function()
+    counter = counter + 1
+    local key = "/stress-rpt-" .. counter
+    local body = "value-" .. counter .. "-" .. string.rep("x", 128)
+    return wrk.format("PUT", key, nil, body)
+end
+
+response = function(status, headers, body)
+    if responses[status] == nil then
+        responses[status] = 0
+    end
+    responses[status] = responses[status] + 1
+end
+
+done = function(summary, latency, requests)
+    io.write("\n--- Response Status Breakdown ---\n")
+    for status, count in pairs(responses) do
+        io.write(string.format("  HTTP %d: %d responses\n", status, count))
+    end
+    io.write(string.format("\n  Total errors (connect/read/write/timeout): %d/%d/%d/%d\n",
+        summary.errors.connect, summary.errors.read, summary.errors.write, summary.errors.timeout))
+end

stress-put.lua

@@ -0,0 +1,11 @@
+-- wrk script for PUT stress test
+-- Each request PUTs a unique key with a small payload
+
+counter = 0
+
+request = function()
+    counter = counter + 1
+    local key = "/stress-key-" .. counter
+    local body = "value-" .. counter .. "-" .. string.rep("x", 128)
+    return wrk.format("PUT", key, nil, body)
+end

stress-test.sh

@@ -0,0 +1,138 @@
+#!/bin/bash
+# Comprehensive stress test for ToastyFS HTTP proxy
+set -e
+
+PROXY="http://127.0.0.1:3000"
+PASS=0
+FAIL=0
+
+pass() { echo "  PASS: $1"; PASS=$((PASS + 1)); }
+fail() { echo "  FAIL: $1"; FAIL=$((FAIL + 1)); }
+
+echo "=== ToastyFS Stress Test Suite ==="
+echo ""
+
+# -------------------------------------------------------
+# Test 1: Basic sanity check
+# -------------------------------------------------------
+echo "--- Test 1: Basic PUT/GET/DELETE ---"
+STATUS=$(curl -s -o /dev/null -w "%{http_code}" -X PUT -d "hello" "$PROXY/sanity-key")
+if [ "$STATUS" = "201" ]; then pass "PUT returns 201"; else fail "PUT returned $STATUS instead of 201"; fi
+
+BODY=$(curl -s "$PROXY/sanity-key")
+if [ "$BODY" = "hello" ]; then pass "GET returns correct data"; else fail "GET returned '$BODY' instead of 'hello'"; fi
+
+STATUS=$(curl -s -o /dev/null -w "%{http_code}" -X DELETE "$PROXY/sanity-key")
+if [ "$STATUS" = "204" ]; then pass "DELETE returns 204"; else fail "DELETE returned $STATUS instead of 204"; fi
+
+echo ""
+
+# -------------------------------------------------------
+# Test 2: wrk PUT stress test with response tracking
+# -------------------------------------------------------
+echo "--- Test 2: PUT stress test (4 threads, 16 connections, 10s) ---"
+WRK_OUT=$(wrk -t4 -c16 -d10s -s stress-put.lua --timeout 10s "$PROXY" 2>&1)
+echo "$WRK_OUT"
+echo ""
+
+TOTAL_REQ=$(echo "$WRK_OUT" | grep "requests in" | awk '{print $1}')
+NON_2XX=$(echo "$WRK_OUT" | grep "Non-2xx" | awk '{print $NF}')
+if [ -z "$NON_2XX" ]; then NON_2XX=0; fi
+
+echo "  Total requests: $TOTAL_REQ"
+echo "  Non-2xx responses: $NON_2XX"
+
+if [ "$NON_2XX" -gt 0 ]; then
+    fail "Got $NON_2XX error responses during PUT stress test"
+else
+    pass "All PUT requests succeeded"
+fi
+echo ""
+
+# -------------------------------------------------------
+# Test 3: Post-stress responsiveness check
+# -------------------------------------------------------
+echo "--- Test 3: Post-stress responsiveness ---"
+# Immediately try to PUT a key after wrk ends
+START=$(date +%s%3N)
+STATUS=$(curl -s -o /dev/null -w "%{http_code}" --max-time 5 -X PUT -d "post-stress" "$PROXY/post-stress-key" 2>&1)
+END=$(date +%s%3N)
+LATENCY=$((END - START))
+
+if [ "$STATUS" = "201" ]; then
+    pass "Proxy responsive after stress (${LATENCY}ms)"
+else
+    fail "Proxy unresponsive after stress: status=$STATUS latency=${LATENCY}ms"
+fi
+echo ""
+
+# -------------------------------------------------------
+# Test 4: High concurrency test (many connections)
+# -------------------------------------------------------
+echo "--- Test 4: High concurrency (4 threads, 64 connections, 10s) ---"
+WRK_OUT=$(wrk -t4 -c64 -d10s -s stress-put.lua --timeout 10s "$PROXY" 2>&1)
+echo "$WRK_OUT"
+echo ""
+
+NON_2XX=$(echo "$WRK_OUT" | grep "Non-2xx" | awk '{print $NF}')
+ERRORS=$(echo "$WRK_OUT" | grep -E "Socket errors|connect" || echo "")
+if [ -z "$NON_2XX" ]; then NON_2XX=0; fi
+
+echo "  Non-2xx: $NON_2XX"
+echo "  Socket errors: $ERRORS"
+
+if [ "$NON_2XX" -gt 0 ]; then
+    fail "Got $NON_2XX error responses under high concurrency"
+else
+    pass "High concurrency test passed"
+fi
+echo ""
+
+# -------------------------------------------------------
+# Test 5: Post-high-concurrency responsiveness
+# -------------------------------------------------------
+echo "--- Test 5: Post-high-concurrency responsiveness ---"
+START=$(date +%s%3N)
+STATUS=$(curl -s -o /dev/null -w "%{http_code}" --max-time 5 -X PUT -d "post-hc" "$PROXY/post-hc-key" 2>&1)
+END=$(date +%s%3N)
+LATENCY=$((END - START))
+
+if [ "$STATUS" = "201" ]; then
+    pass "Proxy responsive after high concurrency (${LATENCY}ms)"
+else
+    fail "Proxy unresponsive after high concurrency: status=$STATUS latency=${LATENCY}ms"
+fi
+echo ""
+
+# -------------------------------------------------------
+# Test 6: Mixed workload
+# -------------------------------------------------------
+echo "--- Test 6: Mixed workload (4 threads, 32 connections, 10s) ---"
+# Seed some data first
+for i in $(seq 1 50); do
+    curl -s -o /dev/null -X PUT -d "seed-$i" "$PROXY/mixed-key-$i"
+done
+
+WRK_OUT=$(wrk -t4 -c32 -d10s -s stress-mixed.lua --timeout 10s "$PROXY" 2>&1)
+echo "$WRK_OUT"
+echo ""
+
+# -------------------------------------------------------
+# Test 7: GET stress test
+# -------------------------------------------------------
+echo "--- Test 7: GET stress test (4 threads, 16 connections, 10s) ---"
+# Seed data for GET test
+for i in $(seq 1 100); do
+    curl -s -o /dev/null -X PUT -d "get-test-data-$i" "$PROXY/stress-key-$i"
+done
+
+WRK_OUT=$(wrk -t4 -c16 -d10s -s stress-get.lua --timeout 10s "$PROXY" 2>&1)
+echo "$WRK_OUT"
+echo ""
+
+# -------------------------------------------------------
+# Summary
+# -------------------------------------------------------
+echo "================================"
+echo "Results: $PASS passed, $FAIL failed"
+echo "================================"