Improve statesync shutdown and fix protocol corruption

This commit improves the statesync shutdown mechanism and fixes a
protocol corruption issue and busy recv loop.

Key changes:

1. Modified statesync_server_recv() to guarantee complete reads:
   - Now always returns exactly the requested number of bytes or -1
   - Accumulates data internally until the full message is received
   - Prevents protocol corruption from partial reads + reconnection
   - Uses poll() instead of busy-waiting, eliminating CPU spinning

2. Removed retry loop in monad_statesync_server_run_once():
   - Simplified to two simple recv() calls (type + request)
   - No longer needed since recv() guarantees complete reads

3. Fixed shutdown handling during reconnect:
   - connect() now polls shutdown_eventfd during retry loop
   - Prevents infinite hang when connection keeps failing
   - recv() reconnects and returns error on connection loss

This code was generated using Claude Sonnet 4.5

Author: maxkozlovsky

category/statesync/CMakeLists.txt

@@ -34,7 +34,8 @@ add_library(
   "statesync_server.h"
   "statesync_server_context.cpp"
   "statesync_server_context.hpp"
-  "statesync_server_context.hpp"
+  "statesync_thread.cpp"
+  "statesync_thread.hpp"
   "statesync_version.cpp"
   "statesync_version.h")
 

category/statesync/statesync_server.cpp

@@ -517,15 +517,10 @@ void monad_statesync_server_run_once(struct monad_statesync_server *const sync)
         return;
     }
     MONAD_ASSERT(buf[0] == SYNC_TYPE_REQUEST);
-    unsigned char *ptr = buf;
-    uint64_t n = sizeof(monad_sync_request);
-    while (n != 0) {
-        auto const res = sync->statesync_server_recv(sync->net, ptr, n);
-        if (res == -1) {
-            continue;
-        }
-        ptr += res;
-        n -= static_cast<size_t>(res);
+    if (sync->statesync_server_recv(
+            sync->net, buf, sizeof(monad_sync_request)) !=
+        static_cast<ssize_t>(sizeof(monad_sync_request))) {
+        return;
     }
     auto const &rq = unaligned_load<monad_sync_request>(buf);
     monad_statesync_server_handle_request(sync, rq);

category/statesync/statesync_server.h

@@ -17,6 +17,8 @@
 
 #include <category/statesync/statesync_messages.h>
 
+#include <sys/types.h>
+
 struct monad_statesync_server;
 struct monad_statesync_server_context;
 struct monad_statesync_server_network;

category/statesync/statesync_server_network.hpp

@@ -21,7 +21,10 @@
 
 #include <quill/Quill.h>
 
+#include <array>
 #include <chrono>
+#include <poll.h>
+#include <sys/eventfd.h>
 #include <sys/socket.h>
 #include <sys/un.h>
 #include <thread>
@@ -30,6 +33,7 @@
 struct monad_statesync_server_network
 {
     int fd;
+    int shutdown_eventfd;
     monad::byte_string obuf;
     std::string path;
 
@@ -42,19 +46,50 @@ struct monad_statesync_server_network
         memset(&addr, 0, sizeof(addr));
         addr.sun_family = AF_UNIX;
         strncpy(addr.sun_path, path.c_str(), sizeof(addr.sun_path) - 1);
+
         while (::connect(fd, (sockaddr *)&addr, sizeof(addr)) != 0) {
-            std::this_thread::sleep_for(std::chrono::milliseconds(1));
+            // Check if shutdown was requested before sleeping
+            std::array<pollfd, 1> pfds{};
+            pfds[0].fd = shutdown_eventfd;
+            pfds[0].events = POLLIN;
+
+            // Poll with short timeout to check for shutdown signal
+            int const poll_ret = poll(pfds.data(), pfds.size(), 1);
+
+            if (poll_ret > 0 && (pfds[0].revents & POLLIN)) {
+                // Shutdown requested, abort connection attempt
+                LOG_WARNING("Connect aborted due to shutdown request");
+                return;
+            }
+        }
+    }
+
+    void signal_shutdown()
+    {
+        uint64_t const val = 1;
+        ssize_t const res = write(shutdown_eventfd, &val, sizeof(val));
+        if (res != sizeof(val)) {
+            LOG_WARNING(
+                "Failed to signal shutdown eventfd: {}", strerror(errno));
         }
     }
 
     monad_statesync_server_network(char const *const path)
         : path{path}
     {
+        shutdown_eventfd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
+        MONAD_ASSERT_PRINTF(
+            shutdown_eventfd >= 0,
+            "failed to create eventfd: %s",
+            strerror(errno));
         connect();
     }
 
     ~monad_statesync_server_network()
     {
+        if (shutdown_eventfd >= 0) {
+            close(shutdown_eventfd);
+        }
         if (fd >= 0) {
             close(fd);
         }
@@ -92,33 +127,76 @@ namespace
     }
 }
 
-ssize_t statesync_server_recv(
+inline ssize_t statesync_server_recv(
     monad_statesync_server_network *const net, unsigned char *buf, size_t n)
 {
-    while (true) {
-        ssize_t ret = recv(net->fd, buf, n, MSG_DONTWAIT);
-        if (ret == 0 ||
-            (ret < 0 && (errno == ECONNRESET || errno == ENOTCONN))) {
-            LOG_WARNING("connection closed, reconnecting");
+    size_t total_received = 0;
+
+    while (total_received < n) {
+        std::array<pollfd, 2> pfds{};
+        pfds[0].fd = net->fd;
+        pfds[0].events = POLLIN;
+
+        pfds[1].fd = net->shutdown_eventfd;
+        pfds[1].events = POLLIN;
+
+        int const poll_ret = poll(pfds.data(), pfds.size(), -1);
+
+        if (poll_ret < 0) {
+            if (errno == EINTR) {
+                continue;
+            }
+            LOG_ERROR("poll error: {}", strerror(errno));
+            return -1;
+        }
+
+        if (pfds[1].revents & POLLIN) {
+            return -1;
+        }
+
+        if (pfds[0].revents & (POLLERR | POLLHUP | POLLNVAL)) {
+            LOG_WARNING("socket error event, reconnecting");
             if (close(net->fd) < 0) {
                 LOG_ERROR("failed to close socket: {}", strerror(errno));
             }
             net->fd = -1;
             net->connect();
-        }
-        else if (
-            ret < 0 &&
-            (errno != EAGAIN && errno != EWOULDBLOCK && errno != EINTR)) {
-            LOG_ERROR("recv error: {}", strerror(errno));
             return -1;
         }
-        else {
-            return ret;
+
+        if (pfds[0].revents & POLLIN) {
+            ssize_t const ret = recv(
+                net->fd,
+                buf + total_received,
+                n - total_received,
+                MSG_DONTWAIT);
+
+            if (ret == 0 ||
+                (ret < 0 && (errno == ECONNRESET || errno == ENOTCONN))) {
+                LOG_WARNING("connection closed, reconnecting");
+                if (close(net->fd) < 0) {
+                    LOG_ERROR("failed to close socket: {}", strerror(errno));
+                }
+                net->fd = -1;
+                net->connect();
+                return -1;
+            }
+            else if (
+                ret < 0 &&
+                (errno != EAGAIN && errno != EWOULDBLOCK && errno != EINTR)) {
+                LOG_ERROR("recv error: {}", strerror(errno));
+                return -1;
+            }
+            else if (ret > 0) {
+                total_received += static_cast<size_t>(ret);
+            }
         }
     }
+
+    return static_cast<ssize_t>(n);
 }
 
-void statesync_server_send_upsert(
+inline void statesync_server_send_upsert(
     monad_statesync_server_network *const net, monad_sync_type const type,
     unsigned char const *const v1, uint64_t const size1,
     unsigned char const *const v2, uint64_t const size2)
@@ -152,15 +230,15 @@ void statesync_server_send_upsert(
     LOG_DEBUG(
         "sending upsert type={} {} ns={}",
         std::to_underlying(type),
-        fmt::format(
+        fmtquill::format(
             "v1=0x{:02x} v2=0x{:02x}",
-            fmt::join(std::as_bytes(std::span(v1, size1)), ""),
-            fmt::join(std::as_bytes(std::span(v2, size2)), "")),
+            fmtquill::join(std::as_bytes(std::span(v1, size1)), ""),
+            fmtquill::join(std::as_bytes(std::span(v2, size2)), "")),
         std::chrono::duration_cast<std::chrono::nanoseconds>(
             std::chrono::steady_clock::now() - start));
 }
 
-void statesync_server_send_done(
+inline void statesync_server_send_done(
     monad_statesync_server_network *const net, monad_sync_done const msg)
 {
     [[maybe_unused]] auto const start = std::chrono::steady_clock::now();

category/statesync/statesync_thread.cpp

@@ -0,0 +1,57 @@
+// Copyright (C) 2025 Category Labs, Inc.
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+#include <category/statesync/statesync_thread.hpp>
+
+#include <category/mpt/ondisk_db_config.hpp>
+
+#include <pthread.h>
+
+MONAD_NAMESPACE_BEGIN
+
+StateSyncServer::StateSyncServer(StateSyncServerConfig const &config)
+    : ctx{std::make_unique<monad_statesync_server_context>(*config.triedb)}
+    , server{monad_statesync_server_create(
+          ctx.get(), config.network, &monad::statesync_server_recv,
+          &monad::statesync_server_send_upsert,
+          &monad::statesync_server_send_done)}
+    , thread{[this, config](std::stop_token const token) {
+        pthread_setname_np(pthread_self(), "statesync");
+
+        mpt::AsyncIOContext io_ctx{mpt::ReadOnlyOnDiskDbConfig{
+            .sq_thread_cpu = config.ro_sq_thread_cpu,
+            .dbname_paths = config.dbname_paths}};
+        mpt::Db ro{io_ctx};
+        ctx->ro = &ro;
+
+        std::stop_callback stop_cb(
+            token, [config]() { config.network->signal_shutdown(); });
+
+        while (!token.stop_requested()) {
+            monad_statesync_server_run_once(server.get());
+        }
+
+        ctx->ro = nullptr;
+    }}
+{
+}
+
+std::unique_ptr<StateSyncServer>
+make_statesync_server(StateSyncServerConfig const &config)
+{
+    return std::make_unique<StateSyncServer>(config);
+}
+
+MONAD_NAMESPACE_END

category/statesync/statesync_thread.hpp

@@ -0,0 +1,67 @@
+// Copyright (C) 2025 Category Labs, Inc.
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+#pragma once
+
+#include <category/statesync/statesync_server.h>
+#include <category/statesync/statesync_server_context.hpp>
+#include <category/statesync/statesync_server_network.hpp>
+
+#include <cstdint>
+#include <filesystem>
+#include <memory>
+#include <optional>
+#include <thread>
+#include <vector>
+
+MONAD_NAMESPACE_BEGIN
+
+struct StateSyncServerConfig
+{
+    monad::TrieDb *triedb;
+    monad_statesync_server_network *network;
+    std::optional<int> ro_sq_thread_cpu;
+    std::vector<std::filesystem::path> dbname_paths;
+};
+
+struct monad_statesync_server_deleter
+{
+    void operator()(monad_statesync_server *server) const
+    {
+        if (server != nullptr) {
+            monad_statesync_server_destroy(server);
+        }
+    }
+};
+
+struct StateSyncServer
+{
+    std::unique_ptr<monad_statesync_server_context> ctx;
+    std::unique_ptr<monad_statesync_server, monad_statesync_server_deleter>
+        server;
+    std::jthread thread;
+
+    explicit StateSyncServer(StateSyncServerConfig const &config);
+
+    StateSyncServer(StateSyncServer const &) = delete;
+    StateSyncServer &operator=(StateSyncServer const &) = delete;
+    StateSyncServer(StateSyncServer &&) = delete;
+    StateSyncServer &operator=(StateSyncServer &&) = delete;
+};
+
+std::unique_ptr<StateSyncServer>
+make_statesync_server(StateSyncServerConfig const &config);
+
+MONAD_NAMESPACE_END