Don't count_down past 0

To prepare for EPOLLET

.gitea/workflows/ci.yml

@@ -46,7 +46,7 @@ jobs:
       - name: Install deps
         run: |
           sudo apt-get update
-          sudo apt-get install -y build-essential clang cmake
+          sudo apt-get install -y build-essential clang cmake gperf
 
       - name: Build and install weaseljson
         run: |

design.md

@@ -264,7 +264,7 @@ CommitRequest {
 
 1. **Request Processing**: Handler creates request-scoped arena for parsing request data
 1. **Response Generation**: Handler uses same arena for response construction (headers, JSON, etc.)
-1. **Response Queuing**: Handler calls `conn->append_message()` passing span + arena ownership
+1. **Response Queuing**: Handler calls `conn->send_response()` passing span + arena ownership
 1. **Response Writing**: I/O thread writes messages to socket, arena freed after completion
 
 > **Note**: Call `conn->reset()` periodically to reclaim arena memory. Best practice is after all outgoing bytes have been written.
@@ -411,7 +411,7 @@ public:
     Arena& arena = conn.get_arena();
 
     // Generate response
-    conn.append_message("HTTP/1.1 200 OK\r\n\r\nHello World");
+    conn.send_response("HTTP/1.1 200 OK\r\n\r\nHello World");
 
     // Server retains ownership
   }
@@ -430,7 +430,7 @@ public:
     work_queue.push([weak_conn, data = std::string(data)]() {
       // Process asynchronously - connection may be closed
       if (auto conn_ref = weak_conn.lock()) {
-        conn_ref->append_message("Async response");
+        conn_ref->send_response("Async response");
       }
     });
   }
@@ -483,7 +483,7 @@ class YesHandler : ConnectionHandler {
 public:
   void on_connection_established(Connection &conn) override {
     // Write an initial "y\n"
-    conn.append_message("y\n");
+    conn.send_response("y\n");
   }
 
   void on_write_progress(Connection &conn) override {
@@ -491,7 +491,7 @@ public:
       // Don't use an unbounded amount of memory
       conn.reset();
       // Write "y\n" repeatedly
-      conn.append_message("y\n");
+      conn.send_response("y\n");
     }
   }
 };
@@ -519,7 +519,7 @@ auto weak_conn = conn.get_weak_ref();
 background_processor.submit([weak_conn]() {
   // Do work...
   if (auto conn_ref = weak_conn.lock()) {
-    conn_ref->append_message("Background result");
+    conn_ref->send_response("Background result");
   }
   // Connection automatically cleaned up by server
 });

src/commit_pipeline.cpp

@@ -335,7 +335,7 @@ void CommitPipeline::run_release_stage(int thread_index) {
 
               // Send the JSON response using protocol-agnostic interface
               // HTTP formatting will happen in on_preprocess_writes()
-              conn_ref->send_response(commit_entry.protocol_context,
+              conn_ref->send_response(commit_entry.handle,
                                       commit_entry.response_json,
                                       std::move(commit_entry.request_arena));
             } else if constexpr (std::is_same_v<T, StatusEntry>) {
@@ -349,7 +349,7 @@ void CommitPipeline::run_release_stage(int thread_index) {
 
               // Send the JSON response using protocol-agnostic interface
               // HTTP formatting will happen in on_preprocess_writes()
-              conn_ref->send_response(status_entry.protocol_context,
+              conn_ref->send_response(status_entry.handle,
                                       status_entry.response_json,
                                       std::move(status_entry.request_arena));
             } else if constexpr (std::is_same_v<T, HealthCheckEntry>) {
@@ -364,8 +364,7 @@ void CommitPipeline::run_release_stage(int thread_index) {
               // Send the response using protocol-agnostic interface
               // HTTP formatting will happen in on_preprocess_writes()
               conn_ref->send_response(
-                  health_check_entry.protocol_context,
-                  health_check_entry.response_json,
+                  health_check_entry.handle, health_check_entry.response_json,
                   std::move(health_check_entry.request_arena));
             } else if constexpr (std::is_same_v<T, GetVersionEntry>) {
               auto &get_version_entry = e;
@@ -378,8 +377,7 @@ void CommitPipeline::run_release_stage(int thread_index) {
               // Send the response using protocol-agnostic interface
               // HTTP formatting will happen in on_preprocess_writes()
               conn_ref->send_response(
-                  get_version_entry.protocol_context,
-                  get_version_entry.response_json,
+                  get_version_entry.handle, get_version_entry.response_json,
                   std::move(get_version_entry.request_arena));
             }
           },

src/connection.cpp

@@ -1,7 +1,6 @@
 #include "connection.hpp"
 
 #include <cerrno>
-#include <climits>
 #include <cstdio>
 #include <cstdlib>
 #include <sys/epoll.h>
@@ -119,13 +118,14 @@ void Connection::append_bytes(std::span<std::string_view> data_parts,
       // I think we have to call epoll_ctl while holding mutex_. Otherwise a
       // call that clears the write interest could get reordered with one that
       // sets it and we would hang.
-      epoll_ctl(server->epoll_fds_[epoll_index_], EPOLL_CTL_MOD, fd_, &event);
+      epoll_ctl(server->event_loops_[epoll_index_].epoll_fd_, EPOLL_CTL_MOD,
+                fd_, &event);
     }
   }
 }
 
 // May be called from a foreign thread!
-void Connection::send_response(void *protocol_context,
+void Connection::send_response(ProtocolHandle handle,
                                std::string_view response_json, Arena arena) {
   std::unique_lock lock(mutex_);
 
@@ -136,7 +136,7 @@ void Connection::send_response(void *protocol_context,
 
   // Store response in queue for protocol handler processing
   pending_response_queue_.emplace_back(
-      PendingResponse{protocol_context, response_json, std::move(arena)});
+      PendingResponse{handle, response_json, std::move(arena)});
 
   // Trigger epoll interest if this is the first pending response
   if (pending_response_queue_.size() == 1) {
@@ -147,12 +147,13 @@ void Connection::send_response(void *protocol_context,
       event.data.fd = fd_;
       event.events = EPOLLIN | EPOLLOUT;
       tsan_release();
-      epoll_ctl(server->epoll_fds_[epoll_index_], EPOLL_CTL_MOD, fd_, &event);
+      epoll_ctl(server->event_loops_[epoll_index_].epoll_fd_, EPOLL_CTL_MOD,
+                fd_, &event);
     }
   }
 }
 
-int Connection::readBytes(char *buf, size_t buffer_size) {
+int Connection::read_bytes(char *buf, size_t buffer_size) {
   int r;
   for (;;) {
     r = read(fd_, buf, buffer_size);
@@ -296,7 +297,8 @@ uint32_t Connection::write_bytes() {
         // I think we have to call epoll_ctl while holding mutex_. Otherwise a
         // call that clears the write interest could get reordered with one that
         // sets it and we would hang.
-        epoll_ctl(server->epoll_fds_[epoll_index_], EPOLL_CTL_MOD, fd_, &event);
+        epoll_ctl(server->event_loops_[epoll_index_].epoll_fd_, EPOLL_CTL_MOD,
+                  fd_, &event);
       }
       // Handle shutdown modes after all messages are sent
       if (shutdown_requested_ == ConnectionShutdown::WriteOnly) {

src/connection.hpp

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <atomic>
 #include <cassert>
 #include <cstring>
 #include <deque>
@@ -32,32 +33,31 @@ enum class ConnectionShutdown {
 /**
  * Base interface for sending messages to a connection.
  * This restricted interface is safe for use by pipeline threads,
- * containing only the append_message method needed for responses.
+ * containing only the send_response method needed for responses.
  * Pipeline threads should use WeakRef<MessageSender> to safely
  * send responses without accessing other connection functionality
  * that should only be used by the I/O thread.
  */
 struct MessageSender {
   /**
-   * @brief Send response with protocol-specific context for ordering.
+   * @brief Send response with protocol-specific handle for correlation.
    *
    * Thread-safe method for pipeline threads to send responses back to clients.
    * Delegates to the connection's protocol handler for ordering logic.
    * The protocol handler may queue the response or send it immediately.
    *
-   * @param protocol_context Arena-allocated protocol-specific context
-   * @param data Response data parts (may be empty for deferred serialization)
+   * @param handle Protocol-specific handle for correlating this response
+   * @param response_json JSON response body (may be empty for deferred
+   * serialization)
    * @param arena Arena containing response data and context
    *
    * Example usage:
    * ```cpp
-   * auto* ctx = arena.allocate<HttpResponseContext>();
-   * ctx->sequence_id = 42;
-   * auto response_data = format_response(arena);
-   * conn.send_response(ctx, response_data, std::move(arena));
+   * ProtocolHandle handle = handler.allocate_response_context(arena);
+   * conn.send_response(handle, response_json, std::move(arena));
    * ```
    */
-  virtual void send_response(void *protocol_context,
+  virtual void send_response(ProtocolHandle handle,
                              std::string_view response_json, Arena arena) = 0;
 
   virtual ~MessageSender() = default;
@@ -76,9 +76,9 @@ struct MessageSender {
  *
  * Threading model:
  * - Single mutex protects state shared with pipeline threads
- * - Pipeline threads call Connection methods (append_message, etc.)
+ * - Pipeline threads call Connection methods (send_response, etc.)
  * - I/O thread processes socket events and message queue
- * - Pipeline threads register epoll write interest via append_message
+ * - Pipeline threads register epoll write interest via send_response
  * - Connection tracks closed state to prevent EBADF errors
  *
  * Arena allocator usage:
@@ -140,7 +140,7 @@ struct Connection : MessageSender {
   append_bytes(std::span<std::string_view> data_parts, Arena arena,
                ConnectionShutdown shutdown_mode = ConnectionShutdown::None);
 
-  void send_response(void *protocol_context, std::string_view response_json,
+  void send_response(ProtocolHandle handle, std::string_view response_json,
                      Arena arena) override;
 
   /**
@@ -165,7 +165,7 @@ struct Connection : MessageSender {
    *   if (auto conn = weak_conn.lock()) {
    *     Arena arena;
    *     auto response = process_request(request_data, arena);
-   *     conn->append_message({&response, 1}, std::move(arena));
+   *     conn->send_response(handle, response_json, std::move(arena));
    *   }
    * });
    * ```
@@ -261,8 +261,9 @@ private:
    *
    * Creates a new connection with the specified network address, file
    * descriptor, and associated handler. Automatically increments the global
-   * active connection counter and calls the handler's
-   * on_connection_established() method.
+   * active connection counter. The caller (Server) is responsible for
+   * initializing the self weak reference and invoking
+   * on_connection_established().
    *
    * @param addr Network address of the remote client (IPv4/IPv6 compatible)
    * @param fd File descriptor for the socket connection
@@ -278,7 +279,7 @@ private:
   friend Ref<T> make_ref(Args &&...args);
 
   // Networking interface - only accessible by Server
-  int readBytes(char *buf, size_t buffer_size);
+  int read_bytes(char *buf, size_t buffer_size);
   enum WriteBytesResult {
     Error = 1 << 0,
     Progress = 1 << 1,

src/connection_handler.hpp

@@ -8,13 +8,19 @@ struct Connection;
 
 // Include Arena header since PendingResponse uses Arena by value
 #include "arena.hpp"
+#include <cstdint>
+
+// Opaque handle used to correlate responses with protocol-specific state.
+// Each ConnectionHandler implementation defines the meaning of a handle value
+// and resolves it back to full context in on_preprocess_writes().
+using ProtocolHandle = int64_t;
 
 /**
  * Represents a response queued by pipeline threads for protocol processing.
  * Contains JSON response data that can be wrapped by any protocol.
  */
 struct PendingResponse {
-  void *protocol_context;         // Arena-allocated protocol-specific context
+  ProtocolHandle handle;          // Protocol-specific handle for correlation
   std::string_view response_json; // JSON response body (arena-allocated)
   Arena arena;                    // Arena containing response data and context
 };
@@ -42,7 +48,7 @@ public:
    * Implementation should:
    * - Create request-scoped Arena for parsing and response generation
    * - Parse incoming data using the request arena
-   * - Use conn.append_message() to queue response data to be sent
+   * - Use conn.send_response() to queue response data to be sent
    * - Handle partial messages and streaming protocols appropriately
    * - Use conn.get_weak_ref() for async processing if needed
    *

src/http_handler.cpp

@@ -52,6 +52,59 @@ HttpRequestState::HttpRequestState()
       current_header_field_buf(ArenaStlAllocator<char>(&arena)),
       current_header_value_buf(ArenaStlAllocator<char>(&arena)) {}
 
+// HttpConnectionState implementation
+HttpConnectionState::~HttpConnectionState() = default;
+
+void HttpConnectionState::register_response_context(
+    int64_t sequence_id, std::unique_ptr<HttpResponseContext> ctx) {
+  contexts_[sequence_id] = std::move(ctx);
+}
+
+HttpResponseContext *
+HttpConnectionState::resolve_response_context(ProtocolHandle handle) {
+  auto it = contexts_.find(handle);
+  if (it == contexts_.end()) {
+    return nullptr;
+  }
+  return it->second.get();
+}
+
+void HttpConnectionState::send_ordered_response(
+    Connection &conn, HttpResponseContext *ctx,
+    std::span<std::string_view> http_response, Arena arena) {
+  assert(ctx);
+  int64_t sequence_id = ctx->sequence_id;
+
+  std::unique_ptr<HttpResponseContext> owned_ctx;
+  auto it = contexts_.find(sequence_id);
+  if (it != contexts_.end()) {
+    owned_ctx = std::move(it->second);
+  } else {
+    owned_ctx.reset(ctx);
+  }
+
+  owned_ctx->ready = true;
+  owned_ctx->data = http_response;
+  owned_ctx->arena = std::move(arena);
+  contexts_[sequence_id] = std::move(owned_ctx);
+
+  // Process ready responses in order and send via append_bytes
+  auto iter = contexts_.begin();
+  while (iter != contexts_.end() && iter->first == next_sequence_to_send) {
+    auto *context = iter->second.get();
+    if (!context || !context->ready) {
+      break;
+    }
+
+    // Send through append_bytes which handles write interest
+    conn.append_bytes(context->data, std::move(context->arena),
+                      context->connection_close ? ConnectionShutdown::WriteOnly
+                                                : ConnectionShutdown::None);
+    next_sequence_to_send++;
+    iter = contexts_.erase(iter);
+  }
+}
+
 // HttpHandler implementation
 void HttpHandler::on_connection_established(Connection &conn) {
   // Allocate HTTP state using server-provided arena for connection lifecycle
@@ -72,7 +125,11 @@ void HttpHandler::on_preprocess_writes(
   // Process incoming responses and add to reorder queue
   {
     for (auto &pending : pending_responses) {
-      auto *ctx = static_cast<HttpResponseContext *>(pending.protocol_context);
+      auto *ctx = state->resolve_response_context(pending.handle);
+      // Handle stale or invalid handles defensively
+      if (!ctx) {
+        continue;
+      }
 
       // Determine HTTP status code and content type from response content
       int status_code = 200;
@@ -96,9 +153,8 @@ void HttpHandler::on_preprocess_writes(
           status_code, content_type, pending.response_json, pending.arena,
           ctx->http_request_id, ctx->connection_close);
 
-      state->send_ordered_response(conn, ctx->sequence_id, http_response,
-                                   std::move(pending.arena),
-                                   ctx->connection_close);
+      state->send_ordered_response(conn, ctx, http_response,
+                                   std::move(pending.arena));
     }
   }
 }
@@ -116,11 +172,18 @@ void HttpHandler::on_batch_complete(std::span<Connection *const> batch) {
       int64_t sequence_id = state->get_next_sequence_id();
       req.sequence_id = sequence_id;
 
-      // Create HttpResponseContext for this request
-      auto *ctx = req.arena.allocate<HttpResponseContext>(1);
+      // Create HttpResponseContext for this request; sequence_id doubles as the
+      // ProtocolHandle for async response correlation. The context is
+      // registered in HttpConnectionState so async responses can resolve it by
+      // handle.
+      auto ctx = std::make_unique<HttpResponseContext>();
       ctx->sequence_id = sequence_id;
       ctx->http_request_id = req.http_request_id;
       ctx->connection_close = req.connection_close;
+      HttpResponseContext *ctx_ptr = ctx.get();
+      req.response_context = ctx_ptr;
+      state->register_response_context(sequence_id, std::move(ctx));
+      ProtocolHandle handle = sequence_id;
 
       RouteMatch route_match;
       auto parse_result =
@@ -132,9 +195,9 @@ void HttpHandler::on_batch_complete(std::span<Connection *const> batch) {
         auto json_response = R"({"error":"Malformed URL encoding"})";
         auto http_response =
             format_json_response(400, json_response, req.arena, 0, true);
-        state->send_ordered_response(*conn, ctx->sequence_id, http_response,
-                                     std::move(req.arena),
-                                     ctx->connection_close);
+        ctx_ptr->connection_close = true;
+        state->send_ordered_response(*conn, ctx_ptr, http_response,
+                                     std::move(req.arena));
         break;
       }
       req.route = route_match.route;
@@ -176,25 +239,25 @@ void HttpHandler::on_batch_complete(std::span<Connection *const> batch) {
       // Create CommitEntry for commit requests
       if (req.route == HttpRoute::PostCommit && req.commit_request &&
           req.parsing_commit && req.basic_validation_passed) {
-        g_batch_entries.emplace_back(CommitEntry(conn->get_weak_ref(), ctx,
+        g_batch_entries.emplace_back(CommitEntry(conn->get_weak_ref(), handle,
                                                  req.commit_request.get(),
                                                  std::move(req.arena)));
       }
       // Create StatusEntry for status requests
       else if (req.route == HttpRoute::GetStatus) {
-        g_batch_entries.emplace_back(StatusEntry(conn->get_weak_ref(), ctx,
+        g_batch_entries.emplace_back(StatusEntry(conn->get_weak_ref(), handle,
                                                  req.status_request_id,
                                                  std::move(req.arena)));
       }
       // Create HealthCheckEntry for health check requests
       else if (req.route == HttpRoute::GetOk) {
-        g_batch_entries.emplace_back(
-            HealthCheckEntry(conn->get_weak_ref(), ctx, std::move(req.arena)));
+        g_batch_entries.emplace_back(HealthCheckEntry(
+            conn->get_weak_ref(), handle, std::move(req.arena)));
       }
       // Create GetVersionEntry for version requests
       else if (req.route == HttpRoute::GetVersion) {
         g_batch_entries.emplace_back(
-            GetVersionEntry(conn->get_weak_ref(), ctx, std::move(req.arena),
+            GetVersionEntry(conn->get_weak_ref(), handle, std::move(req.arena),
                             commit_pipeline_.get_committed_version()));
       }
     }
@@ -235,13 +298,17 @@ void HttpHandler::on_data_arrived(std::string_view data, Connection &conn) {
       break;
     }
     // Parse error - send response directly since this is before sequence
-    // assignment
+    // assignment. Use a temporary response context to carry metadata.
     auto json_response = R"({"error":"Bad request"})";
+    auto ctx = std::make_unique<HttpResponseContext>();
+    ctx->sequence_id = state->get_next_sequence_id();
+    ctx->http_request_id = 0;
+    ctx->connection_close = true;
     auto http_response =
-        format_json_response(400, json_response, state->pending.arena, 0, true);
-    state->send_ordered_response(conn, state->get_next_sequence_id(),
-                                 http_response, std::move(state->pending.arena),
-                                 true);
+        format_json_response(400, json_response, state->pending.arena,
+                             ctx->http_request_id, ctx->connection_close);
+    state->send_ordered_response(conn, ctx.get(), http_response,
+                                 std::move(state->pending.arena));
     return;
   }
 }
@@ -255,17 +322,18 @@ void HttpHandler::handle_get_version(Connection &, HttpRequestState &) {
 void HttpHandler::handle_post_commit(Connection &conn,
                                      HttpRequestState &state) {
   commit_counter.inc();
+  auto *ctx = state.response_context;
+  assert(ctx);
   // Check if streaming parse was successful
   if (!state.commit_request || !state.parsing_commit) {
     auto json_response = R"({"error":"Parse failed"})";
     auto http_response =
         format_json_response(400, json_response, state.arena,
-                             state.http_request_id, state.connection_close);
+                             ctx->http_request_id, ctx->connection_close);
 
     auto *conn_state = static_cast<HttpConnectionState *>(conn.user_data);
-    conn_state->send_ordered_response(conn, state.sequence_id, http_response,
-                                      std::move(state.arena),
-                                      state.connection_close);
+    conn_state->send_ordered_response(conn, ctx, http_response,
+                                      std::move(state.arena));
     return;
   }
 
@@ -310,12 +378,11 @@ void HttpHandler::handle_post_commit(Connection &conn,
                static_cast<int>(error_msg.size()), error_msg.data());
     auto http_response =
         format_json_response(400, json_response, state.arena,
-                             state.http_request_id, state.connection_close);
+                             ctx->http_request_id, ctx->connection_close);
 
     auto *conn_state = static_cast<HttpConnectionState *>(conn.user_data);
-    conn_state->send_ordered_response(conn, state.sequence_id, http_response,
-                                      std::move(state.arena),
-                                      state.connection_close);
+    conn_state->send_ordered_response(conn, ctx, http_response,
+                                      std::move(state.arena));
     return;
   }
 
@@ -326,22 +393,25 @@ void HttpHandler::handle_post_commit(Connection &conn,
 
 void HttpHandler::handle_get_subscribe(Connection &conn,
                                        HttpRequestState &state) {
+  auto *ctx = state.response_context;
+  assert(ctx);
   // TODO: Implement subscription streaming
   auto json_response =
       R"({"message":"Subscription endpoint - streaming not yet implemented"})";
   auto http_response =
       format_json_response(200, json_response, state.arena,
-                           state.http_request_id, state.connection_close);
+                           ctx->http_request_id, ctx->connection_close);
 
   auto *conn_state = static_cast<HttpConnectionState *>(conn.user_data);
-  conn_state->send_ordered_response(conn, state.sequence_id, http_response,
-                                    std::move(state.arena),
-                                    state.connection_close);
+  conn_state->send_ordered_response(conn, ctx, http_response,
+                                    std::move(state.arena));
 }
 
 void HttpHandler::handle_get_status(Connection &conn, HttpRequestState &state,
                                     const RouteMatch &route_match) {
   status_counter.inc();
+  auto *ctx = state.response_context;
+  assert(ctx);
   // Status requests are processed through the pipeline
   // Response will be generated in the sequence stage
   // This handler extracts request_id from query parameters and prepares for
@@ -354,13 +424,12 @@ void HttpHandler::handle_get_status(Connection &conn, HttpRequestState &state,
         R"({"error":"Missing required query parameter: request_id"})";
     auto http_response =
         format_json_response(400, json_response, state.arena,
-                             state.http_request_id, state.connection_close);
+                             ctx->http_request_id, ctx->connection_close);
 
     // Add directly to response queue with proper sequencing
     auto *conn_state = static_cast<HttpConnectionState *>(conn.user_data);
-    conn_state->send_ordered_response(conn, state.sequence_id, http_response,
-                                      std::move(state.arena),
-                                      state.connection_close);
+    conn_state->send_ordered_response(conn, ctx, http_response,
+                                      std::move(state.arena));
     return;
   }
 
@@ -368,13 +437,12 @@ void HttpHandler::handle_get_status(Connection &conn, HttpRequestState &state,
     auto json_response = R"({"error":"Empty request_id parameter"})";
     auto http_response =
         format_json_response(400, json_response, state.arena,
-                             state.http_request_id, state.connection_close);
+                             ctx->http_request_id, ctx->connection_close);
 
     // Add directly to response queue with proper sequencing
     auto *conn_state = static_cast<HttpConnectionState *>(conn.user_data);
-    conn_state->send_ordered_response(conn, state.sequence_id, http_response,
-                                      std::move(state.arena),
-                                      state.connection_close);
+    conn_state->send_ordered_response(conn, ctx, http_response,
+                                      std::move(state.arena));
     return;
   }
 
@@ -387,53 +455,58 @@ void HttpHandler::handle_get_status(Connection &conn, HttpRequestState &state,
 void HttpHandler::handle_put_retention(Connection &conn,
                                        HttpRequestState &state,
                                        const RouteMatch &) {
+  auto *ctx = state.response_context;
+  assert(ctx);
   // TODO: Parse retention policy from body and store
   auto json_response = R"({"policy_id":"example","status":"created"})";
   auto http_response =
       format_json_response(200, json_response, state.arena,
-                           state.http_request_id, state.connection_close);
+                           ctx->http_request_id, ctx->connection_close);
 
   // Send through reorder queue and preprocessing to maintain proper ordering
   auto *conn_state = static_cast<HttpConnectionState *>(conn.user_data);
-  conn_state->send_ordered_response(conn, state.sequence_id, http_response,
-                                    std::move(state.arena),
-                                    state.connection_close);
+  conn_state->send_ordered_response(conn, ctx, http_response,
+                                    std::move(state.arena));
 }
 
 void HttpHandler::handle_get_retention(Connection &conn,
                                        HttpRequestState &state,
                                        const RouteMatch &) {
+  auto *ctx = state.response_context;
+  assert(ctx);
   // TODO: Extract policy_id from URL or return all policies
   auto json_response = R"({"policies":[]})";
   auto http_response =
       format_json_response(200, json_response, state.arena,
-                           state.http_request_id, state.connection_close);
+                           ctx->http_request_id, ctx->connection_close);
 
   // Send through reorder queue and preprocessing to maintain proper ordering
   auto *conn_state = static_cast<HttpConnectionState *>(conn.user_data);
-  conn_state->send_ordered_response(conn, state.sequence_id, http_response,
-                                    std::move(state.arena),
-                                    state.connection_close);
+  conn_state->send_ordered_response(conn, ctx, http_response,
+                                    std::move(state.arena));
 }
 
 void HttpHandler::handle_delete_retention(Connection &conn,
                                           HttpRequestState &state,
                                           const RouteMatch &) {
+  auto *ctx = state.response_context;
+  assert(ctx);
   // TODO: Extract policy_id from URL and delete
   auto json_response = R"({"policy_id":"example","status":"deleted"})";
   auto http_response =
       format_json_response(200, json_response, state.arena,
-                           state.http_request_id, state.connection_close);
+                           ctx->http_request_id, ctx->connection_close);
 
   // Send through reorder queue and preprocessing to maintain proper ordering
   auto *conn_state = static_cast<HttpConnectionState *>(conn.user_data);
-  conn_state->send_ordered_response(conn, state.sequence_id, http_response,
-                                    std::move(state.arena),
-                                    state.connection_close);
+  conn_state->send_ordered_response(conn, ctx, http_response,
+                                    std::move(state.arena));
 }
 
 void HttpHandler::handle_get_metrics(Connection &conn,
                                      HttpRequestState &state) {
+  auto *ctx = state.response_context;
+  assert(ctx);
   metrics_counter.inc();
   auto metrics_span = metric::render(state.arena);
 
@@ -450,19 +523,19 @@ void HttpHandler::handle_get_metrics(Connection &conn,
 
   // Build HTTP headers
   std::string_view headers;
-  if (state.connection_close) {
+  if (ctx->connection_close) {
     headers = static_format(
         state.arena, "HTTP/1.1 200 OK\r\n",
         "Content-Type: text/plain; version=0.0.4\r\n",
         "Content-Length: ", static_cast<uint64_t>(total_size), "\r\n",
-        "X-Response-ID: ", static_cast<int64_t>(state.http_request_id), "\r\n",
+        "X-Response-ID: ", static_cast<int64_t>(ctx->http_request_id), "\r\n",
         "Connection: close\r\n", "\r\n");
   } else {
     headers = static_format(
         state.arena, "HTTP/1.1 200 OK\r\n",
         "Content-Type: text/plain; version=0.0.4\r\n",
         "Content-Length: ", static_cast<uint64_t>(total_size), "\r\n",
-        "X-Response-ID: ", static_cast<int64_t>(state.http_request_id), "\r\n",
+        "X-Response-ID: ", static_cast<int64_t>(ctx->http_request_id), "\r\n",
         "Connection: keep-alive\r\n", "\r\n");
   }
 
@@ -472,9 +545,7 @@ void HttpHandler::handle_get_metrics(Connection &conn,
   }
 
   auto *conn_state = static_cast<HttpConnectionState *>(conn.user_data);
-  conn_state->send_ordered_response(conn, state.sequence_id, result,
-                                    std::move(state.arena),
-                                    state.connection_close);
+  conn_state->send_ordered_response(conn, ctx, result, std::move(state.arena));
 }
 
 void HttpHandler::handle_get_ok(Connection &, HttpRequestState &) {
@@ -485,41 +556,17 @@ void HttpHandler::handle_get_ok(Connection &, HttpRequestState &) {
 }
 
 void HttpHandler::handle_not_found(Connection &conn, HttpRequestState &state) {
+  auto *ctx = state.response_context;
+  assert(ctx);
   not_found_counter.inc();
   auto json_response = R"({"error":"Not found"})";
   auto http_response =
       format_json_response(404, json_response, state.arena,
-                           state.http_request_id, state.connection_close);
+                           ctx->http_request_id, ctx->connection_close);
 
   auto *conn_state = static_cast<HttpConnectionState *>(conn.user_data);
-  conn_state->send_ordered_response(conn, state.sequence_id, http_response,
-                                    std::move(state.arena),
-                                    state.connection_close);
-}
-
-void HttpConnectionState::send_ordered_response(
-    Connection &conn, int64_t sequence_id,
-    std::span<std::string_view> http_response, Arena arena,
-    bool close_connection) {
-
-  // Add to reorder queue with proper sequencing
-  ready_responses[sequence_id] =
-      ResponseData{http_response, std::move(arena), close_connection};
-
-  // Process ready responses in order and send via append_bytes
-  auto iter = ready_responses.begin();
-  while (iter != ready_responses.end() &&
-         iter->first == next_sequence_to_send) {
-    auto &[sequence_id, response_data] = *iter;
-
-    // Send through append_bytes which handles write interest
-    conn.append_bytes(response_data.data, std::move(response_data.arena),
-                      response_data.connection_close
-                          ? ConnectionShutdown::WriteOnly
-                          : ConnectionShutdown::None);
-    next_sequence_to_send++;
-    iter = ready_responses.erase(iter);
-  }
+  conn_state->send_ordered_response(conn, ctx, http_response,
+                                    std::move(state.arena));
 }
 
 std::span<std::string_view>

src/http_handler.hpp

@@ -1,6 +1,7 @@
 #pragma once
 
 #include <map>
+#include <memory>
 #include <string_view>
 
 #include <llhttp.h>
@@ -19,22 +20,18 @@ struct RouteMatch;
 
 /**
  * HTTP-specific response context stored in pipeline entries.
- * Arena-allocated and passed through pipeline for response correlation.
+ * Arena-allocated and identified by a ProtocolHandle (sequence_id).
+ * Also owns the response data once it becomes ready.
  */
 struct HttpResponseContext {
   int64_t sequence_id;     // For response ordering in pipelining
   int64_t http_request_id; // For X-Response-ID header
   bool connection_close;   // Whether to close connection after response
-};
 
-/**
- * Response data ready to send (sequence_id -> response data).
- * Absence from map indicates response not ready yet.
- */
-struct ResponseData {
+  // Response payload; populated when the response is ready.
+  bool ready = false;
   std::span<std::string_view> data;
   Arena arena;
-  bool connection_close;
 };
 
 /**
@@ -69,6 +66,10 @@ struct HttpRequestState {
       0;                   // X-Request-Id header value (for tracing/logging)
   int64_t sequence_id = 0; // Assigned for response ordering in pipelining
 
+  // HTTP response context for this request (set during batch processing).
+  // Non-owning: the context is owned by HttpConnectionState::contexts_.
+  HttpResponseContext *response_context = nullptr;
+
   // Streaming parser for POST requests
   Arena::Ptr<JsonCommitRequestParser> commit_parser;
   Arena::Ptr<CommitRequest> commit_request;
@@ -89,15 +90,27 @@ struct HttpConnectionState {
   int64_t get_next_sequence_id() { return next_sequence_id++; }
 
   HttpConnectionState();
+  ~HttpConnectionState();
 
-  void send_ordered_response(Connection &conn, int64_t sequence_id,
+  // Register an HttpResponseContext so the pipeline can resolve it by handle.
+  void register_response_context(int64_t sequence_id,
+                                 std::unique_ptr<HttpResponseContext> ctx);
+
+  // Resolve a ProtocolHandle back to its HttpResponseContext.
+  HttpResponseContext *resolve_response_context(ProtocolHandle handle);
+
+  // Mark a context's response data as ready and try to send in-order.
+  // The context may already be registered in contexts_; if so, it is moved out
+  // and populated. A transient context pointer (e.g., for parse errors) works
+  // too.
+  void send_ordered_response(Connection &conn, HttpResponseContext *ctx,
                              std::span<std::string_view> http_response,
-                             Arena arena, bool close_connection);
+                             Arena arena);
 
 private:
-  // Response ordering for HTTP pipelining
-  std::map<int64_t, ResponseData>
-      ready_responses; // sequence_id -> response data
+  // All registered response contexts keyed by sequence_id.
+  // A context stays here until its response has been dispatched.
+  std::map<int64_t, std::unique_ptr<HttpResponseContext>> contexts_;
   int64_t next_sequence_to_send = 0;
   int64_t next_sequence_id = 0;
 };

src/json_commit_request_parser.hpp

@@ -3,7 +3,7 @@
 #include <memory>
 
 #include <simdutf.h>
-#include <weaseljson/weaseljson.h>
+#include <weaseljson.h>
 
 #include "commit_request_parser.hpp"
 #include "json_token_enum.hpp"

src/pipeline_entry.hpp

@@ -17,8 +17,8 @@ struct CommitEntry {
   bool resolve_success = false;  // Set by resolve stage
   bool persist_success = false;  // Set by persist stage
 
-  // Protocol-agnostic context (arena-allocated, protocol-specific)
-  void *protocol_context = nullptr;
+  // Protocol-agnostic handle for correlating the response
+  ProtocolHandle handle = -1;
   const CommitRequest *commit_request = nullptr; // Points to request_arena data
 
   // Request arena contains parsed request data and response data
@@ -28,9 +28,9 @@ struct CommitEntry {
   std::string_view response_json;
 
   CommitEntry() = default; // Default constructor for variant
-  explicit CommitEntry(WeakRef<MessageSender> conn, void *ctx,
+  explicit CommitEntry(WeakRef<MessageSender> conn, ProtocolHandle handle,
                        const CommitRequest *req, Arena arena)
-      : connection(std::move(conn)), protocol_context(ctx), commit_request(req),
+      : connection(std::move(conn)), handle(handle), commit_request(req),
         request_arena(std::move(arena)) {}
 };
 
@@ -42,8 +42,8 @@ struct StatusEntry {
   WeakRef<MessageSender> connection;
   int64_t version_upper_bound = 0; // Set by sequence stage
 
-  // Protocol-agnostic context (arena-allocated, protocol-specific)
-  void *protocol_context = nullptr;
+  // Protocol-agnostic handle for correlating the response
+  ProtocolHandle handle = -1;
   std::string_view status_request_id; // Points to request_arena data
 
   // Request arena for request data
@@ -53,9 +53,9 @@ struct StatusEntry {
   std::string_view response_json;
 
   StatusEntry() = default; // Default constructor for variant
-  explicit StatusEntry(WeakRef<MessageSender> conn, void *ctx,
+  explicit StatusEntry(WeakRef<MessageSender> conn, ProtocolHandle handle,
                        std::string_view request_id, Arena arena)
-      : connection(std::move(conn)), protocol_context(ctx),
+      : connection(std::move(conn)), handle(handle),
         status_request_id(request_id), request_arena(std::move(arena)) {}
 };
 
@@ -67,8 +67,8 @@ struct StatusEntry {
 struct HealthCheckEntry {
   WeakRef<MessageSender> connection;
 
-  // Protocol-agnostic context (arena-allocated, protocol-specific)
-  void *protocol_context = nullptr;
+  // Protocol-agnostic handle for correlating the response
+  ProtocolHandle handle = -1;
 
   // Request arena for response data
   Arena request_arena;
@@ -77,8 +77,9 @@ struct HealthCheckEntry {
   std::string_view response_json;
 
   HealthCheckEntry() = default; // Default constructor for variant
-  explicit HealthCheckEntry(WeakRef<MessageSender> conn, void *ctx, Arena arena)
-      : connection(std::move(conn)), protocol_context(ctx),
+  explicit HealthCheckEntry(WeakRef<MessageSender> conn, ProtocolHandle handle,
+                            Arena arena)
+      : connection(std::move(conn)), handle(handle),
         request_arena(std::move(arena)) {}
 };
 
@@ -89,8 +90,8 @@ struct HealthCheckEntry {
 struct GetVersionEntry {
   WeakRef<MessageSender> connection;
 
-  // Protocol-agnostic context (arena-allocated, protocol-specific)
-  void *protocol_context = nullptr;
+  // Protocol-agnostic handle for correlating the response
+  ProtocolHandle handle = -1;
 
   // Request arena for response data
   Arena request_arena;
@@ -102,9 +103,9 @@ struct GetVersionEntry {
   int64_t version;
 
   GetVersionEntry() = default; // Default constructor for variant
-  explicit GetVersionEntry(WeakRef<MessageSender> conn, void *ctx, Arena arena,
-                           int64_t version)
-      : connection(std::move(conn)), protocol_context(ctx),
+  explicit GetVersionEntry(WeakRef<MessageSender> conn, ProtocolHandle handle,
+                           Arena arena, int64_t version)
+      : connection(std::move(conn)), handle(handle),
         request_arena(std::move(arena)), version(version) {}
 };
 

src/server.cpp

@@ -1,6 +1,5 @@
 #include "server.hpp"
 
-#include <csignal>
 #include <cstdio>
 #include <cstdlib>
 #include <cstring>
@@ -75,7 +74,7 @@ Server::~Server() {
   }
 
   // Close all epoll instances
-  for (int epollfd : epoll_fds_) {
+  for (auto [epollfd] : event_loops_) {
     if (epollfd != -1) {
       int e = close(epollfd);
       if (e == -1 && errno != EINTR) {
@@ -84,7 +83,7 @@ Server::~Server() {
       }
     }
   }
-  epoll_fds_.clear();
+  event_loops_.clear();
 
   // Close all listen sockets (Server always owns them)
   for (int fd : listen_fds_) {
@@ -167,7 +166,7 @@ int Server::create_local_connection() {
   // Use round-robin distribution for local connections across epoll instances
   size_t epoll_index =
       connection_distribution_counter_.fetch_add(1, std::memory_order_relaxed) %
-      epoll_fds_.size();
+      event_loops_.size();
 
   // Create Connection object
   auto connection = make_ref<Connection>(
@@ -184,7 +183,7 @@ int Server::create_local_connection() {
   event.events = EPOLLIN;
   event.data.fd = server_fd;
 
-  int epollfd = epoll_fds_[epoll_index];
+  int epollfd = event_loops_[epoll_index].epoll_fd_;
   if (epoll_ctl(epollfd, EPOLL_CTL_ADD, server_fd, &event) == -1) {
     perror("epoll_ctl ADD local connection");
     connection_registry_.remove(server_fd);
@@ -221,11 +220,11 @@ void Server::setup_shutdown_pipe() {
 void Server::create_epoll_instances() {
   // Create one epoll instance per I/O thread (1:1 mapping) to eliminate
   // contention
-  epoll_fds_.resize(config_.server.io_threads);
+  event_loops_.resize(config_.server.io_threads);
 
   for (int i = 0; i < config_.server.io_threads; ++i) {
-    epoll_fds_[i] = epoll_create1(EPOLL_CLOEXEC);
-    if (epoll_fds_[i] == -1) {
+    event_loops_[i].epoll_fd_ = epoll_create1(EPOLL_CLOEXEC);
+    if (event_loops_[i].epoll_fd_ == -1) {
       perror("epoll_create1");
       std::abort();
     }
@@ -235,7 +234,7 @@ void Server::create_epoll_instances() {
     shutdown_event.events = EPOLLIN;
     shutdown_event.data.fd = shutdown_pipe_[0];
 
-    if (epoll_ctl(epoll_fds_[i], EPOLL_CTL_ADD, shutdown_pipe_[0],
+    if (epoll_ctl(event_loops_[i].epoll_fd_, EPOLL_CTL_ADD, shutdown_pipe_[0],
                   &shutdown_event) == -1) {
       perror("epoll_ctl shutdown pipe");
       std::abort();
@@ -247,8 +246,8 @@ void Server::create_epoll_instances() {
       struct epoll_event listen_event;
       listen_event.events = EPOLLIN | EPOLLEXCLUSIVE;
       listen_event.data.fd = listen_fd;
-      if (epoll_ctl(epoll_fds_[i], EPOLL_CTL_ADD, listen_fd, &listen_event) ==
-          -1) {
+      if (epoll_ctl(event_loops_[i].epoll_fd_, EPOLL_CTL_ADD, listen_fd,
+                    &listen_event) == -1) {
         perror("epoll_ctl listen socket");
         std::abort();
       }
@@ -265,7 +264,7 @@ void Server::start_io_threads(std::vector<std::thread> &threads) {
                          ("io-" + std::to_string(thread_id)).c_str());
 
       // Each thread uses its assigned epoll instance (1:1 mapping)
-      int epollfd = epoll_fds_[thread_id];
+      int epollfd = event_loops_[thread_id].epoll_fd_;
 
       std::vector<epoll_event> events(config_.server.event_batch_size);
       std::vector<Ref<Connection>> batch(config_.server.event_batch_size);
@@ -410,7 +409,10 @@ void Server::process_connection_reads(Ref<Connection> &conn, int events) {
     auto buf_size = config_.server.read_buffer_size;
     g_read_buffer.resize(buf_size);
     char *buf = g_read_buffer.data();
-    int r = conn->readBytes(buf, buf_size);
+
+    // Once we do EPOLLET we must drain the socket until read returns EAGAIN.
+    for (;;) {
+      int r = conn->read_bytes(buf, buf_size);
 
       if (r < 0) {
         // Error or EOF - connection should be closed
@@ -419,12 +421,18 @@ void Server::process_connection_reads(Ref<Connection> &conn, int events) {
       }
 
       if (r == 0) {
-      // No data available (EAGAIN) - skip read processing but continue
+        // No data available (EAGAIN) - read side drained
         return;
       }
 
-    // Call handler with connection reference - server retains ownership
+      // Call handler with connection reference - server retains ownership.
       handler_.on_data_arrived(std::string_view{buf, size_t(r)}, *conn);
+
+      // The connection may have been closed by the handler; stop reading.
+      if (!conn) {
+        return;
+      }
+    }
   }
 }
 

src/server.hpp

@@ -131,14 +131,17 @@ private:
   // Shutdown coordination
   int shutdown_pipe_[2] = {-1, -1};
 
+  struct EventLoopState {
+    int epoll_fd_;
+  };
+
   // Multiple epoll file descriptors (1:1 with I/O threads) to reduce contention
-  std::vector<int> epoll_fds_;
+  std::vector<EventLoopState> event_loops_;
   std::vector<int>
       listen_fds_; // FDs to accept connections on (Server owns these)
 
   // Private helper methods
   void setup_shutdown_pipe();
-  void setup_signal_handling();
   void create_epoll_instances();
   void start_io_threads(std::vector<std::thread> &threads);
 

tests/test_server.cpp

@@ -3,22 +3,36 @@
 #include "connection_handler.hpp"
 #include "server.hpp"
 
+#include <atomic>
 #include <doctest/doctest.h>
 #include <latch>
 #include <string_view>
 #include <thread>
 
+struct Event {
+  void wait() { done_.wait(); }
+  void set() {
+    if (!set_.exchange(true, std::memory_order_relaxed)) {
+      done_.count_down();
+    }
+  }
+
+private:
+  std::latch done_{1};
+  std::atomic<bool> set_;
+};
+
 struct EchoHandler : ConnectionHandler {
   Arena arena;
   std::span<std::string_view> reply;
   WeakRef<MessageSender> wconn;
-  std::latch done{1};
+  Event done;
   void on_data_arrived(std::string_view data, Connection &conn) override {
     reply = arena.allocate_span<std::string_view>(1);
     reply[0] = arena.copy_string(data);
     wconn = conn.get_weak_ref();
     CHECK(wconn.lock());
-    done.count_down();
+    done.set();
   }
 };
 
@@ -60,8 +74,8 @@ struct ShutdownTestHandler : ConnectionHandler {
   Arena arena;
   std::span<std::string_view> reply;
   WeakRef<MessageSender> wconn;
-  std::latch received_data{1};
-  std::latch connection_closed_latch{1};
+  Event received_data;
+  Event connection_closed_latch;
   ConnectionShutdown shutdown_mode = ConnectionShutdown::None;
   std::atomic<bool> connection_closed{false};
 
@@ -69,12 +83,12 @@ struct ShutdownTestHandler : ConnectionHandler {
     reply = arena.allocate_span<std::string_view>(1);
     reply[0] = arena.copy_string(data);
     wconn = conn.get_weak_ref();
-    received_data.count_down();
+    received_data.set();
   }
 
   void on_connection_closed(Connection &) override {
     connection_closed = true;
-    connection_closed_latch.count_down();
+    connection_closed_latch.set();
   }
 };
 

tools/check_snake_case.py

@@ -78,7 +78,7 @@ def check_snake_case_violations(filepath, check_new_only=True):
         # Common HTTP parser callback names (external API)
         r"\b(onUrl|onHeaderField|onHeaderFieldComplete|onHeaderValue|onHeaderValueComplete|onHeadersComplete|onBody|onMessageComplete)\b",
         # Known legacy APIs we can't easily change
-        r"\b(user_data|get_arena|append_message)\b",
+        r"\b(user_data|get_arena|send_response)\b",
     ]
 
     try: