Don't count_down past 0

To prepare for EPOLLET

.gitea/workflows/ci.yml

@@ -0,0 +1,69 @@
+name: CI
+
+on: [push, pull_request]
+
+jobs:
+  pre-commit:
+    runs-on: ubuntu-latest-amd64
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Install pre-commit
+        run: pipx install pre-commit
+
+      - name: Run pre-commit
+        run: ~/.local/bin/pre-commit run --all-files --show-diff-on-failure
+
+  build:
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - name: clang-amd64
+            runner: ubuntu-latest-amd64
+            cmake_args: -DCMAKE_C_COMPILER=clang -DCMAKE_CXX_COMPILER=clang++
+          - name: clang-arm64
+            runner: ubuntu-latest-arm64
+            cmake_args: -DCMAKE_C_COMPILER=clang -DCMAKE_CXX_COMPILER=clang++
+          - name: gcc-amd64
+            runner: ubuntu-latest-amd64
+            cmake_args: -DCMAKE_C_COMPILER=gcc -DCMAKE_CXX_COMPILER=g++
+          - name: gcc-arm64
+            runner: ubuntu-latest-arm64
+            cmake_args: -DCMAKE_C_COMPILER=gcc -DCMAKE_CXX_COMPILER=g++
+    runs-on: ${{ matrix.runner }}
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          path: weaseldb
+
+      - name: Checkout weaseljson
+        uses: actions/checkout@v4
+        with:
+          repository: weaselab/weaseljson
+          path: weaseljson
+
+      - name: Install deps
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y build-essential clang cmake gperf
+
+      - name: Build and install weaseljson
+        run: |
+          cmake -S weaseljson -B weaseljson/build \
+            -DCMAKE_BUILD_TYPE=Release \
+            -DCMAKE_INSTALL_PREFIX="$(pwd)/weaseljson/install"
+          cmake --build weaseljson/build -j "$(nproc)"
+          cmake --install weaseljson/build
+
+      - name: Build weaseldb
+        run: |
+          cmake -S weaseldb -B weaseldb/build ${{ matrix.cmake_args }} \
+            -DCMAKE_POLICY_VERSION_MINIMUM=3.5 \
+            -DCMAKE_PREFIX_PATH="$(pwd)/weaseljson/install"
+          cmake --build weaseldb/build -j "$(nproc)"
+
+      - name: Test
+        run: |
+          cd weaseldb/build
+          ctest --output-on-failure -j "$(nproc)" --timeout 90

CMakeLists.txt

@@ -80,10 +80,9 @@ FetchContent_MakeAvailable(simdutf)
 
 FetchContent_Declare(
   llhttp
-  URL "https://github.com/nodejs/llhttp/archive/refs/tags/release/v9.2.1.tar.gz"
+  GIT_REPOSITORY https://github.com/nodejs/llhttp.git
-  URL_HASH
+  GIT_TAG 610a87d755f6bae466cd871c2ba97574ccac5483 # release/v9.2.1
-    SHA256=3c163891446e529604b590f9ad097b2e98b5ef7e4d3ddcf1cf98b62ca668f23e
+)
-  DOWNLOAD_EXTRACT_TIMESTAMP ON)
 set(BUILD_SHARED_LIBS
     OFF
     CACHE INTERNAL "")

benchmarks/bench_thread_pipeline.cpp

@@ -24,15 +24,15 @@ int main() {
       }
     });
 
-    StaticThreadPipeline<std::latch *, WaitStrategy::WaitIfStageEmpty, 1>
+    ThreadPipeline<std::latch *> pipeline(WaitStrategy::WaitIfStageEmpty, {1},
-        pipeline(LOG_PIPELINE_SIZE);
+                                          LOG_PIPELINE_SIZE);
 
     std::latch done{0};
 
     // Stage 0 consumer thread
     std::thread stage0_thread([&pipeline, &done]() {
       for (;;) {
-        auto guard = pipeline.acquire<0, 0>();
+        auto guard = pipeline.acquire(0, 0);
 
         for (auto &item : guard.batch) {
           spend_cpu_cycles(BUSY_ITERS);
@@ -89,15 +89,15 @@ int main() {
                      .warmup(100);
 
     for (int batch_size : {1, 4, 16, 64, 256}) {
-      StaticThreadPipeline<std::latch *, WaitStrategy::WaitIfStageEmpty, 1>
+      ThreadPipeline<std::latch *> pipeline(WaitStrategy::WaitIfStageEmpty, {1},
-          pipeline(LOG_PIPELINE_SIZE);
+                                            LOG_PIPELINE_SIZE);
 
       std::latch done{0};
 
       // Stage 0 consumer thread
       std::thread stage0_thread([&pipeline, &done]() {
         for (;;) {
-          auto guard = pipeline.acquire<0, 0>();
+          auto guard = pipeline.acquire(0, 0);
 
           for (auto &item : guard.batch) {
             spend_cpu_cycles(BUSY_ITERS);
@@ -142,74 +142,73 @@ int main() {
   }
 
   // Helper function for wait strategy benchmarks
-  auto benchmark_wait_strategy =
+  auto benchmark_wait_strategy = [](WaitStrategy strategy,
-      []<WaitStrategy strategy>(const std::string &name,
+                                    const std::string &name,
-                                ankerl::nanobench::Bench &bench) {
+                                    ankerl::nanobench::Bench &bench) {
-        constexpr int LOG_PIPELINE_SIZE =
+    constexpr int LOG_PIPELINE_SIZE =
-            8; // Smaller buffer to increase contention
+        8; // Smaller buffer to increase contention
-        constexpr int NUM_ITEMS = 50'000;
+    constexpr int NUM_ITEMS = 50'000;
-        constexpr int BATCH_SIZE = 4; // Small batches to increase coordination
+    constexpr int BATCH_SIZE = 4; // Small batches to increase coordination
-        constexpr int BUSY_ITERS =
+    constexpr int BUSY_ITERS =
-            10; // Light work to emphasize coordination overhead
+        10; // Light work to emphasize coordination overhead
 
-        StaticThreadPipeline<std::latch *, strategy, 1, 1> pipeline(
+    ThreadPipeline<std::latch *> pipeline(strategy, {1, 1}, LOG_PIPELINE_SIZE);
-            LOG_PIPELINE_SIZE);
 
-        std::latch done{0};
+    std::latch done{0};
 
-        // Stage 0 worker
+    // Stage 0 worker
-        std::thread stage0_thread([&pipeline, &done]() {
+    std::thread stage0_thread([&pipeline, &done]() {
-          for (;;) {
+      for (;;) {
-            auto guard = pipeline.template acquire<0, 0>();
+        auto guard = pipeline.acquire(0, 0);
-            for (auto &item : guard.batch) {
+        for (auto &item : guard.batch) {
-              spend_cpu_cycles(BUSY_ITERS);
+          spend_cpu_cycles(BUSY_ITERS);
-              if (item == &done)
+          if (item == &done)
-                return;
+            return;
-            }
-          }
-        });
-
-        // Stage 1 worker (final stage - always calls futex wake)
-        std::thread stage1_thread([&pipeline, &done]() {
-          for (;;) {
-            auto guard = pipeline.template acquire<1, 0>();
-            for (auto &item : guard.batch) {
-              spend_cpu_cycles(BUSY_ITERS);
-              if (item == &done)
-                return;
-              if (item)
-                item->count_down();
-            }
-          }
-        });
-
-        bench.run(name, [&] {
-          int items_pushed = 0;
-          while (items_pushed < NUM_ITEMS - 1) {
-            auto guard = pipeline.push(
-                std::min(NUM_ITEMS - 1 - items_pushed, BATCH_SIZE), true);
-            auto it = guard.batch.begin();
-            items_pushed += guard.batch.size();
-            for (size_t i = 0; i < guard.batch.size(); ++i, ++it) {
-              *it = nullptr;
-            }
-          }
-          std::latch finish{1};
-          {
-            auto guard = pipeline.push(1, true);
-            guard.batch[0] = &finish;
-          }
-          finish.wait();
-        });
-
-        // Shutdown
-        {
-          auto guard = pipeline.push(1, true);
-          guard.batch[0] = &done;
         }
-        stage0_thread.join();
+      }
-        stage1_thread.join();
+    });
-      };
+
+    // Stage 1 worker (final stage - always calls futex wake)
+    std::thread stage1_thread([&pipeline, &done]() {
+      for (;;) {
+        auto guard = pipeline.acquire(1, 0);
+        for (auto &item : guard.batch) {
+          spend_cpu_cycles(BUSY_ITERS);
+          if (item == &done)
+            return;
+          if (item)
+            item->count_down();
+        }
+      }
+    });
+
+    bench.run(name, [&] {
+      int items_pushed = 0;
+      while (items_pushed < NUM_ITEMS - 1) {
+        auto guard = pipeline.push(
+            std::min(NUM_ITEMS - 1 - items_pushed, BATCH_SIZE), true);
+        auto it = guard.batch.begin();
+        items_pushed += guard.batch.size();
+        for (size_t i = 0; i < guard.batch.size(); ++i, ++it) {
+          *it = nullptr;
+        }
+      }
+      std::latch finish{1};
+      {
+        auto guard = pipeline.push(1, true);
+        guard.batch[0] = &finish;
+      }
+      finish.wait();
+    });
+
+    // Shutdown
+    {
+      auto guard = pipeline.push(1, true);
+      guard.batch[0] = &done;
+    }
+    stage0_thread.join();
+    stage1_thread.join();
+  };
 
   // Wait strategy comparison benchmark - multiple stages to trigger futex wakes
   {
@@ -220,12 +219,11 @@ int main() {
                      .relative(true)
                      .warmup(50);
 
-    benchmark_wait_strategy.template operator()<WaitStrategy::WaitIfStageEmpty>(
+    benchmark_wait_strategy(WaitStrategy::WaitIfStageEmpty, "WaitIfStageEmpty",
-        "WaitIfStageEmpty", bench);
+                            bench);
-    benchmark_wait_strategy.template
+    benchmark_wait_strategy(WaitStrategy::WaitIfUpstreamIdle,
-    operator()<WaitStrategy::WaitIfUpstreamIdle>("WaitIfUpstreamIdle", bench);
+                            "WaitIfUpstreamIdle", bench);
-    benchmark_wait_strategy.template operator()<WaitStrategy::Never>("Never",
+    benchmark_wait_strategy(WaitStrategy::Never, "Never", bench);
-                                                                     bench);
   }
 
   // TODO: Add more benchmarks for:

config.md

@@ -28,13 +28,15 @@ Controls server networking, threading, and request handling behavior.
 
 ### Commit Configuration (`[commit]`)
 
-Controls behavior of the `/v1/commit` endpoint and request ID management.
+Controls behavior of the `/v1/commit` endpoint, request ID management, and commit pipeline threading.
 
 | Parameter | Type | Default | Description |
 |-----------|------|---------|-------------|
 | `min_request_id_length` | integer | `20` | Minimum length required for client-provided `request_id` fields to ensure sufficient entropy for collision avoidance |
 | `request_id_retention_hours` | integer | `24` | How long to retain request IDs in memory for `/v1/status` queries. Longer retention reduces the chance of `log_truncated` responses |
 | `request_id_retention_versions` | integer | `100000000` | Minimum number of versions to retain request IDs for, regardless of time. Provides additional protection against `log_truncated` responses |
+| `pipeline_wait_strategy` | string | `"WaitIfUpstreamIdle"` | Wait strategy for the commit pipeline. `"WaitIfStageEmpty"` = block when individual stages are empty (safe for shared CPUs), `"WaitIfUpstreamIdle"` = block only when all upstream stages are idle (requires dedicated cores, highest throughput), `"Never"` = never block, busy-wait continuously (requires dedicated cores, lowest latency) |
+| `pipeline_release_threads` | integer | `1` | Number of threads in the release stage (final stage of commit pipeline). Higher values increase parallelism for connection release and response transmission |
 
 ### Subscription Configuration (`[subscription]`)
 
@@ -77,6 +79,8 @@ read_buffer_size = 32768  # 32KB
 min_request_id_length = 32
 request_id_retention_hours = 48
 request_id_retention_versions = 50000
+pipeline_wait_strategy = "WaitIfUpstreamIdle"  # Options: "WaitIfStageEmpty", "WaitIfUpstreamIdle", "Never"
+pipeline_release_threads = 4  # Default: 1, increase for higher throughput
 
 [subscription]
 max_buffer_size_bytes = 52428800  # 50MB
@@ -115,6 +119,11 @@ These configuration parameters directly affect server and API behavior:
 
 - **`request_id_retention_*`**: Affects availability of data for `/v1/status` queries and likelihood of `log_truncated` responses
 
+**Commit Pipeline Performance:**
+
+- **`pipeline_wait_strategy`**: Controls CPU usage vs latency tradeoff in commit processing. `WaitIfStageEmpty` is safest for shared CPUs, `WaitIfUpstreamIdle` provides highest throughput with dedicated cores, `Never` provides lowest latency but uses 100% CPU
+- **`pipeline_release_threads`**: Determines parallelism in the final stage of commit processing. More threads can improve throughput when processing many concurrent requests
+
 **Subscription Streaming:**
 
 - **`max_buffer_size_bytes`**: Controls when `/v1/subscribe` connections are terminated due to slow consumption
@@ -137,6 +146,8 @@ The configuration system includes comprehensive validation with specific bounds
 - **`min_request_id_length`**: Must be between 8 and 256 characters
 - **`request_id_retention_hours`**: Must be between 1 and 8760 hours (1 year)
 - **`request_id_retention_versions`**: Must be > 0
+- **`pipeline_wait_strategy`**: Must be one of: `"WaitIfStageEmpty"`, `"WaitIfUpstreamIdle"`, or `"Never"`
+- **`pipeline_release_threads`**: Must be between 1 and 64
 
 ### Subscription Configuration Limits
 

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
 });

reproduce_threading_report.sh

@@ -0,0 +1,89 @@
+#!/bin/bash
+# Reproduce the threading performance report results.
+# Run from the project root, e.g.:
+#   ./reproduce_threading_report.sh
+set -euo pipefail
+
+BUILD_DIR="build"
+CONFIG="test_benchmark_config.toml"
+DURATION=120
+
+if [ ! -d "$BUILD_DIR" ]; then
+    echo "Error: build directory '$BUILD_DIR' not found. Build the project first." >&2
+    exit 1
+fi
+
+if [ ! -f "$CONFIG" ]; then
+    echo "Error: config '$CONFIG' not found. Run this script from the project root." >&2
+    exit 1
+fi
+
+if [ ! -x "$BUILD_DIR/weaseldb" ] || [ ! -x "$BUILD_DIR/load_tester" ]; then
+    echo "Error: required binaries not found in '$BUILD_DIR'. Build the project first." >&2
+    exit 1
+fi
+
+cd "$BUILD_DIR"
+
+# Increase file descriptor limit for high concurrency. Best-effort only:
+# it may fail if the hard limit is lower, especially in containers.
+ulimit -n 65536 2>/dev/null || echo "Warning: could not raise ulimit -n (continuing)" >&2
+
+# Clean up any leftover socket or server log from a previous run
+rm -f weaseldb.sock server.log
+
+SERVER_PID=""
+
+cleanup() {
+    if [ -n "$SERVER_PID" ] && kill -0 "$SERVER_PID" 2>/dev/null; then
+        echo ""
+        echo "=== Stopping server ==="
+        kill "$SERVER_PID" 2>/dev/null || true
+        wait "$SERVER_PID" 2>/dev/null || true
+    fi
+    echo "=== Server log tail ==="
+    tail -50 server.log 2>/dev/null || true
+}
+trap cleanup EXIT INT TERM
+
+echo "=== Starting WeaselDB server ==="
+./weaseldb --config "../$CONFIG" > server.log 2>&1 &
+SERVER_PID=$!
+
+echo "Server PID: $SERVER_PID"
+
+# Wait for server to be ready (unix socket created)
+for i in {1..30}; do
+    if [ -S weaseldb.sock ]; then
+        echo "Server ready after $((i * 100))ms"
+        break
+    fi
+    if ! kill -0 "$SERVER_PID" 2>/dev/null; then
+        echo "Server died unexpectedly"
+        cat server.log
+        exit 1
+    fi
+    sleep 0.1
+done
+
+if [ ! -S weaseldb.sock ]; then
+    echo "Error: server failed to create socket within 3 seconds" >&2
+    exit 1
+fi
+
+echo "=== Server log (first lines) ==="
+head -30 server.log
+
+echo ""
+echo "=== Running load tester ==="
+./load_tester \
+    --unix-socket weaseldb.sock \
+    --concurrency 2000 \
+    --requests-per-conn 500 \
+    --connect-threads 2 \
+    --network-threads 10 \
+    --duration "$DURATION" \
+    --stats-interval 1
+
+echo ""
+echo "=== Load test complete ==="

src/commit_pipeline.cpp

@@ -17,7 +17,9 @@ auto banned_request_ids_memory_gauge =
         .create({});
 
 CommitPipeline::CommitPipeline(const weaseldb::Config &config)
-    : config_(config), pipeline_(lg_size) {
+    : config_(config),
+      pipeline_(config.commit.pipeline_wait_strategy,
+                {1, 1, 1, config.commit.pipeline_release_threads}, lg_size) {
 
   // Stage 0: Sequence assignment thread
   sequence_thread_ = std::thread{[this]() {
@@ -37,32 +39,35 @@ CommitPipeline::CommitPipeline(const weaseldb::Config &config)
     run_persist_stage();
   }};
 
-  // Stage 3: Connection return to server threads (2 threads)
+  // Stage 3: Connection return to server threads (configurable count)
-  release_thread_1_ = std::thread{[this]() {
+  release_threads_.reserve(config.commit.pipeline_release_threads);
-    pthread_setname_np(pthread_self(), "txn-release-1");
+  for (int i = 0; i < config.commit.pipeline_release_threads; ++i) {
-    run_release_stage<0>();
+    release_threads_.emplace_back([this, i]() {
-  }};
+      char name[16];
-
+      std::snprintf(name, sizeof(name), "txn-release-%d", i);
-  release_thread_2_ = std::thread{[this]() {
+      pthread_setname_np(pthread_self(), name);
-    pthread_setname_np(pthread_self(), "txn-release-2");
+      run_release_stage(i);
-    run_release_stage<1>();
+    });
-  }};
+  }
 }
 
 CommitPipeline::~CommitPipeline() {
-  // Send two shutdown signals for both release threads (adjacent in same batch)
+  // Send shutdown signals for all release threads (adjacent in same batch)
   {
-    auto guard = pipeline_.push(2, true);
+    int num_release_threads = static_cast<int>(release_threads_.size());
-    guard.batch[0] = ShutdownEntry{};
+    auto guard = pipeline_.push(num_release_threads, true);
-    guard.batch[1] = ShutdownEntry{};
+    for (int i = 0; i < num_release_threads; ++i) {
+      guard.batch[i] = ShutdownEntry{};
+    }
   }
 
   // Join all pipeline threads
   sequence_thread_.join();
   resolve_thread_.join();
   persist_thread_.join();
-  release_thread_1_.join();
+  for (auto &thread : release_threads_) {
-  release_thread_2_.join();
+    thread.join();
+  }
 }
 
 void CommitPipeline::submit_batch(std::span<PipelineEntry> entries) {
@@ -92,8 +97,9 @@ void CommitPipeline::run_sequence_stage() {
   BannedRequestIdSet banned_request_ids{
       ArenaStlAllocator<std::string_view>(&banned_request_arena)};
 
-  for (int shutdowns_received = 0; shutdowns_received < 2;) {
+  int expected_shutdowns = config_.commit.pipeline_release_threads;
-    auto guard = pipeline_.acquire<0, 0>();
+  for (int shutdowns_received = 0; shutdowns_received < expected_shutdowns;) {
+    auto guard = pipeline_.acquire(0, 0);
     auto &batch = guard.batch;
 
     // Stage 0: Sequence assignment
@@ -160,8 +166,9 @@ void CommitPipeline::run_sequence_stage() {
 
 // AVOID BLOCKING IN THIS STAGE!
 void CommitPipeline::run_resolve_stage() {
-  for (int shutdowns_received = 0; shutdowns_received < 2;) {
+  int expected_shutdowns = config_.commit.pipeline_release_threads;
-    auto guard = pipeline_.acquire<1, 0>(/*maxBatch*/ 1);
+  for (int shutdowns_received = 0; shutdowns_received < expected_shutdowns;) {
+    auto guard = pipeline_.acquire(1, 0, /*maxBatch*/ 1);
     auto &batch = guard.batch;
 
     // Stage 1: Precondition resolution
@@ -197,8 +204,9 @@ void CommitPipeline::run_resolve_stage() {
 }
 
 void CommitPipeline::run_persist_stage() {
-  for (int shutdowns_received = 0; shutdowns_received < 2;) {
+  int expected_shutdowns = config_.commit.pipeline_release_threads;
-    auto guard = pipeline_.acquire<2, 0>();
+  for (int shutdowns_received = 0; shutdowns_received < expected_shutdowns;) {
+    auto guard = pipeline_.acquire(2, 0);
     auto &batch = guard.batch;
 
     // Stage 2: Transaction persistence
@@ -289,9 +297,9 @@ void CommitPipeline::run_persist_stage() {
   }
 }
 
-template <int thread_index> void CommitPipeline::run_release_stage() {
+void CommitPipeline::run_release_stage(int thread_index) {
   for (int shutdowns_received = 0; shutdowns_received < 1;) {
-    auto guard = pipeline_.acquire<3, thread_index>();
+    auto guard = pipeline_.acquire(3, thread_index);
     auto &batch = guard.batch;
 
     // Stage 3: Connection release
@@ -302,7 +310,9 @@ template <int thread_index> void CommitPipeline::run_release_stage() {
 
       // Partition work: thread 0 handles even indices, thread 1 handles odd
       // indices
-      if (static_cast<int>(it.index() % 2) != thread_index) {
+      if (static_cast<int>(it.index() %
+                           config_.commit.pipeline_release_threads) !=
+          thread_index) {
         continue;
       }
 
@@ -325,7 +335,7 @@ template <int thread_index> void CommitPipeline::run_release_stage() {
 
               // 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>) {
@@ -339,7 +349,7 @@ template <int thread_index> void CommitPipeline::run_release_stage() {
 
               // 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>) {
@@ -354,8 +364,7 @@ template <int thread_index> void CommitPipeline::run_release_stage() {
               // 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.handle, health_check_entry.response_json,
-                  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;
@@ -368,8 +377,7 @@ template <int thread_index> void CommitPipeline::run_release_stage() {
               // 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.handle, get_version_entry.response_json,
-                  get_version_entry.response_json,
                   std::move(get_version_entry.request_arena));
             }
           },

src/commit_pipeline.hpp

@@ -93,27 +93,24 @@ private:
 
   // Lock-free pipeline configuration
   static constexpr int lg_size = 16; // Ring buffer size (2^16 slots)
-  static constexpr auto wait_strategy = WaitStrategy::WaitIfUpstreamIdle;
 
   // 4-stage pipeline: sequence -> resolve -> persist -> release
-  StaticThreadPipeline<PipelineEntry, wait_strategy, 1, 1, 1, 2> pipeline_;
+  ThreadPipeline<PipelineEntry> pipeline_;
 
   // Stage processing threads
   std::thread sequence_thread_;
   std::thread resolve_thread_;
   std::thread persist_thread_;
-  std::thread release_thread_1_;
+  std::vector<std::thread> release_threads_;
-  std::thread release_thread_2_;
 
   // Pipeline stage main loops
   void run_sequence_stage();
   void run_resolve_stage();
   void run_persist_stage();
-  template <int thread_index> void run_release_stage();
+  void run_release_stage(int thread_index);
 
   // Pipeline batch type alias
-  using BatchType =
+  using BatchType = ThreadPipeline<PipelineEntry>::Batch;
-      StaticThreadPipeline<PipelineEntry, wait_strategy, 1, 1, 1, 2>::Batch;
 
   // Make non-copyable and non-movable
   CommitPipeline(const CommitPipeline &) = delete;

src/config.cpp

@@ -126,6 +126,25 @@ void ConfigParser::parse_commit_config(const auto &toml_data,
                          config.request_id_retention_hours);
     parse_field(commit, "request_id_retention_versions",
                 config.request_id_retention_versions);
+
+    // Parse wait strategy
+    if (commit.contains("pipeline_wait_strategy")) {
+      std::string strategy_str =
+          toml::get<std::string>(commit.at("pipeline_wait_strategy"));
+      if (strategy_str == "WaitIfStageEmpty") {
+        config.pipeline_wait_strategy = WaitStrategy::WaitIfStageEmpty;
+      } else if (strategy_str == "WaitIfUpstreamIdle") {
+        config.pipeline_wait_strategy = WaitStrategy::WaitIfUpstreamIdle;
+      } else if (strategy_str == "Never") {
+        config.pipeline_wait_strategy = WaitStrategy::Never;
+      } else {
+        std::cerr << "Warning: Unknown pipeline_wait_strategy '" << strategy_str
+                  << "', using default (WaitIfUpstreamIdle)" << std::endl;
+      }
+    }
+
+    parse_field(commit, "pipeline_release_threads",
+                config.pipeline_release_threads);
   });
 }
 
@@ -253,6 +272,14 @@ bool ConfigParser::validate_config(const Config &config) {
     valid = false;
   }
 
+  if (config.commit.pipeline_release_threads < 1 ||
+      config.commit.pipeline_release_threads > 64) {
+    std::cerr << "Configuration error: commit.pipeline_release_threads must be "
+                 "between 1 and 64, got "
+              << config.commit.pipeline_release_threads << std::endl;
+    valid = false;
+  }
+
   // Validate subscription configuration
   if (config.subscription.max_buffer_size_bytes == 0) {
     std::cerr << "Configuration error: subscription.max_buffer_size_bytes must "

src/config.hpp

@@ -5,6 +5,8 @@
 #include <string>
 #include <vector>
 
+#include "thread_pipeline.hpp"
+
 namespace weaseldb {
 
 /**
@@ -60,6 +62,16 @@ struct CommitConfig {
   std::chrono::hours request_id_retention_hours{24};
   /// Minimum number of commit versions to retain request IDs for
   int64_t request_id_retention_versions = 100000000;
+  /// Wait strategy for the commit pipeline
+  /// - WaitIfStageEmpty: Block when individual stages are empty (default, safe
+  /// for shared CPUs)
+  /// - WaitIfUpstreamIdle: Block only when all upstream stages are idle
+  /// (requires dedicated cores)
+  /// - Never: Never block, busy-wait continuously (requires dedicated cores)
+  WaitStrategy pipeline_wait_strategy = WaitStrategy::WaitIfUpstreamIdle;
+  /// Number of threads in the release stage (final stage of commit pipeline)
+  /// Default: 1 thread for simplicity (can increase for higher throughput)
+  int pipeline_release_threads = 1;
 };
 
 /**

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 handle Protocol-specific handle for correlating this response
-   * @param data Response data parts (may be empty for deferred serialization)
+   * @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>();
+   * ProtocolHandle handle = handler.allocate_response_context(arena);
-   * ctx->sequence_id = 42;
+   * conn.send_response(handle, response_json, std::move(arena));
-   * auto response_data = format_response(arena);
-   * conn.send_response(ctx, response_data, 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
+   * active connection counter. The caller (Server) is responsible for
-   * on_connection_established() method.
+   * 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/cpu_work.cpp

@@ -55,7 +55,7 @@ asm(".text\n"
     "    b.ne .L_loop\n"    // Branch back if not zero
     ".L_end:\n"             // End
     "    ret\n"             // Return
-    ".size spend_cpu_cycles, spend_cpu_cycles\n");
+    ".size spend_cpu_cycles, .-spend_cpu_cycles\n");
 #endif
 
 #endif

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,
+      state->send_ordered_response(conn, ctx, http_response,
-                                   std::move(pending.arena),
+                                   std::move(pending.arena));
-                                   ctx->connection_close);
     }
   }
 }
@@ -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
+      // Create HttpResponseContext for this request; sequence_id doubles as the
-      auto *ctx = req.arena.allocate<HttpResponseContext>(1);
+      // 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,
+        ctx_ptr->connection_close = true;
-                                     std::move(req.arena),
+        state->send_ordered_response(*conn, ctx_ptr, http_response,
-                                     ctx->connection_close);
+                                     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(
+        g_batch_entries.emplace_back(HealthCheckEntry(
-            HealthCheckEntry(conn->get_weak_ref(), ctx, std::move(req.arena)));
+            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);
+        format_json_response(400, json_response, state->pending.arena,
-    state->send_ordered_response(conn, state->get_next_sequence_id(),
+                             ctx->http_request_id, ctx->connection_close);
-                                 http_response, std::move(state->pending.arena),
+    state->send_ordered_response(conn, ctx.get(), http_response,
-                                 true);
+                                 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,
+    conn_state->send_ordered_response(conn, ctx, http_response,
-                                      std::move(state.arena),
+                                      std::move(state.arena));
-                                      state.connection_close);
     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,
+    conn_state->send_ordered_response(conn, ctx, http_response,
-                                      std::move(state.arena),
+                                      std::move(state.arena));
-                                      state.connection_close);
     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,
+  conn_state->send_ordered_response(conn, ctx, http_response,
-                                    std::move(state.arena),
+                                    std::move(state.arena));
-                                    state.connection_close);
 }
 
 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,
+    conn_state->send_ordered_response(conn, ctx, http_response,
-                                      std::move(state.arena),
+                                      std::move(state.arena));
-                                      state.connection_close);
     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,
+    conn_state->send_ordered_response(conn, ctx, http_response,
-                                      std::move(state.arena),
+                                      std::move(state.arena));
-                                      state.connection_close);
     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,
+  conn_state->send_ordered_response(conn, ctx, http_response,
-                                    std::move(state.arena),
+                                    std::move(state.arena));
-                                    state.connection_close);
 }
 
 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,
+  conn_state->send_ordered_response(conn, ctx, http_response,
-                                    std::move(state.arena),
+                                    std::move(state.arena));
-                                    state.connection_close);
 }
 
 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,
+  conn_state->send_ordered_response(conn, ctx, http_response,
-                                    std::move(state.arena),
+                                    std::move(state.arena));
-                                    state.connection_close);
 }
 
 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,
+  conn_state->send_ordered_response(conn, ctx, result, std::move(state.arena));
-                                    std::move(state.arena),
-                                    state.connection_close);
 }
 
 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,
+  conn_state->send_ordered_response(conn, ctx, http_response,
-                                    std::move(state.arena),
+                                    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);
-  }
 }
 
 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 payload; populated when the response is ready.
- * Response data ready to send (sequence_id -> response data).
+  bool ready = false;
- * Absence from map indicates response not ready yet.
- */
-struct ResponseData {
   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
+  // All registered response contexts keyed by sequence_id.
-  std::map<int64_t, ResponseData>
+  // A context stays here until its response has been dispatched.
-      ready_responses; // sequence_id -> response data
+  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/main.cpp

@@ -246,6 +246,24 @@ int main(int argc, char *argv[]) {
   std::cout << "Request ID retention: "
             << config->commit.request_id_retention_hours.count() << " hours"
             << std::endl;
+
+  // Print pipeline configuration
+  std::string wait_strategy_str;
+  switch (config->commit.pipeline_wait_strategy) {
+  case WaitStrategy::WaitIfStageEmpty:
+    wait_strategy_str = "WaitIfStageEmpty";
+    break;
+  case WaitStrategy::WaitIfUpstreamIdle:
+    wait_strategy_str = "WaitIfUpstreamIdle";
+    break;
+  case WaitStrategy::Never:
+    wait_strategy_str = "Never";
+    break;
+  }
+  std::cout << "Pipeline wait strategy: " << wait_strategy_str << std::endl;
+  std::cout << "Pipeline release threads: "
+            << config->commit.pipeline_release_threads << std::endl;
+
   std::cout << "Subscription buffer size: "
             << config->subscription.max_buffer_size_bytes << " bytes"
             << std::endl;

src/metric.cpp

@@ -22,7 +22,11 @@
 #include <unordered_set>
 #include <vector>
 
+#if defined(__x86_64__) || defined(__amd64__) || defined(_M_X64)
 #include <immintrin.h>
+#elif defined(__aarch64__)
+#include <arm_neon.h>
+#endif
 #include <simdutf.h>
 
 #include "arena.hpp"
@@ -501,6 +505,20 @@ struct Metric {
             // Acquire lock to get consistent snapshot
             std::lock_guard lock(instance->mutex);
 
+            // BUGFIX: Flush pending observations into shared before
+            // accumulating
+            if (instance->pending.observations > 0) {
+              // Add pending to shared
+              for (size_t i = 0; i < instance->pending.bucket_counts.size();
+                   ++i) {
+                instance->shared.bucket_counts[i] +=
+                    instance->pending.bucket_counts[i];
+              }
+              instance->shared.sum += instance->pending.sum;
+              instance->shared.observations += instance->pending.observations;
+              // No need to reset pending since instance is being destroyed
+            }
+
             // Global accumulator should have been created when we made the
             // histogram
             auto &global_state = family->global_accumulated_values[labels_key];
@@ -1384,8 +1402,10 @@ void Gauge::set(double x) {
 Histogram::Histogram() = default;
 
 // Vectorized histogram bucket updates with mutex protection for consistency
-// AVX-optimized implementation for high performance
+// AVX-optimized implementation for high performance on x86-64, NEON-optimized
+// implementation on ARM64, and a scalar fallback for other architectures.
 
+#if defined(__x86_64__) || defined(__amd64__) || defined(_M_X64)
 __attribute__((target("avx"))) static void
 update_histogram_buckets_simd(std::span<const double> thresholds,
                               std::span<uint64_t> counts, double x,
@@ -1425,6 +1445,55 @@ update_histogram_buckets_simd(std::span<const double> thresholds,
     }
   }
 }
+#elif defined(__aarch64__)
+static void update_histogram_buckets_simd(std::span<const double> thresholds,
+                                          std::span<uint64_t> counts, double x,
+                                          size_t start_idx) {
+  const size_t size = thresholds.size();
+  size_t i = start_idx;
+
+  // Process 2 buckets at a time with 128-bit NEON vectors
+  const float64x2_t x_vec = vdupq_n_f64(x);
+  const uint64x2_t one = vdupq_n_u64(1);
+
+  for (; i + 2 <= size; i += 2) {
+    // Compare x <= thresholds per lane; true lanes are all ones.
+    float64x2_t thresholds_vec = vld1q_f64(&thresholds[i]);
+    uint64x2_t cmp_result = vcleq_f64(x_vec, thresholds_vec);
+
+    // Convert all-ones/all-zeros masks to per-lane 1/0 increments.
+    uint64x2_t increments = vandq_u64(cmp_result, one);
+
+    // Load current counts, add increments, and store back.
+    uint64x2_t current_counts = vld1q_u64(&counts[i]);
+    uint64x2_t updated_counts = vaddq_u64(current_counts, increments);
+    vst1q_u64(&counts[i], updated_counts);
+  }
+
+  // Handle remainder with scalar operations
+  for (; i < size; ++i) {
+    if (x <= thresholds[i]) {
+      counts[i]++;
+    }
+  }
+}
+#endif
+
+static void update_histogram_buckets(std::span<const double> thresholds,
+                                     std::span<uint64_t> counts, double x,
+                                     size_t start_idx) {
+#if defined(__x86_64__) || defined(__amd64__) || defined(_M_X64) ||            \
+    defined(__aarch64__)
+  update_histogram_buckets_simd(thresholds, counts, x, start_idx);
+#else
+  const size_t size = thresholds.size();
+  for (size_t i = start_idx; i < size; ++i) {
+    if (x <= thresholds[i]) {
+      counts[i]++;
+    }
+  }
+#endif
+}
 
 void Histogram::observe(double x) {
   assert(p->thresholds.size() == p->shared.bucket_counts.size());
@@ -1445,15 +1514,14 @@ void Histogram::observe(double x) {
     }
 
     // Update shared directly
-    update_histogram_buckets_simd(p->thresholds, p->shared.bucket_counts, x, 0);
+    update_histogram_buckets(p->thresholds, p->shared.bucket_counts, x, 0);
     p->shared.sum += x;
     p->shared.observations++;
 
     p->mutex.unlock();
   } else {
     // Slow path: accumulate in pending (lock-free)
-    update_histogram_buckets_simd(p->thresholds, p->pending.bucket_counts, x,
+    update_histogram_buckets(p->thresholds, p->pending.bucket_counts, x, 0);
-                                  0);
     p->pending.sum += x;
     p->pending.observations++;
   }

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)
+  // Protocol-agnostic handle for correlating the response
-  void *protocol_context = nullptr;
+  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)
+  // Protocol-agnostic handle for correlating the response
-  void *protocol_context = nullptr;
+  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)
+  // Protocol-agnostic handle for correlating the response
-  void *protocol_context = nullptr;
+  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)
+  explicit HealthCheckEntry(WeakRef<MessageSender> conn, ProtocolHandle handle,
-      : connection(std::move(conn)), protocol_context(ctx),
+                            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)
+  // Protocol-agnostic handle for correlating the response
-  void *protocol_context = nullptr;
+  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,
+  explicit GetVersionEntry(WeakRef<MessageSender> conn, ProtocolHandle handle,
-                           int64_t version)
+                           Arena arena, int64_t version)
-      : connection(std::move(conn)), protocol_context(ctx),
+      : 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);
+    event_loops_[i].epoll_fd_ = epoll_create1(EPOLL_CLOEXEC);
-    if (epoll_fds_[i] == -1) {
+    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) ==
+      if (epoll_ctl(event_loops_[i].epoll_fd_, EPOLL_CTL_ADD, listen_fd,
-          -1) {
+                    &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,21 +409,30 @@ 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);
 
-    if (r < 0) {
+    // Once we do EPOLLET we must drain the socket until read returns EAGAIN.
-      // Error or EOF - connection should be closed
+    for (;;) {
-      close_connection(conn);
+      int r = conn->read_bytes(buf, buf_size);
-      return;
+
+      if (r < 0) {
+        // Error or EOF - connection should be closed
+        close_connection(conn);
+        return;
+      }
+
+      if (r == 0) {
+        // No data available (EAGAIN) - read side drained
+        return;
+      }
+
+      // 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;
+      }
     }
-
-    if (r == 0) {
-      // No data available (EAGAIN) - skip read processing but continue
-      return;
-    }
-
-    // Call handler with connection reference - server retains ownership
-    handler_.on_data_arrived(std::string_view{buf, size_t(r)}, *conn);
   }
 }
 

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);
 

src/thread_pipeline.hpp

@@ -1,6 +1,5 @@
 #pragma once
 
-#include <array>
 #include <atomic>
 #include <cassert>
 #include <cstddef>
@@ -48,151 +47,174 @@ struct ThreadState {
   bool last_stage;
 };
 
-// Compile-time topology configuration for static pipelines
+// Runtime topology configuration for dynamic pipelines
 //
-// This template defines a pipeline topology at compile-time:
+// This class defines a pipeline topology at runtime:
-// - Stage and thread calculations done at compile-time
+// - Stage and thread calculations done at runtime
-// - Type-safe indexing: Stage and thread indices validated at compile-time
+// - Flexible configuration: topology can be set via constructor
-// - Fixed-size arrays with known bounds
+// - Dynamic arrays with runtime bounds checking
-// - Code specialization for each topology
+// - Single implementation works for any topology
 //
-// Example: StaticPipelineTopology<1, 4, 2> creates:
+// Example: PipelineTopology({1, 4, 2}) creates:
 //   - Stage 0: 1 thread  (index 0)
 //   - Stage 1: 4 threads (indices 1-4)
 //   - Stage 2: 2 threads (indices 5-6)
 //   - Total: 7 threads across 3 stages
-template <int... ThreadsPerStage> struct StaticPipelineTopology {
+struct PipelineTopology {
-  static_assert(sizeof...(ThreadsPerStage) > 0,
+  const std::vector<int> threads_per_stage;
-                "Must specify at least one stage");
+  const int num_stages;
-  static_assert(((ThreadsPerStage > 0) && ...),
+  const std::vector<int> stage_offsets;
-                "All stages must have at least one thread");
+  const int total_threads;
 
-  static constexpr int num_stages = sizeof...(ThreadsPerStage);
+  explicit PipelineTopology(std::vector<int> threads_per_stage_)
-  static constexpr std::array<int, num_stages> threads_per_stage = {
+      : threads_per_stage(validate_and_move(std::move(threads_per_stage_))),
-      ThreadsPerStage...};
+        num_stages(static_cast<int>(threads_per_stage.size())),
-  static constexpr int total_threads = (ThreadsPerStage + ...);
+        stage_offsets(build_stage_offsets(threads_per_stage)),
+        total_threads(build_total_threads(threads_per_stage)) {}
 
-  // Compile-time stage offset calculation
+  // Runtime stage offset calculation
-  template <int Stage> static constexpr int stage_offset() {
+  int stage_offset(int stage) const {
-    static_assert(Stage >= 0 && Stage < num_stages,
+    if (stage < 0 || stage >= num_stages) {
-                  "Stage index out of bounds");
+      std::abort(); // Stage index out of bounds
-    if constexpr (Stage == 0) {
-      return 0;
-    } else {
-      return stage_offset<Stage - 1>() + threads_per_stage[Stage - 1];
     }
+    return stage_offsets[stage];
   }
 
-  // Compile-time thread index calculation
+  // Runtime thread index calculation
-  template <int Stage, int Thread> static constexpr int thread_index() {
+  int thread_index(int stage, int thread) const {
-    static_assert(Stage >= 0 && Stage < num_stages,
+    if (stage < 0 || stage >= num_stages) {
-                  "Stage index out of bounds");
+      std::abort(); // Stage index out of bounds
-    static_assert(Thread >= 0 && Thread < threads_per_stage[Stage],
+    }
-                  "Thread index out of bounds");
+    if (thread < 0 || thread >= threads_per_stage[stage]) {
-    return stage_offset<Stage>() + Thread;
+      std::abort(); // Thread index out of bounds
+    }
+    return stage_offsets[stage] + thread;
   }
 
-  // Compile-time previous stage thread count
+  // Runtime previous stage thread count
-  template <int Stage> static constexpr int prev_stage_thread_count() {
+  int prev_stage_thread_count(int stage) const {
-    static_assert(Stage >= 0 && Stage < num_stages,
+    if (stage < 0 || stage >= num_stages) {
-                  "Stage index out of bounds");
+      std::abort(); // Stage index out of bounds
-    if constexpr (Stage == 0) {
+    }
+    if (stage == 0) {
       return 1;
     } else {
-      return threads_per_stage[Stage - 1];
+      return threads_per_stage[stage - 1];
     }
   }
+
+private:
+  static std::vector<int> validate_and_move(std::vector<int> threads) {
+    if (threads.empty()) {
+      std::abort(); // Must specify at least one stage
+    }
+    for (int count : threads) {
+      if (count <= 0) {
+        std::abort(); // All stages must have at least one thread
+      }
+    }
+    return threads;
+  }
+
+  static std::vector<int>
+  build_stage_offsets(const std::vector<int> &threads_per_stage) {
+    std::vector<int> offsets(threads_per_stage.size());
+    int offset = 0;
+    for (size_t i = 0; i < threads_per_stage.size(); ++i) {
+      offsets[i] = offset;
+      offset += threads_per_stage[i];
+    }
+    return offsets;
+  }
+
+  static int build_total_threads(const std::vector<int> &threads_per_stage) {
+    int total = 0;
+    for (int count : threads_per_stage) {
+      total += count;
+    }
+    return total;
+  }
 };
 
-// Static pipeline algorithms - compile-time specialized versions
+// Pipeline algorithms - runtime configurable versions
-namespace StaticPipelineAlgorithms {
+namespace PipelineAlgorithms {
 
-template <WaitStrategy wait_strategy, typename Topology, int Stage,
+inline uint32_t calculate_safe_len(WaitStrategy wait_strategy,
-          int ThreadInStage>
+                                   const PipelineTopology &topology, int stage,
-uint32_t calculate_safe_len(
+                                   int thread_in_stage,
-    std::array<ThreadState, Topology::total_threads> &all_threads,
+                                   std::vector<ThreadState> &all_threads,
-    std::atomic<uint32_t> &pushes, bool may_block) {
+                                   std::atomic<uint32_t> &pushes,
-  constexpr int thread_idx =
+                                   bool may_block) {
-      Topology::template thread_index<Stage, ThreadInStage>();
+  int thread_idx = topology.thread_index(stage, thread_in_stage);
   auto &thread = all_threads[thread_idx];
   uint32_t safe_len = UINT32_MAX;
 
-  constexpr int prev_stage_threads =
+  int prev_stage_threads = topology.prev_stage_thread_count(stage);
-      Topology::template prev_stage_thread_count<Stage>();
 
-  // Compile-time loop over previous stage threads
+  // Runtime loop over previous stage threads
-  [&]<std::size_t... Is>(std::index_sequence<Is...>) {
+  for (int i = 0; i < prev_stage_threads; ++i) {
-    (
+    std::atomic<uint32_t> &last_push = [&]() -> std::atomic<uint32_t> & {
-        [&] {
+      if (stage == 0) {
-          auto &last_push = [&]() -> std::atomic<uint32_t> & {
+        return pushes;
-            if constexpr (Stage == 0) {
+      } else {
-              return pushes;
+        int prev_thread_idx = topology.thread_index(stage - 1, i);
-            } else {
+        return all_threads[prev_thread_idx].pops;
-              constexpr int prev_thread_idx =
+      }
-                  Topology::template thread_index<Stage - 1, Is>();
+    }();
-              return all_threads[prev_thread_idx].pops;
+
+    if (thread.last_push_read[i] == thread.local_pops) {
+      thread.last_push_read[i] = last_push.load(std::memory_order_acquire);
+      if (thread.last_push_read[i] == thread.local_pops) {
+        if (!may_block) {
+          safe_len = 0;
+          return safe_len;
+        }
+
+        if (wait_strategy == WaitStrategy::Never) {
+          // Empty - busy wait
+        } else if (wait_strategy == WaitStrategy::WaitIfUpstreamIdle) {
+          // We're allowed to spin as long as we eventually go to 0% cpu
+          // usage on idle
+          uint32_t push;
+          bool should_wait = true;
+          for (int j = 0; j < 100000; ++j) {
+            push = pushes.load(std::memory_order_relaxed);
+            if (push != thread.local_pops) {
+              should_wait = false;
+              break;
             }
-          }();
-
-          if (thread.last_push_read[Is] == thread.local_pops) {
-            thread.last_push_read[Is] =
-                last_push.load(std::memory_order_acquire);
-            if (thread.last_push_read[Is] == thread.local_pops) {
-              if (!may_block) {
-                safe_len = 0;
-                return;
-              }
-
-              if constexpr (wait_strategy == WaitStrategy::Never) {
-                // Empty - busy wait
-              } else if constexpr (wait_strategy ==
-                                   WaitStrategy::WaitIfUpstreamIdle) {
-                // We're allowed to spin as long as we eventually go to 0% cpu
-                // usage on idle
-                uint32_t push;
-                for (int i = 0; i < 100000; ++i) {
-                  push = pushes.load(std::memory_order_relaxed);
-                  if (push != thread.local_pops) {
-                    goto dont_wait;
-                  }
 #if defined(__x86_64__) || defined(_M_X64)
-                  _mm_pause();
+            _mm_pause();
 #endif
-                }
-                pushes.wait(push, std::memory_order_relaxed);
-              dont_wait:;
-              } else {
-                static_assert(wait_strategy == WaitStrategy::WaitIfStageEmpty);
-                last_push.wait(thread.last_push_read[Is],
-                               std::memory_order_relaxed);
-              }
-
-              thread.last_push_read[Is] =
-                  last_push.load(std::memory_order_acquire);
-            }
           }
-          safe_len =
+          if (should_wait) {
-              std::min(safe_len, thread.last_push_read[Is] - thread.local_pops);
+            pushes.wait(push, std::memory_order_relaxed);
-        }(),
+          }
-        ...);
+        } else { // WaitStrategy::WaitIfStageEmpty
-  }(std::make_index_sequence<prev_stage_threads>{});
+          last_push.wait(thread.last_push_read[i], std::memory_order_relaxed);
+        }
+
+        thread.last_push_read[i] = last_push.load(std::memory_order_acquire);
+      }
+    }
+    safe_len = std::min(safe_len, thread.last_push_read[i] - thread.local_pops);
+  }
 
   return safe_len;
 }
 
-template <WaitStrategy wait_strategy, typename Topology, int Stage,
+inline void update_thread_pops(WaitStrategy wait_strategy,
-          int ThreadInStage>
+                               const PipelineTopology &topology, int stage,
-void update_thread_pops(
+                               int thread_in_stage,
-    std::array<ThreadState, Topology::total_threads> &all_threads,
+                               std::vector<ThreadState> &all_threads,
-    uint32_t local_pops) {
+                               uint32_t local_pops) {
-  constexpr int thread_idx =
+  int thread_idx = topology.thread_index(stage, thread_in_stage);
-      Topology::template thread_index<Stage, ThreadInStage>();
   auto &thread_state = all_threads[thread_idx];
 
-  if constexpr (wait_strategy == WaitStrategy::WaitIfStageEmpty) {
+  if (wait_strategy == WaitStrategy::WaitIfStageEmpty) {
     thread_state.pops.store(local_pops, std::memory_order_seq_cst);
     thread_state.pops.notify_all();
-  } else if constexpr (Stage == Topology::num_stages - 1) { // last stage
+  } else if (stage == topology.num_stages - 1) { // last stage
     thread_state.pops.store(local_pops, std::memory_order_seq_cst);
     thread_state.pops.notify_all();
   } else {
@@ -200,15 +222,13 @@ void update_thread_pops(
   }
 }
 
-template <typename Topology>
+inline int check_producer_capacity(const PipelineTopology &topology,
-int check_producer_capacity(
+                                   std::vector<ThreadState> &all_threads,
-    std::array<ThreadState, Topology::total_threads> &all_threads,
+                                   uint32_t slot, uint32_t size,
-    uint32_t slot, uint32_t size, uint32_t slot_count, bool block) {
+                                   uint32_t slot_count, bool block) {
-  constexpr int last_stage = Topology::num_stages - 1;
+  int last_stage = topology.num_stages - 1;
-  constexpr int last_stage_offset =
+  int last_stage_offset = topology.stage_offset(last_stage);
-      Topology::template stage_offset<last_stage>();
+  int last_stage_thread_count = topology.threads_per_stage[last_stage];
-  constexpr int last_stage_thread_count =
-      Topology::threads_per_stage[last_stage];
 
   for (int i = 0; i < last_stage_thread_count; ++i) {
     auto &thread = all_threads[last_stage_offset + i];
@@ -223,10 +243,10 @@ int check_producer_capacity(
   }
   return 0; // Can proceed
 }
-} // namespace StaticPipelineAlgorithms
+} // namespace PipelineAlgorithms
 
-// Static multi-stage lock-free pipeline for inter-thread communication
+// Multi-stage lock-free pipeline for inter-thread communication
-// with compile-time topology specification.
+// with runtime-configurable topology and wait strategy.
 //
 // Overview:
 // - Items flow from producers through multiple processing stages (stage 0 ->
@@ -234,25 +254,17 @@ int check_producer_capacity(
 // - Each stage can have multiple worker threads processing items in parallel
 // - Uses a shared ring buffer with atomic counters for lock-free coordination
 // - Supports batch processing for efficiency
-// - Compile-time topology specification via template parameters
+// - Runtime-configurable topology and wait strategy via constructor parameters
 //
 // Architecture:
 // - Producers: External threads that add items to the pipeline via push()
 // - Stages: Processing stages numbered 0, 1, 2, ... that consume items via
-// acquire<Stage, Thread>()
+// acquire(stage, thread)
 // - Items flow: Producers -> Stage 0 -> Stage 1 -> ... -> Final Stage
 //
-// Differences from Dynamic Version:
-// - Template parameters specify topology at compile-time (e.g., <Item,
-// WaitStrategy::Never, 1, 4, 2>)
-// - Stage and thread indices are template parameters, validated at compile-time
-// - Fixed-size arrays replace dynamic vectors
-// - Specialized algorithms for each stage/thread combination
-// - Type-safe guards prevent runtime indexing errors
-//
 // Usage Pattern:
-//   using Pipeline = StaticThreadPipeline<Item, WaitStrategy::WaitIfStageEmpty,
+//   ThreadPipeline<Item> pipeline(WaitStrategy::WaitIfStageEmpty, {1, 4, 2},
-//   1, 4, 2>; Pipeline pipeline(lgSlotCount);
+//   lgSlotCount);
 //
 //   // Producer threads (add items for stage 0 to consume):
 //   auto guard = pipeline.push(batchSize, /*block=*/true);
@@ -262,20 +274,21 @@ int check_producer_capacity(
 //   // Guard destructor publishes batch to stage 0 consumers
 //
 //   // Stage worker threads (process items and pass to next stage):
-//   auto guard = pipeline.acquire<Stage, Thread>(maxBatch, /*may_block=*/true);
+//   auto guard = pipeline.acquire(stage, thread, maxBatch, /*may_block=*/true);
 //   for (auto& item : guard.batch) {
 //     // Process item
 //   }
 //   // Guard destructor marks items as consumed and available to next stage
 //
 // Multi-Thread Stage Processing:
-// When a stage has multiple threads (e.g., 1, 1, 1, 2 = 2 threads in stage 3):
+// When a stage has multiple threads (e.g., {1, 1, 1, 2} = 2 threads in stage
+// 3):
 //
 // OVERLAPPING BATCHES - EACH THREAD SEES EVERY ENTRY:
 // - Multiple threads in the same stage get OVERLAPPING batches from the ring
 // buffer
-// - Thread 0: calls acquire<3, 0>() - gets batch from ring positions 100-110
+// - Thread 0: calls acquire(3, 0) - gets batch from ring positions 100-110
-// - Thread 1: calls acquire<3, 1>() - gets batch from ring positions 100-110
+// - Thread 1: calls acquire(3, 1) - gets batch from ring positions 100-110
 // (SAME)
 // - Both threads see the same entries and must coordinate processing
 //
@@ -319,27 +332,27 @@ int check_producer_capacity(
 // ordering
 // - Uses C++20 atomic wait/notify for efficient blocking when no work available
 // - RAII guards ensure proper cleanup even with exceptions
-template <class T, WaitStrategy wait_strategy, int... ThreadsPerStage>
+template <class T> struct ThreadPipeline {
-struct StaticThreadPipeline {
-  using Topology = StaticPipelineTopology<ThreadsPerStage...>;
-
   // Constructor
+  // wait_strategy: blocking behavior when no work is available
+  // threads_per_stage: number of threads in each stage (e.g., {1, 4, 2})
   // lgSlotCount: log2 of ring buffer size (e.g., 10 -> 1024 slots)
-  // Template parameters specify pipeline topology (e.g., <Item, Never, 1, 4,
+  // Note: Producer threads are external to the pipeline and not counted in
-  // 2>) Note: Producer threads are external to the pipeline and not counted in
+  // threads_per_stage
-  // ThreadsPerStage
+  explicit ThreadPipeline(WaitStrategy wait_strategy,
-  explicit StaticThreadPipeline(int lgSlotCount)
+                          std::vector<int> threads_per_stage, int lgSlotCount)
-      : slot_count(1 << lgSlotCount), slot_count_mask(slot_count - 1),
+      : wait_strategy_(wait_strategy), topology_(std::move(threads_per_stage)),
-        ring(slot_count) {
+        slot_count(1 << lgSlotCount), slot_count_mask(slot_count - 1),
+        ring(slot_count), all_threads(topology_.total_threads) {
     // Otherwise we can't tell the difference between full and empty.
     assert(!(slot_count_mask & 0x80000000));
     initialize_all_threads();
   }
 
-  StaticThreadPipeline(StaticThreadPipeline const &) = delete;
+  ThreadPipeline(ThreadPipeline const &) = delete;
-  StaticThreadPipeline &operator=(StaticThreadPipeline const &) = delete;
+  ThreadPipeline &operator=(ThreadPipeline const &) = delete;
-  StaticThreadPipeline(StaticThreadPipeline &&) = delete;
+  ThreadPipeline(ThreadPipeline &&) = delete;
-  StaticThreadPipeline &operator=(StaticThreadPipeline &&) = delete;
+  ThreadPipeline &operator=(ThreadPipeline &&) = delete;
 
   struct Batch {
     Batch() : ring(), begin_(), end_() {}
@@ -442,7 +455,7 @@ struct StaticThreadPipeline {
     }
 
   private:
-    friend struct StaticThreadPipeline;
+    friend struct ThreadPipeline;
     Batch(std::vector<T> *const ring, uint32_t begin_, uint32_t end_)
         : ring(ring), begin_(begin_), end_(end_) {}
     std::vector<T> *const ring;
@@ -450,29 +463,29 @@ struct StaticThreadPipeline {
     uint32_t end_;
   };
 
-  // Static thread storage - fixed size array
-  std::array<ThreadState, Topology::total_threads> all_threads;
-
 private:
+  WaitStrategy wait_strategy_;
+  PipelineTopology topology_;
+
   alignas(128) std::atomic<uint32_t> slots{0};
   alignas(128) std::atomic<uint32_t> pushes{0};
   const uint32_t slot_count;
   const uint32_t slot_count_mask;
 
   std::vector<T> ring;
+  std::vector<ThreadState> all_threads;
 
   void initialize_all_threads() {
-    [&]<std::size_t... StageIndices>(std::index_sequence<StageIndices...>) {
+    for (int stage = 0; stage < topology_.num_stages; ++stage) {
-      (init_stage_threads<StageIndices>(), ...);
+      init_stage_threads(stage);
-    }(std::make_index_sequence<Topology::num_stages>{});
+    }
   }
 
-  template <int Stage> void init_stage_threads() {
+  void init_stage_threads(int stage) {
-    constexpr int stage_offset = Topology::template stage_offset<Stage>();
+    int stage_offset = topology_.stage_offset(stage);
-    constexpr int stage_thread_count = Topology::threads_per_stage[Stage];
+    int stage_thread_count = topology_.threads_per_stage[stage];
-    constexpr int prev_stage_threads =
+    int prev_stage_threads = topology_.prev_stage_thread_count(stage);
-        Topology::template prev_stage_thread_count<Stage>();
+    bool is_last_stage = (stage == topology_.num_stages - 1);
-    constexpr bool is_last_stage = (Stage == Topology::num_stages - 1);
 
     for (int thread = 0; thread < stage_thread_count; ++thread) {
       auto &thread_state = all_threads[stage_offset + thread];
@@ -481,14 +494,15 @@ private:
     }
   }
 
-  template <int Stage, int Thread>
+  Batch acquire_helper(int stage, int thread, uint32_t maxBatch,
-  Batch acquire_helper(uint32_t maxBatch, bool mayBlock) {
+                       bool may_block) {
-    constexpr int thread_idx = Topology::template thread_index<Stage, Thread>();
+    int thread_idx = topology_.thread_index(stage, thread);
     auto &thread_state = all_threads[thread_idx];
 
     uint32_t begin = thread_state.local_pops & slot_count_mask;
-    uint32_t len = StaticPipelineAlgorithms::calculate_safe_len<
+    uint32_t len = PipelineAlgorithms::calculate_safe_len(
-        wait_strategy, Topology, Stage, Thread>(all_threads, pushes, mayBlock);
+        wait_strategy_, topology_, stage, thread, all_threads, pushes,
+        may_block);
 
     if (maxBatch != 0) {
       len = std::min(len, maxBatch);
@@ -503,13 +517,13 @@ private:
   }
 
 public:
-  template <int Stage, int Thread> struct StageGuard {
+  struct StageGuard {
     Batch batch;
 
     ~StageGuard() {
       if (!batch.empty()) {
-        StaticPipelineAlgorithms::update_thread_pops<wait_strategy, Topology,
+        PipelineAlgorithms::update_thread_pops(
-                                                     Stage, Thread>(
+            pipeline->wait_strategy_, pipeline->topology_, stage, thread,
             pipeline->all_threads, local_pops);
       }
     }
@@ -517,22 +531,28 @@ public:
     StageGuard(StageGuard const &) = delete;
     StageGuard &operator=(StageGuard const &) = delete;
     StageGuard(StageGuard &&other) noexcept
-        : batch(other.batch), local_pops(other.local_pops),
+        : batch(other.batch), local_pops(other.local_pops), stage(other.stage),
+          thread(other.thread),
           pipeline(std::exchange(other.pipeline, nullptr)) {}
     StageGuard &operator=(StageGuard &&other) noexcept {
       batch = other.batch;
       local_pops = other.local_pops;
+      stage = other.stage;
+      thread = other.thread;
       pipeline = std::exchange(other.pipeline, nullptr);
       return *this;
     }
 
   private:
-    friend struct StaticThreadPipeline;
+    friend struct ThreadPipeline;
     uint32_t local_pops;
-    StaticThreadPipeline *pipeline;
+    int stage;
+    int thread;
+    ThreadPipeline *pipeline;
 
-    StageGuard(Batch batch, uint32_t local_pops, StaticThreadPipeline *pipeline)
+    StageGuard(Batch batch, uint32_t local_pops, int stage, int thread,
-        : batch(batch), local_pops(local_pops),
+               ThreadPipeline *pipeline)
+        : batch(batch), local_pops(local_pops), stage(stage), thread(thread),
           pipeline(batch.empty() ? nullptr : pipeline) {}
   };
 
@@ -555,37 +575,30 @@ public:
     }
 
   private:
-    friend struct StaticThreadPipeline;
+    friend struct ThreadPipeline;
     ProducerGuard() : batch(), tp() {}
-    ProducerGuard(Batch batch, StaticThreadPipeline *tp, uint32_t old_slot,
+    ProducerGuard(Batch batch, ThreadPipeline *tp, uint32_t old_slot,
                   uint32_t new_slot)
         : batch(batch), tp(tp), old_slot(old_slot), new_slot(new_slot) {}
-    StaticThreadPipeline *const tp;
+    ThreadPipeline *const tp;
     uint32_t old_slot;
     uint32_t new_slot;
   };
 
   // Acquire a batch of items for processing by a consumer thread.
-  // Stage: which processing stage (0 = first consumer stage after producers) -
+  // stage: which processing stage (0 = first consumer stage after producers)
-  // compile-time parameter Thread: thread ID within the stage (0 to
+  // thread: thread ID within the stage (0 to threads_per_stage[stage]-1)
-  // ThreadsPerStage[Stage]-1) - compile-time parameter maxBatch: maximum items
+  // maxBatch: maximum items to acquire (0 = no limit)
-  // to acquire (0 = no limit) may_block: whether to block waiting for items
+  // may_block: whether to block waiting for items (false = return empty batch
-  // (false = return empty batch if none available) Returns: StageGuard<Stage,
+  // if none available) Returns: StageGuard with batch of items to process
-  // Thread> with batch of items to process and compile-time type safety
+  [[nodiscard]] StageGuard acquire(int stage, int thread, int maxBatch = 0,
-  template <int Stage, int Thread>
+                                   bool may_block = true) {
-  [[nodiscard]] StageGuard<Stage, Thread> acquire(int maxBatch = 0,
+    auto batch = acquire_helper(stage, thread, maxBatch, may_block);
-                                                  bool may_block = true) {
-    static_assert(Stage >= 0 && Stage < Topology::num_stages,
-                  "Stage index out of bounds");
-    static_assert(Thread >= 0 && Thread < Topology::threads_per_stage[Stage],
-                  "Thread index out of bounds");
 
-    auto batch = acquire_helper<Stage, Thread>(maxBatch, may_block);
+    int thread_idx = topology_.thread_index(stage, thread);
-
-    constexpr int thread_idx = Topology::template thread_index<Stage, Thread>();
     uint32_t local_pops = all_threads[thread_idx].local_pops;
 
-    return StageGuard<Stage, Thread>{std::move(batch), local_pops, this};
+    return StageGuard{std::move(batch), local_pops, stage, thread, this};
   }
 
   // Reserve slots in the ring buffer for a producer thread to fill with items.
@@ -618,9 +631,8 @@ public:
       slot = slots.load(std::memory_order_relaxed);
       begin = slot & slot_count_mask;
 
-      int capacity_result =
+      int capacity_result = PipelineAlgorithms::check_producer_capacity(
-          StaticPipelineAlgorithms::check_producer_capacity<Topology>(
+          topology_, all_threads, slot, size, slot_count, block);
-              all_threads, slot, size, slot_count, block);
       if (capacity_result == 1) {
         continue;
       }

test_benchmark_config.toml

@@ -3,7 +3,7 @@
 [server]
 # Network interfaces to listen on - both TCP for external access and Unix socket for high-performance local testing
 interfaces = [
-  { type = "tcp", address = "127.0.0.1", port = 8080 },
+  { type = "tcp", address = "0.0.0.0", port = 8123 },
   { type = "unix", path = "weaseldb.sock" }
 ]
 # Maximum request size in bytes (for 413 Content Too Large responses)

tests/test_metric.cpp

@@ -627,6 +627,113 @@ TEST_CASE("memory management") {
   }
 }
 
+TEST_CASE("histogram pending buffer thread cleanup bug") {
+  for (int iterations = 0; iterations < 1000; ++iterations) {
+    // This test demonstrates the bug where pending histogram observations
+    // are lost when a thread dies because ThreadInit destructor doesn't
+    // flush pending data into shared before accumulating into global state.
+
+    metric::reset_metrics_for_testing();
+
+    auto hist_family = metric::create_histogram(
+        "pending_bug_test", "Test histogram for pending buffer bug",
+        {1.0}); // Single bucket for simplicity
+
+    std::atomic<bool> keep_rendering{true};
+    constexpr int num_threads = 100;
+
+    std::latch ready{2};
+
+    // Background thread that calls render in a tight loop to hold global mutex
+    std::thread render_thread([&]() {
+      ready.arrive_and_wait();
+      Arena arena;
+      while (keep_rendering.load(std::memory_order_relaxed)) {
+        metric::render(arena);
+        arena.reset();
+      }
+    });
+    // Don't spawn threads until render thread is running
+    ready.arrive_and_wait();
+
+    // Spawn threads that observe once and exit
+    std::vector<std::thread> observer_threads;
+    for (int i = 0; i < num_threads; ++i) {
+      observer_threads.emplace_back([&hist_family]() {
+        auto hist = hist_family.create({{"test", "observer"}});
+        hist.observe(0.5); // Goes into first bucket (le="1.0")
+        // Thread dies here - pending observations should be lost due to bug
+      });
+    }
+
+    // Join all observer threads
+    for (auto &t : observer_threads) {
+      t.join();
+    }
+
+    // Stop render thread
+    keep_rendering.store(false, std::memory_order_relaxed);
+    render_thread.join();
+
+    // Check if the worker's observations were preserved
+    Arena arena;
+    auto output = metric::render(arena);
+
+    // First, let's debug what we actually got
+    std::ostringstream debug_output;
+    for (const auto &line : output) {
+      debug_output << line;
+    }
+    std::string full_output = debug_output.str();
+
+    // Parse the output to find the worker's bucket count for le="2.0"
+    uint64_t worker_bucket_2_count = 0;
+    bool found_worker_metric = false;
+
+    // The render output alternates between metric name and value in separate
+    // string_views
+    for (size_t i = 0; i < output.size(); ++i) {
+      const auto &line = output[i];
+      // Look for: pending_bug_test_bucket{test="observer",le="1.0"}
+      if (line.find("pending_bug_test_bucket{test=\"observer\",le=\"1.0\"}") !=
+          std::string_view::npos) {
+        found_worker_metric = true;
+        // The value should be in the next element
+        if (i + 1 < output.size()) {
+          auto value_str = output[i + 1];
+          // Remove trailing newline if present
+          while (!value_str.empty() &&
+                 (value_str.back() == '\n' || value_str.back() == '\r')) {
+            value_str.remove_suffix(1);
+          }
+          try {
+            worker_bucket_2_count = std::stoull(std::string(value_str));
+          } catch (const std::exception &e) {
+            MESSAGE("Failed to parse value: '"
+                    << value_str << "' from metric line: '" << line << "'");
+            MESSAGE("Full output:\n" << full_output);
+            throw;
+          }
+        }
+        break;
+      }
+    }
+
+    REQUIRE(found_worker_metric); // The metric should exist
+
+    // BUG: This will fail because pending observations are lost on thread death
+    // Expected: num_threads observations (each thread made 1 observation)
+    // Actual: less than num_threads (observations stuck in pending are lost
+    // when threads die)
+    CHECK_MESSAGE(
+        worker_bucket_2_count == num_threads,
+        "Expected "
+            << num_threads << " observations but got " << worker_bucket_2_count
+            << ". This indicates the pending buffer bug where observations "
+            << "stuck in pending are lost when thread dies.");
+  }
+}
+
 TEST_CASE("render output deterministic order golden test") {
   // Clean slate - reset all metrics before this test
   metric::reset_metrics_for_testing();

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};
+  Event received_data;
-  std::latch connection_closed_latch{1};
+  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();
   }
 };
 

threading_performance_report.md

@@ -2,15 +2,17 @@
 
 ## Summary
 
-WeaselDB's /ok health check endpoint achieves 1M requests/second with 740ns of configurable CPU work per request through the 4-stage commit pipeline, while maintaining 0% CPU usage when idle. The configurable CPU work serves both as a health check (validating the full pipeline) and as a benchmarking tool for measuring per-request processing capacity.
+WeaselDB's /ok health check endpoint achieves approximately 825k requests/second with 740ns of configurable CPU work per request through the 4-stage commit pipeline, while maintaining 0% CPU usage when idle. The configurable CPU work serves both as a health check (validating the full pipeline) and as a benchmarking tool for measuring per-request processing capacity.
+
+> **Note on historical numbers**: An earlier version of this report claimed 1.0M requests/second at 740ns serial CPU work. That measurement was made using a design that transferred unique ownership of connections through the pipeline. The current server-owned connection model adds per-request synchronization overhead that lowers the raw /ok throughput, but enables streaming endpoints such as `/v1/subscribe` and safer async response handling.
 
 ## Performance Metrics
 
 ### Throughput
 
-- **1.0M requests/second** /ok health check endpoint (4-stage commit pipeline)
+- **~825k requests/second** /ok health check endpoint (4-stage commit pipeline)
 - 8 I/O threads with 8 epoll instances
-- Load tester used 12 network threads
+- Load tester used 10 network threads
 - **0% CPU usage when idle** (optimized futex wake implementation)
 
 ### Threading Architecture
@@ -24,10 +26,10 @@ WeaselDB's /ok health check endpoint achieves 1M requests/second with 740ns of c
 
 **Health Check Pipeline (/ok endpoint)**:
 
-- **Throughput**: 1.0M requests/second
+- **Throughput**: ~825k requests/second (sustained over a 30-second run)
 - **Configurable CPU work**: 740ns (4000 iterations, validated with nanobench)
-- **Theoretical maximum CPU time**: 1000ns (1,000,000,000ns ÷ 1,000,000 req/s)
+- **Theoretical maximum CPU time at this throughput**: ~1212ns (1,000,000,000ns ÷ 825,000 req/s)
-- **CPU work efficiency**: 74% (740ns ÷ 1000ns)
+- **CPU work efficiency**: ~61% (740ns ÷ 1212ns)
 - **Pipeline stages**: Sequence (noop) → Resolve (CPU work) → Persist (response) → Release (cleanup)
 - **CPU usage when idle**: 0%
 
@@ -76,7 +78,7 @@ I/O Threads (8) → HttpHandler::on_batch_complete() → Commit Pipeline
 
 - Server: test_benchmark_config.toml with 8 io_threads, 8 epoll_instances
 - Configuration: `ok_resolve_iterations = 4000` (740ns CPU work)
-- Load tester: targeting /ok endpoint
+- Load tester: targeting /ok endpoint, 10 network threads, 8 connect threads, 2000 concurrent connections, 500 requests per connection
 - Benchmark validation: ./bench_cpu_work 4000
-- Build: ninja
+- Build: ninja Release
 - Command: ./weaseldb --config test_benchmark_config.toml

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: