Bug Fixes

docs/OPERATIONS.md

@@ -2,61 +2,47 @@
 
 ## Known issues
 
-### NULL-buffer heap corruption in HTTP body handlers (root cause under investigation)
+### NULL-buffer heap corruption in HTTP body handlers (root cause: data races — fixed)
 
 **Symptom:** Historically, `buffer->data` was observed NULL inside the HTTP
 body handling paths — `_on_body` (http_parser body callback in
 `src/ClientAPI/HTTP/http_connection.c`), `_put_on_request_data` (streamed PUT
 body handler in `src/ClientAPI/HTTP/off_routes.c`), and `_pipe_on_data`
 (response piping callback in `src/ClientAPI/HTTP/http_response.c`). The
 `buffer_t` struct had `capacity` set correctly but `data == NULL` and `size`
-held garbage, suggesting the struct fields were overwritten by heap corruption
-from an unknown source.
-
-**Guards in place:** Defensive sentinels were restored at the entry of each
-of those three handlers. A NULL `buffer->data` (or NULL `chunk`/`at` pointer)
-now logs an `error`-level message identifying the handler and the suspicious
-pointer values, then returns without dereferencing. In `_on_body` the parse
-is aborted by returning `1` to http-parser; in the streamed-PUT and
-response-pipe paths the chunk is dropped. This prevents the NULL dereference
-crash but does not address the underlying corruption.
-
-**Root cause status:** Under investigation. The corruption is not easily
-reproducible under ASAN (ASAN redzones mask the bug), and the flaky
-`TestStream*` segfaults observed in ASAN builds do not produce an ASAN
-report (the SIGSEGV bypasses ASAN's signal handler, produces no core dump,
-does not reproduce under gdb/strace/pty, and does not reproduce in non-ASAN
-builds). The failing `TestStream*` tests (`TestPushFileStream.*`,
-`TestPullFileStream.*`, `TestStreamActor.*`) are file-stream + scheduler
-tests and do not exercise the HTTP body handlers directly, so the sentinel
-guards do not resolve their segfaults — but the guards are retained as the
-spec's accepted fallback for the historical NULL-buffer crash.
-
-**Investigation notes:**
-
-- The flaky ASAN segfault is timing-dependent (reproduces only when stdout
-  is file-redirected, not under a pty; ~30% rate in isolated process runs).
-- ASAN installs its SIGSEGV handler but does not fire a report when the
-  segfault occurs, suggesting the fault happens in a state where ASAN's
-  handler cannot safely run (e.g. during process teardown after main
-  returns, or in a thread that hasn't registered its stack with ASAN).
-- `buffer_ensure_capacity` aborts on OOM, so `buffer->data` is never NULL
-  in normal operation — the NULL must come from external heap corruption.
-- Candidates not yet ruled out: a missing `REFERENCE` on a `buffer_t*`
-  crossing an actor boundary; a `stream_notify` CONSUME/yield ownership
-  bug; a double-free in dispatch (per the
-  `feedback_double_free_dispatch.md` memory note — `actor_run` frees
-  `msg->payload`; dispatch must not also free it); a `stream_deactivate`
-  freeing a buffer while a handler still reads it.
-
-**Next steps for a future investigation:**
-
-1. Run the `TestStream*` tests under ThreadSanitizer (TSAN) to catch the
-   race that ASAN misses.
-2. Audit `actor_run`'s payload destroy path against every dispatch handler
-   in `src/Streams/` and `src/ClientAPI/HTTP/` for the double-free pattern
-   documented in `feedback_double_free_dispatch.md`.
-3. Stress-run the file-stream pipeline under valgrind with
-   `--track-origins=yes` to capture the corruption source.
-4. Once the root cause is found, remove the sentinel guards and replace
-   with the minimal fix.
+held garbage.
+
+**Root cause:** The NULL `buffer->data` was a downstream symptom of heap
+corruption from two cross-thread data races, not a `buffer_t` bug. TSAN
+caught both (ASAN and valgrind miss them):
+
+1. **`connection->sock` use-after-free** — `_connection_close_fd` (worker
+   thread) freed the socket and set `connection->sock = NULL` while
+   `_connection_read_callback` (I/O thread) read it. Fixed by making
+   `connection->sock` an `ATOMIC(platform_socket_t*)` and deferring the
+   socket's close+free to the I/O thread's destroy stack
+   (`http_server_defer_socket_destroy`), mirroring the existing
+   watcher/timer deferral.
+
+2. **`pipe_notifiers` use-after-free WRITE** — `readable_push_stream_pipe` /
+   `writeable_pull_stream_pipe` called `on_pipe`/`on_piped` synchronously on
+   the caller's thread, writing `pipe_notifiers` while
+   `stream_unsubscribe_pipe_notifiers` (worker thread) freed it. Fixed by
+   routing pipe/piped through the stream actor via the already-declared
+   `STREAM_PIPE`/`STREAM_PIPED` messages and `stream_pipe_internal` /
+   `stream_piped_internal`.
+
+**Guards in place:** The defensive sentinels at the entry of the three body
+handlers remain as cheap no-op checks (http-parser never legitimately passes
+NULL `at`/`length`), but they are no longer the fix — the underlying races
+are resolved.
+
+**Verification:** `TestPushFileStream.*`, `TestPullFileStream.*`,
+`TestStreamActor.*`, `TestHttpServer.*`, `TestOffRoutes.*`, and
+`TestHttpServerSsl.*` all pass under TSAN with zero data-race reports, and
+the full 849-test suite passes. The GET-path pipeline refcount leak in
+`_setup_stream_pipeline` (off_routes.c) that previously leaked 48 bytes
+direct + 209 bytes indirect per GET request has also been fixed — the
+`get_pipeline_t` refcount now reaches zero in all paths via a `desc_done`
+flag that ensures desc contributes exactly one deref whether close or
+error fires first.

src/ClientAPI/HTTP/http_connection.c

@@ -339,9 +339,13 @@ static void _connection_stop_watcher(http_connection_t* connection) {
 
 /* Close the fd and mark connection as closing. Used from dispatch (worker thread). */
 static void _connection_close_fd(http_connection_t* connection) {
-  if (connection->sock != NULL) {
-    platform_socket_destroy(connection->sock);
-    connection->sock = NULL;
+  platform_socket_t* sock = ATOMIC_EXCHANGE(&connection->sock, NULL);
+  if (sock != NULL) {
+    if (connection->server != NULL) {
+      http_server_defer_socket_destroy(connection->server, sock);
+    } else {
+      platform_socket_destroy(sock);
+    }
   }
   connection->is_closing = 1;
 }
@@ -359,8 +363,9 @@ static int _connection_send_raw_blocking(http_connection_t* connection,
                                           const uint8_t* data, size_t len,
                                           size_t* out_sent) {
   size_t sent_total = 0;
+  platform_socket_t* sock = ATOMIC_LOAD(&connection->sock);
   for (int attempts = 0; attempts < 2000 && sent_total < len; attempts++) {
-    ssize_t sent = platform_socket_send(connection->sock, data + sent_total,
+    ssize_t sent = platform_socket_send(sock, data + sent_total,
                                          len - sent_total);
     if (sent > 0) {
       sent_total += (size_t)sent;
@@ -461,13 +466,13 @@ static int _connection_send_all_blocking(http_connection_t* connection,
  * cross-thread race on the BIO. */
 static void _connection_ssl_data_handle(http_connection_t* connection,
                                         buffer_t* data) {
-  if (connection->sock == NULL || connection->ssl == NULL) {
+  if (ATOMIC_LOAD(&connection->sock) == NULL || connection->ssl == NULL) {
     return;
   }
   BIO_write(connection->rbio, data->data, (int)data->size);
 
   for (int batch = 0; batch < 16; batch++) {
-    if (connection->sock == NULL) {
+    if (ATOMIC_LOAD(&connection->sock) == NULL) {
       return;
     }
     char buffer[READ_BUFFER_SIZE];
@@ -566,7 +571,7 @@ void http_connection_dispatch(void* state, message_t* msg) {
       /* ASIO-style: the I/O thread notified us that data is available.
          Perform the actual recv() and parsing here on the scheduler worker. */
       atomic_store(&connection->read_pending, 0);
-      if (connection->sock == NULL) {
+      if (ATOMIC_LOAD(&connection->sock) == NULL) {
         break;
       }
 #ifndef _WIN32
@@ -582,7 +587,7 @@ void http_connection_dispatch(void* state, message_t* msg) {
          goes through READABLE -> _connection_do_reads. */
       buffer_t* data = (buffer_t*)msg->payload;
       msg->payload = NULL; /* Take ownership — actor_run won't destroy it */
-      if (connection->sock == NULL) {
+      if (ATOMIC_LOAD(&connection->sock) == NULL) {
         DESTROY(data, buffer);
         break;
       }
@@ -629,7 +634,7 @@ void http_connection_dispatch(void* state, message_t* msg) {
     case HTTP_CONNECTION_WRITE: {
       buffer_t* buf = (buffer_t*)msg->payload;
       msg->payload = NULL; /* Take ownership — actor_run won't destroy it */
-      if (connection->sock == NULL) {
+      if (ATOMIC_LOAD(&connection->sock) == NULL) {
         DESTROY(buf, buffer);
         break;
       }
@@ -696,7 +701,7 @@ void http_connection_dispatch(void* state, message_t* msg) {
         break;
       }
       /* Try direct send */
-      ssize_t sent = platform_socket_send(connection->sock, buf->data, buf->size);
+      ssize_t sent = platform_socket_send(ATOMIC_LOAD(&connection->sock), buf->data, buf->size);
       if (sent < 0) {
         if (errno == EAGAIN || errno == EWOULDBLOCK) {
           connection->write_buffer = buf;
@@ -728,15 +733,15 @@ void http_connection_dispatch(void* state, message_t* msg) {
     }
 
     case HTTP_CONNECTION_WRITABLE: {
-      if (connection->sock == NULL) {
+      if (ATOMIC_LOAD(&connection->sock) == NULL) {
         break;
       }
       if (connection->write_buffer == NULL || connection->write_buffer->size == 0) {
         connection->write_pending = 0;
         _connection_update_watcher(connection, PD_EVENT_READ);
         break;
       }
-      ssize_t sent = platform_socket_send(connection->sock, connection->write_buffer->data,
+      ssize_t sent = platform_socket_send(ATOMIC_LOAD(&connection->sock), connection->write_buffer->data,
                            connection->write_buffer->size);
       if (sent > 0) {
         if ((size_t)sent >= connection->write_buffer->size) {
@@ -745,8 +750,8 @@ void http_connection_dispatch(void* state, message_t* msg) {
           connection->write_pending = 0;
           if (connection->is_closing) {
             /* All data flushed — finish the deferred close */
-            if (connection->sock != NULL) {
-              platform_socket_shutdown(connection->sock, PLATFORM_SHUT_WR);
+            if (ATOMIC_LOAD(&connection->sock) != NULL) {
+              platform_socket_shutdown(ATOMIC_LOAD(&connection->sock), PLATFORM_SHUT_WR);
             }
             _connection_stop_watcher(connection);
             _connection_close_fd(connection);
@@ -785,8 +790,8 @@ void http_connection_dispatch(void* state, message_t* msg) {
         _connection_update_watcher(connection, PD_EVENT_READ | PD_EVENT_WRITE);
         break;
       }
-      if (connection->sock != NULL) {
-        platform_socket_shutdown(connection->sock, PLATFORM_SHUT_WR);
+      if (ATOMIC_LOAD(&connection->sock) != NULL) {
+        platform_socket_shutdown(ATOMIC_LOAD(&connection->sock), PLATFORM_SHUT_WR);
       }
       _connection_stop_watcher(connection);
       _connection_close_fd(connection);
@@ -915,10 +920,11 @@ static void _connection_read_callback(pd_loop_t* loop, pd_watcher_t* watcher,
     if (total_read == 0) {
       /* POSIX path: synchronous recv. The socket may already be closed if the
          connection was torn down concurrently with a pending READ event. */
-      if (connection->sock == NULL) {
+      platform_socket_t* sock = ATOMIC_LOAD(&connection->sock);
+      if (sock == NULL) {
         return;
       }
-      ssize_t bytes_read = platform_socket_recv(connection->sock, buffer, sizeof(buffer));
+      ssize_t bytes_read = platform_socket_recv(sock, buffer, sizeof(buffer));
       if (bytes_read <= 0) {
         if (bytes_read == 0) {
           message_t msg;
@@ -965,7 +971,7 @@ static void _connection_do_reads(http_connection_t* connection) {
   }
   for (int batch = 0; batch < 16; batch++) {
     char buffer[READ_BUFFER_SIZE];
-    if (connection->sock == NULL) {
+    if (ATOMIC_LOAD(&connection->sock) == NULL) {
       return;
     }
 
@@ -1012,7 +1018,7 @@ http_connection_t* http_connection_create(http_server_t* server, platform_socket
   http_connection_t* connection = get_clear_memory(sizeof(http_connection_t));
   refcounter_init((refcounter_t*)connection);
   connection->server = server;
-  connection->sock = sock;
+  ATOMIC_STORE(&connection->sock, sock);
   connection->ssl = NULL;
   connection->rbio = NULL;
   connection->wbio = NULL;
@@ -1116,9 +1122,9 @@ void http_connection_destroy(http_connection_t* connection) {
         pd_timer_destroy(timer);
       }
     }
-    if (connection->sock != NULL) {
-      platform_socket_destroy(connection->sock);
-      connection->sock = NULL;
+    platform_socket_t* sock = ATOMIC_EXCHANGE(&connection->sock, NULL);
+    if (sock != NULL) {
+      platform_socket_destroy(sock);
     }
     if (connection->request != NULL) {
       DESTROY(connection->request, http_request);
@@ -1140,7 +1146,7 @@ void http_connection_destroy(http_connection_t* connection) {
 }
 
 void http_connection_write(http_connection_t* connection, const char* data, size_t length) {
-  if (connection == NULL || connection->sock == NULL) {
+  if (connection == NULL || ATOMIC_LOAD(&connection->sock) == NULL) {
     return;
   }
   buffer_t* buf = buffer_create_from_pointer_copy((uint8_t*)data, length);

src/ClientAPI/HTTP/http_connection.h

@@ -32,7 +32,7 @@ typedef struct http_connection_t {
   refcounter_t refcounter;
   actor_t actor;
   http_server_t* server;
-  platform_socket_t* sock;
+  ATOMIC(platform_socket_t*) sock;
   ATOMIC(pd_watcher_t*) watcher;
   SSL* ssl;
   /* Windows IOCP only: memory BIO pair decoupling OpenSSL from the socket so

src/ClientAPI/HTTP/http_server.c

@@ -28,7 +28,7 @@ static void _destroy_stack_init(http_server_t* server) {
 static void _destroy_stack_push_watcher(http_server_t* server, pd_watcher_t* watcher) {
   server_destroy_node_t* node = get_clear_memory(sizeof(server_destroy_node_t));
   node->watcher = watcher;
-  node->is_timer = 0;
+  node->type = 0;
   platform_mutex_lock(server->destroy_lock);
   node->next = server->destroy_head;
   server->destroy_head = node;
@@ -39,7 +39,18 @@ static void _destroy_stack_push_watcher(http_server_t* server, pd_watcher_t* wat
 static void _destroy_stack_push_timer(http_server_t* server, pd_timer_t* timer) {
   server_destroy_node_t* node = get_clear_memory(sizeof(server_destroy_node_t));
   node->timer = timer;
-  node->is_timer = 1;
+  node->type = 1;
+  platform_mutex_lock(server->destroy_lock);
+  node->next = server->destroy_head;
+  server->destroy_head = node;
+  platform_mutex_unlock(server->destroy_lock);
+  pd_loop_async_send(server->loop, NULL);
+}
+
+void http_server_defer_socket_destroy(http_server_t* server, platform_socket_t* sock) {
+  server_destroy_node_t* node = get_clear_memory(sizeof(server_destroy_node_t));
+  node->sock = sock;
+  node->type = 2;
   platform_mutex_lock(server->destroy_lock);
   node->next = server->destroy_head;
   server->destroy_head = node;
@@ -55,9 +66,11 @@ static void _destroy_stack_drain(http_server_t* server) {
   platform_mutex_unlock(server->destroy_lock);
   while (node != NULL) {
     server_destroy_node_t* next = node->next;
-    if (node->is_timer) {
+    if (node->type == 1) {
       pd_timer_stop(node->timer);
       pd_timer_destroy(node->timer);
+    } else if (node->type == 2) {
+      platform_socket_destroy(node->sock);
     } else {
       pd_watcher_destroy(node->watcher);
     }
@@ -254,9 +267,9 @@ void http_server_destroy(http_server_t* server) {
   for (int i = 0; i < server->connections.length; i++) {
     http_connection_t* conn = server->connections.data[i];
     conn->is_closing = 1;
-    if (conn->sock != NULL) {
-      platform_socket_destroy(conn->sock);
-      conn->sock = NULL;
+    platform_socket_t* sock = ATOMIC_EXCHANGE(&conn->sock, NULL);
+    if (sock != NULL) {
+      platform_socket_destroy(sock);
     }
     conn->server = NULL;
   }

src/ClientAPI/HTTP/http_server.h

@@ -31,9 +31,10 @@ typedef vec_t(http_middleware_entry_t) vec_middleware_t;
 typedef vec_t(http_connection_t*) vec_connection_t;
 
 typedef struct server_destroy_node_t {
-  pd_watcher_t* watcher;     /* valid when is_timer == 0 */
-  pd_timer_t* timer;         /* valid when is_timer == 1 */
-  uint8_t is_timer;
+  pd_watcher_t* watcher;     /* valid when type == 0 */
+  pd_timer_t* timer;         /* valid when type == 1 */
+  platform_socket_t* sock;   /* valid when type == 2 */
+  uint8_t type;              /* 0 = watcher, 1 = timer, 2 = socket */
   struct server_destroy_node_t* next;
 } server_destroy_node_t;
 
@@ -85,6 +86,10 @@ void http_server_set_timeouts(http_server_t* server, uint32_t idle_ms, uint32_t
 
 void http_server_dispatch(http_server_t* server, http_request_t* request, http_response_t* response);
 
+/* Defer a connection socket's close+free to the I/O thread's destroy stack so
+   it is never freed while the I/O-thread read callback may still be using it. */
+void http_server_defer_socket_destroy(http_server_t* server, platform_socket_t* sock);
+
 http_route_t* http_server_match_route(http_server_t* server, int method, const char* path);
 
 void http_server_use(http_server_t* server, http_middleware_t middleware, void* user_data, void (*user_data_destroy)(void*));