feat(distributed): transition to Multi-Group Raft with RaftManager

Author: poyrazK

include/catalog/catalog.hpp

@@ -22,7 +22,7 @@
 namespace cloudsql {
 
 namespace raft {
-class RaftNode;
+class RaftGroup;
 }
 
 // Type aliases
@@ -158,9 +158,9 @@ class Catalog {
     [[nodiscard]] static std::unique_ptr<Catalog> create();
 
     /**
-     * @brief Set Raft node for distributed operations
+     * @brief Set Raft group for distributed operations
      */
-    void set_raft_node(raft::RaftNode* raft_node) { raft_node_ = raft_node; }
+    void set_raft_group(raft::RaftGroup* raft_group) { raft_group_ = raft_group; }
 
     /**
      * @brief Load catalog from file
@@ -271,7 +271,7 @@ class Catalog {
     DatabaseInfo database_;
     oid_t next_oid_ = 1;
     uint64_t version_ = 1;
-    raft::RaftNode* raft_node_ = nullptr;
+    raft::RaftGroup* raft_group_ = nullptr;
 
     [[nodiscard]] static uint64_t get_current_time();
 };

include/distributed/raft_node.hpp include/distributed/raft_group.hppinclude/distributed/raft_node.hpp renamed to include/distributed/raft_group.hpp

@@ -1,10 +1,10 @@
 /**
- * @file raft_node.hpp
- * @brief Raft consensus node implementation
+ * @file raft_group.hpp
+ * @brief Raft consensus group implementation
  */
 
-#ifndef SQL_ENGINE_DISTRIBUTED_RAFT_NODE_HPP
-#define SQL_ENGINE_DISTRIBUTED_RAFT_NODE_HPP
+#ifndef SQL_ENGINE_DISTRIBUTED_RAFT_GROUP_HPP
+#define SQL_ENGINE_DISTRIBUTED_RAFT_GROUP_HPP
 
 #include <atomic>
 #include <chrono>
@@ -22,24 +22,24 @@
 namespace cloudsql::raft {
 
 /**
- * @brief Implementation of a Raft consensus node
+ * @brief Implementation of a Raft consensus group
  */
-class RaftNode {
+class RaftGroup {
    public:
-    RaftNode(std::string node_id, cluster::ClusterManager& cluster_manager,
-             network::RpcServer& rpc_server);
-    ~RaftNode();
+    RaftGroup(uint16_t group_id, std::string node_id, cluster::ClusterManager& cluster_manager,
+              network::RpcServer& rpc_server);
+    ~RaftGroup();
 
     // Prevent copying and moving
-    RaftNode(const RaftNode&) = delete;
-    RaftNode& operator=(const RaftNode&) = delete;
-    RaftNode(RaftNode&&) = delete;
-    RaftNode& operator=(RaftNode&&) = delete;
+    RaftGroup(const RaftGroup&) = delete;
+    RaftGroup& operator=(const RaftGroup&) = delete;
+    RaftGroup(RaftGroup&&) = delete;
+    RaftGroup& operator=(RaftGroup&&) = delete;
 
     void start();
     void stop();
 
-    // Raft RPC Handlers
+    // Raft RPC Handlers (called by RaftManager)
     void handle_request_vote(const network::RpcHeader& header, const std::vector<uint8_t>& payload,
                              int client_fd);
     void handle_append_entries(const network::RpcHeader& header,
@@ -48,6 +48,7 @@ class RaftNode {
     // Client interface
     bool replicate(const std::string& command);
     [[nodiscard]] bool is_leader() const { return state_.load() == NodeState::Leader; }
+    [[nodiscard]] uint16_t group_id() const { return group_id_; }
 
    private:
     void run_loop();
@@ -62,6 +63,7 @@ class RaftNode {
     // Helpers
     [[nodiscard]] std::chrono::milliseconds get_random_timeout() const;
 
+    uint16_t group_id_;
     std::string node_id_;
     cluster::ClusterManager& cluster_manager_;
     network::RpcServer& rpc_server_;
@@ -83,4 +85,4 @@ class RaftNode {
 
 }  // namespace cloudsql::raft
 
-#endif  // SQL_ENGINE_DISTRIBUTED_RAFT_NODE_HPP
+#endif  // SQL_ENGINE_DISTRIBUTED_RAFT_GROUP_HPP

include/distributed/raft_manager.hpp

@@ -0,0 +1,62 @@
+/**
+ * @file raft_manager.hpp
+ * @brief Manages multiple Raft consensus groups on a single node
+ */
+
+#ifndef SQL_ENGINE_DISTRIBUTED_RAFT_MANAGER_HPP
+#define SQL_ENGINE_DISTRIBUTED_RAFT_MANAGER_HPP
+
+#include <memory>
+#include <mutex>
+#include <unordered_map>
+#include <vector>
+
+#include "distributed/raft_group.hpp"
+#include "network/rpc_server.hpp"
+
+namespace cloudsql::raft {
+
+/**
+ * @brief Manager for Multi-Raft implementation
+ */
+class RaftManager {
+   public:
+    RaftManager(std::string node_id, cluster::ClusterManager& cluster_manager,
+                network::RpcServer& rpc_server);
+    ~RaftManager() = default;
+
+    // Prevent copying
+    RaftManager(const RaftManager&) = delete;
+    RaftManager& operator=(const RaftManager&) = delete;
+
+    void start();
+    void stop();
+
+    /**
+     * @brief Create or get a Raft group
+     */
+    std::shared_ptr<RaftGroup> get_or_create_group(uint16_t group_id);
+
+    /**
+     * @brief Get an existing group
+     */
+    std::shared_ptr<RaftGroup> get_group(uint16_t group_id);
+
+   private:
+    /**
+     * @brief Route incoming Raft RPCs to the correct group
+     */
+    void handle_raft_rpc(const network::RpcHeader& header, const std::vector<uint8_t>& payload,
+                         int client_fd);
+
+    std::string node_id_;
+    cluster::ClusterManager& cluster_manager_;
+    network::RpcServer& rpc_server_;
+
+    std::mutex mutex_;
+    std::unordered_map<uint16_t, std::shared_ptr<RaftGroup>> groups_;
+};
+
+}  // namespace cloudsql::raft
+
+#endif  // SQL_ENGINE_DISTRIBUTED_RAFT_MANAGER_HPP

include/network/rpc_message.hpp

@@ -157,35 +157,47 @@ class Serializer {
 };
 
 /**
- * @brief Header for all internal RPC messages (fixed 8 bytes)
+ * @brief Header for all internal RPC messages (fixed 12 bytes)
  */
 struct RpcHeader {
     static constexpr uint32_t MAGIC = 0x4353514C;  // 'CSQL'
-    static constexpr size_t HEADER_SIZE = 8;
+    static constexpr size_t HEADER_SIZE = 12;
 
     uint32_t magic = MAGIC;
     RpcType type = RpcType::Error;
     uint8_t flags = 0;
+    uint16_t group_id = 0;  // For Multi-Group Raft
+    uint16_t reserved = 0;
     uint16_t payload_len = 0;
 
     void encode(char* out) const {
         uint32_t n_magic = htonl(magic);
+        uint16_t n_group = htons(group_id);
+        uint16_t n_reserved = htons(reserved);
         uint16_t n_len = htons(payload_len);
         std::memcpy(out, &n_magic, 4);
         out[4] = static_cast<char>(type);
         out[5] = static_cast<char>(flags);
-        std::memcpy(out + 6, &n_len, 2);
+        std::memcpy(out + 6, &n_group, 2);
+        std::memcpy(out + 8, &n_reserved, 2);
+        std::memcpy(out + 10, &n_len, 2);
     }
 
     static RpcHeader decode(const char* in) {
         RpcHeader h;
         uint32_t n_magic = 0;
+        uint16_t n_group = 0;
+        uint16_t n_reserved = 0;
         uint16_t n_len = 0;
         std::memcpy(&n_magic, in, 4);
         h.magic = ntohl(n_magic);
         h.type = static_cast<RpcType>(static_cast<uint8_t>(in[4]));
         h.flags = static_cast<uint8_t>(in[5]);
-        std::memcpy(&n_len, in + 6, 2);
+        std::memcpy(&n_group, in + 6, 2);
+        h.group_id = ntohs(n_group);
+        std::memcpy(&n_reserved, in + 8, 2);
+        h.reserved = ntohs(n_reserved);
+        std::memcpy(&n_len, in + 10, 2);
         h.payload_len = ntohs(n_len);
         return h;
     }