feat(raft): integrate raft consensus into the redis write path

Route cluster-mode write commands through Raft consensus by making the
storage batch layer mode-aware, instead of per-command serialization.

- storage: implement BinlogBatch capturing CF put/delete as BinlogEntry;
  Redis gains an injectable append_log_fn (OnceLock) so create_batch returns
  BinlogBatch in cluster mode and RocksBatch otherwise; the apply path
  (on_binlog_write) always uses create_rocks_batch to avoid recursive propose;
  Storage::set_append_log_fn propagates to all instances.
- raft: add a lightweight LeaderGate trait implemented by RaftApp.
- net: thread an optional LeaderGate to the executor and reject writes on
  non-leaders with MOVED / not-leader before execution.
- server: in cluster mode, create the raft node, bridge storage->raft via an
  mpsc channel and a drain task calling client_write, inject append_log_fn
  (block_in_place around blocking_recv to avoid an async-context panic), and
  pass the LeaderGate to the network server.
- cmd: remove the now-dead per-command to_binlog/needs_raft.

Author: guozhihao

Cargo.lock

@@ -233,13 +233,6 @@ pub trait Cmd: Send + Sync {
         None
     }
 
-    fn to_binlog(&self, _client: &Client) -> Option<conf::raft_type::Binlog> {
-        None
-    }
-
-    fn needs_raft(&self) -> bool {
-        self.has_flag(CmdFlags::RAFT)
-    }
 }
 
 #[macro_export]

src/cmd/src/lib.rs

@@ -23,8 +23,6 @@ use storage::storage::Storage;
 
 use crate::{Cmd, CmdFlags, CmdMeta};
 use crate::{impl_cmd_clone_box, impl_cmd_meta};
-use conf::raft_type::{Binlog, BinlogEntry, ColumnFamilyIndex as RaftCfIndex, OperateType};
-use storage::slot_indexer::key_to_slot_id;
 
 #[derive(Clone, Default)]
 pub struct SetCmd {
@@ -74,24 +72,4 @@ impl Cmd for SetCmd {
             }
         }
     }
-
-    fn to_binlog(&self, client: &Client) -> Option<Binlog> {
-        let key = client.key();
-        let value = &client.argv()[2];
-
-        let slot_id = key_to_slot_id(&key) as u32;
-
-        let entry = BinlogEntry {
-            cf_idx: RaftCfIndex::MetaCF as u32,
-            op_type: OperateType::Put,
-            key: key.clone(),
-            value: Some(value.clone()),
-        };
-
-        Some(Binlog {
-            db_id: 0,
-            slot_idx: slot_id,
-            entries: vec![entry],
-        })
-    }
 }

src/cmd/src/set.rs

@@ -21,7 +21,7 @@ executor = { path = "../executor" }
 runtime = { path = "../common/runtime" }
 thiserror = "2.0"
 parking_lot = "0.12"
-# raft = { path = "../raft" } # removed: src/raft no longer exists
+raft = { path = "../raft" }
 
 [dev-dependencies]
 env_logger = "0.11"

src/net/Cargo.toml

@@ -29,6 +29,7 @@ use executor::CmdExecutor;
 use log::{debug, error, warn};
 use resp::RespData;
 use runtime::DualRuntimeError;
+use cmd::CmdFlags;
 
 use crate::network_execution::NetworkCmdExecution;
 
@@ -58,6 +59,22 @@ impl CmdExecutorNetworkExt for CmdExecutor {
                 return Ok(());
             }
 
+            // Cluster-mode leader gate: reject writes on non-leaders before any
+            // command-specific setup runs.
+            if let Some(gate) = exec.leader_gate.as_ref() {
+                if exec.cmd.has_flag(CmdFlags::WRITE) && !gate.is_leader() {
+                    // Simplified redirect: Kiwi returns "MOVED <addr>" (no hash slot,
+                    // unlike Redis Cluster's "MOVED <slot> <ip:port>"). Clients are
+                    // expected to reconnect to the returned leader address directly.
+                    let reply = match gate.leader_resp_addr() {
+                        Some(addr) => format!("MOVED {addr}"),
+                        None => "ERR not leader".to_string(),
+                    };
+                    exec.client.set_reply(RespData::Error(reply.into()));
+                    return Ok(());
+                }
+            }
+
             // Execute do_initial if needed
             if !exec.cmd.do_initial(&exec.client) {
                 debug!("Command initial check failed for: {}", cmd_name);

src/net/src/executor_ext.rs

@@ -129,8 +129,9 @@ pub async fn process_connection_with_storage_client(
     storage_client: Arc<StorageClient>,
     cmd_table: Arc<CmdTable>,
     executor: Arc<CmdExecutor>,
+    leader_gate: Option<std::sync::Arc<dyn raft::leader_gate::LeaderGate>>,
 ) -> std::io::Result<()> {
     // Delegate to the network-aware connection handler
-    crate::network_handle::process_network_connection(client, storage_client, cmd_table, executor)
+    crate::network_handle::process_network_connection(client, storage_client, cmd_table, executor, leader_gate)
         .await
 }

src/net/src/handle.rs

@@ -57,9 +57,10 @@ impl ServerFactory {
         addr: Option<String>,
         runtime_manager: &RuntimeManager,
         requirepass: Option<String>,
+        leader_gate: Option<Arc<dyn raft::leader_gate::LeaderGate>>,
     ) -> Option<Box<dyn ServerTrait>> {
         match protocol.to_lowercase().as_str() {
-            "tcp" => match Self::create_network_server(addr, runtime_manager, requirepass) {
+            "tcp" => match Self::create_network_server(addr, runtime_manager, requirepass, leader_gate) {
                 Ok(server) => Some(Box::new(server) as Box<dyn ServerTrait>),
                 Err(e) => {
                     log::error!("Failed to create NetworkServer: {}", e);
@@ -113,6 +114,7 @@ impl ServerFactory {
         addr: Option<String>,
         runtime_manager: &RuntimeManager,
         requirepass: Option<String>,
+        leader_gate: Option<Arc<dyn raft::leader_gate::LeaderGate>>,
     ) -> Result<NetworkServer, Box<dyn std::error::Error>> {
         // Get the storage client from RuntimeManager
         let runtime_storage_client = runtime_manager.storage_client().map_err(|e| {
@@ -132,6 +134,6 @@ impl ServerFactory {
         })));
         let executor = Arc::new(CmdExecutorBuilder::new().build());
 
-        NetworkServer::new(addr, storage_client, cmd_table, executor, requirepass)
+        NetworkServer::new(addr, storage_client, cmd_table, executor, requirepass, leader_gate)
     }
 }

src/net/src/lib.rs

@@ -40,15 +40,7 @@ pub struct NetworkCmdExecution {
     pub client: Arc<Client>,
     /// The storage client for network-to-storage communication
     pub storage_client: Arc<StorageClient>,
-}
-
-impl NetworkCmdExecution {
-    /// Create a new NetworkCmdExecution
-    pub fn new(cmd: Arc<dyn Cmd>, client: Arc<Client>, storage_client: Arc<StorageClient>) -> Self {
-        Self {
-            cmd,
-            client,
-            storage_client,
-        }
-    }
+    /// Optional leadership gate; `None` in standalone mode. When `Some` and the
+    /// command is a write on a non-leader, the executor replies `-MOVED` (Task 7).
+    pub leader_gate: Option<std::sync::Arc<dyn raft::leader_gate::LeaderGate>>,
 }

src/net/src/network_execution.rs

@@ -49,6 +49,7 @@ pub async fn process_network_connection(
     storage_client: Arc<StorageClient>,
     cmd_table: Arc<CmdTable>,
     executor: Arc<CmdExecutor>,
+    leader_gate: Option<std::sync::Arc<dyn raft::leader_gate::LeaderGate>>,
 ) -> std::io::Result<()> {
     let mut buf = vec![0; 4096]; // Increased buffer size for better performance
     let mut resp_parser = resp::RespParse::new(resp::RespVersion::RESP2);
@@ -71,6 +72,7 @@ pub async fn process_network_connection(
                                     storage_client.clone(),
                                     cmd_table.clone(),
                                     executor.clone(),
+                                    leader_gate.clone(),
                                 ).await;
                             }
                             return Ok(());
@@ -97,6 +99,7 @@ pub async fn process_network_connection(
                                                 storage_client.clone(),
                                                 cmd_table.clone(),
                                                 executor.clone(),
+                                                leader_gate.clone(),
                                             ).await;
                                             pending_commands.clear();
                                         }
@@ -124,6 +127,7 @@ pub async fn process_network_connection(
                                 storage_client.clone(),
                                 cmd_table.clone(),
                                 executor.clone(),
+                                leader_gate.clone(),
                             ).await;
                             pending_commands.clear();
                         }
@@ -151,6 +155,7 @@ async fn handle_network_command(
     storage_client: Arc<StorageClient>,
     cmd_table: Arc<CmdTable>,
     executor: Arc<CmdExecutor>,
+    leader_gate: Option<std::sync::Arc<dyn raft::leader_gate::LeaderGate>>,
 ) {
     // Convert the command name from &[u8] to a lowercase String for lookup
     let cmd_name = String::from_utf8_lossy(&client.cmd_name()).to_lowercase();
@@ -164,6 +169,7 @@ async fn handle_network_command(
             cmd: cmd.clone(),
             client: client.clone(),
             storage_client: storage_client.clone(),
+            leader_gate: leader_gate.clone(),
         };
 
         // Execute the command using the network-aware executor
@@ -340,6 +346,7 @@ async fn process_command_batch(
     storage_client: Arc<StorageClient>,
     cmd_table: Arc<CmdTable>,
     executor: Arc<CmdExecutor>,
+    leader_gate: Option<std::sync::Arc<dyn raft::leader_gate::LeaderGate>>,
 ) {
     debug!("Processing command batch of {} commands", commands.len());
 
@@ -371,6 +378,7 @@ async fn process_command_batch(
             storage_client.clone(),
             cmd_table.clone(),
             executor.clone(),
+            leader_gate.clone(),
         )
         .await;
 

src/net/src/network_handle.rs

@@ -69,6 +69,8 @@ pub struct NetworkServer {
     connection_pool: Arc<ConnectionPool<NetworkResources>>,
     /// Authentication password; when set, clients must AUTH before running commands
     requirepass: Option<String>,
+    /// Optional leadership gate for cluster-mode write rejection
+    leader_gate: Option<std::sync::Arc<dyn raft::leader_gate::LeaderGate>>,
 }
 
 impl NetworkServer {
@@ -81,6 +83,7 @@ impl NetworkServer {
         cmd_table: Arc<CmdTable>,
         executor: Arc<CmdExecutor>,
         requirepass: Option<String>,
+        leader_gate: Option<std::sync::Arc<dyn raft::leader_gate::LeaderGate>>,
     ) -> Result<Self, Box<dyn Error>> {
         let pool_config = default_network_pool_config();
 
@@ -91,6 +94,7 @@ impl NetworkServer {
             executor: executor.clone(),
             connection_pool: Arc::new(ConnectionPool::new(pool_config)),
             requirepass,
+            leader_gate,
         })
     }
 
@@ -102,6 +106,7 @@ impl NetworkServer {
         executor: Arc<CmdExecutor>,
         pool_config: PoolConfig,
         requirepass: Option<String>,
+        leader_gate: Option<std::sync::Arc<dyn raft::leader_gate::LeaderGate>>,
     ) -> Result<Self, Box<dyn Error>> {
         Ok(Self {
             addr: addr.unwrap_or("127.0.0.1:7379".to_string()),
@@ -110,6 +115,7 @@ impl NetworkServer {
             executor: executor.clone(),
             connection_pool: Arc::new(ConnectionPool::new(pool_config)),
             requirepass,
+            leader_gate,
         })
     }
 
@@ -175,6 +181,7 @@ impl ServerTrait for NetworkServer {
             let cmd_table = self.cmd_table.clone();
             let executor = self.executor.clone();
             let requirepass = self.requirepass.clone();
+            let leader_gate = self.leader_gate.clone();
 
             tokio::spawn(async move {
                 // Get or create resources from the pool
@@ -214,6 +221,7 @@ impl ServerTrait for NetworkServer {
                     pooled_resources.inner().storage_client.clone(),
                     pooled_resources.inner().cmd_table.clone(),
                     pooled_resources.inner().executor.clone(),
+                    leader_gate,
                 )
                 .await;
 
@@ -258,6 +266,7 @@ mod tests {
             cmd_table,
             executor,
             None,
+            None,
         );
 
         assert!(server.is_ok());
@@ -291,6 +300,7 @@ mod tests {
             executor,
             pool_config,
             None,
+            None,
         );
 
         assert!(server.is_ok());
@@ -311,7 +321,7 @@ mod tests {
         let cmd_table = Arc::new(create_command_table(Arc::new(|| None)));
         let executor = Arc::new(CmdExecutorBuilder::new().build());
 
-        let server = NetworkServer::new(None, storage_client, cmd_table, executor, None);
+        let server = NetworkServer::new(None, storage_client, cmd_table, executor, None, None);
 
         assert!(server.is_ok());
         let server = server.unwrap();