Don't count_down past 0

To prepare for EPOLLET

src/connection.cpp

@@ -153,7 +153,7 @@ void Connection::send_response(ProtocolHandle handle,
   }
 }
 
-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);

src/connection.hpp

@@ -279,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/server.cpp

@@ -409,7 +409,10 @@ void Server::process_connection_reads(Ref<Connection> &conn, int events) {
     auto buf_size = config_.server.read_buffer_size;
     g_read_buffer.resize(buf_size);
     char *buf = g_read_buffer.data();
-    int r = conn->readBytes(buf, buf_size);
+
+    // Once we do EPOLLET we must drain the socket until read returns EAGAIN.
+    for (;;) {
+      int r = conn->read_bytes(buf, buf_size);
 
       if (r < 0) {
         // Error or EOF - connection should be closed
@@ -418,12 +421,18 @@ void Server::process_connection_reads(Ref<Connection> &conn, int events) {
       }
 
       if (r == 0) {
-      // No data available (EAGAIN) - skip read processing but continue
+        // No data available (EAGAIN) - read side drained
         return;
       }
 
-    // Call handler with connection reference - server retains ownership
+      // Call handler with connection reference - server retains ownership.
       handler_.on_data_arrived(std::string_view{buf, size_t(r)}, *conn);
+
+      // The connection may have been closed by the handler; stop reading.
+      if (!conn) {
+        return;
+      }
+    }
   }
 }
 

tests/test_server.cpp

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