Merge remote-tracking branch 'upstream/main' into fix-snapshot-logindex-sync

# Conflicts:
#	src/storage/src/lib.rs
#	src/storage/src/redis.rs
#	src/storage/src/storage.rs

Author: lupengfan1

Cargo.lock

@@ -0,0 +1,138 @@
+# Cluster Quickstart & Write-Path Verification
+
+This guide shows how to run Kiwi in standalone and Raft cluster modes, and how to
+manually verify the Raft write path (leader writes replicate to followers).
+
+> Automated multi-node integration tests are tracked separately. The steps below
+> are the manual procedure used to validate the write-path integration.
+
+## Prerequisites
+
+- Rust toolchain (stable) and `protoc` (see the project README / `CLAUDE.md`).
+- `redis-cli` — to drive the RESP port (`brew install redis` / `apt install redis-tools`).
+- `grpcurl` — to call the Raft admin gRPC API for cluster init (`brew install grpcurl`).
+  The Raft gRPC server has reflection enabled, so no `.proto` files are needed.
+
+Build the server binary:
+
+```bash
+cargo build --release --bin kiwi   # binary at target/release/kiwi
+```
+
+## Standalone mode (default)
+
+With no Raft configuration, Kiwi runs as a single standalone node. Writes go
+directly to local RocksDB (no consensus).
+
+```bash
+cargo run --release --bin kiwi          # listens on 127.0.0.1:7379 by default
+redis-cli -p 7379 set k1 hello          # OK
+redis-cli -p 7379 get k1                # "hello"
+```
+
+## Cluster mode
+
+A node runs in cluster mode when its config file contains the Raft keys below.
+Config files use the Redis-style `key value` format (one per line).
+
+### 1. Write a config per node
+
+`node1.conf`:
+
+```conf
+port 7401
+binding 127.0.0.1
+db-path /tmp/kiwi/n1/db
+raft-node-id 1
+raft-addr 127.0.0.1:8501
+raft-resp-addr 127.0.0.1:7401
+raft-data-dir /tmp/kiwi/n1/raft
+raft-use-memory-log-store true
+```
+
+Create `node2.conf` and `node3.conf` likewise, changing `port`/`raft-addr`/
+`raft-resp-addr`/`db-path`/`raft-data-dir` and `raft-node-id` (2 and 3). Keep
+`raft-resp-addr` equal to `<binding>:<port>` for each node — it is the address
+returned to clients on redirect.
+
+Raft config keys:
+
+| Key | Meaning |
+|-----|---------|
+| `raft-node-id` | Unique node id (`u64`) |
+| `raft-addr` | gRPC address for Raft internal traffic |
+| `raft-resp-addr` | RESP address advertised to clients (used in `MOVED`) |
+| `raft-data-dir` | Directory for Raft logs / snapshots |
+| `raft-use-memory-log-store` | `true` = in-memory log (testing); `false` = RocksDB-backed |
+
+### 2. Start the nodes
+
+```bash
+RUST_LOG=info ./target/release/kiwi --config node1.conf &
+RUST_LOG=info ./target/release/kiwi --config node2.conf &
+RUST_LOG=info ./target/release/kiwi --config node3.conf &
+```
+
+Each node now listens on its RESP port and its Raft gRPC port. Until the cluster
+is initialized there is no leader, so **writes are rejected** with `ERR not leader`.
+
+### 3. Initialize the cluster
+
+Call `Initialize` once, on any node's Raft gRPC port, listing all members:
+
+```bash
+grpcurl -plaintext -d '{"nodes":[
+  {"node_id":1,"raft_addr":"127.0.0.1:8501","resp_addr":"127.0.0.1:7401"},
+  {"node_id":2,"raft_addr":"127.0.0.1:8502","resp_addr":"127.0.0.1:7402"},
+  {"node_id":3,"raft_addr":"127.0.0.1:8503","resp_addr":"127.0.0.1:7403"}
+]}' 127.0.0.1:8501 kiwi.raft.v1.RaftAdminService/Initialize
+# => { "response": { "success": true, "message": "OK" }, "leaderId": "1" }
+```
+
+A leader is elected within the election-timeout window (a second or two).
+
+## Verify the write path
+
+### Find the leader
+
+A write returns `OK` on the leader and `MOVED <leader-resp-addr>` on followers:
+
+```bash
+redis-cli -p 7401 set probe v   # OK            -> node1 is leader
+redis-cli -p 7402 set probe v   # MOVED 127.0.0.1:7401
+redis-cli -p 7403 set probe v   # MOVED 127.0.0.1:7401
+```
+
+### Leader write → follower read (replication)
+
+```bash
+# Write on the leader
+redis-cli -p 7401 set repltest hello_from_leader   # OK
+redis-cli -p 7401 incr counter                     # 1
+
+# Read from a follower (eventually consistent — allow replication to arrive)
+sleep 1
+redis-cli -p 7402 get repltest    # "hello_from_leader"
+redis-cli -p 7403 get repltest    # "hello_from_leader"
+redis-cli -p 7402 get counter     # "1"
+```
+
+What this exercises end to end: command → leader gate (passes on leader) →
+`BinlogBatch` captures the encoded CF mutations → Raft `client_write` (consensus)
+→ each node's state machine applies via `on_binlog_write` to local RocksDB →
+follower local reads observe the replicated value.
+
+## Notes & current limitations
+
+- **Reads are eventually consistent.** Followers serve local reads; a value is
+  visible only after the entry has replicated and applied. Linearizable reads
+  (a read barrier) are not implemented yet.
+- **`MOVED` is simplified.** Kiwi returns `MOVED <addr>` (no hash slot), unlike
+  Redis Cluster's `MOVED <slot> <ip:port>`. Off-the-shelf cluster-aware clients
+  will not auto-follow it; reconnect to the returned address directly.
+- **Snapshot install.** After a Raft snapshot install, a node's storage is
+  swapped and currently does not re-arm the Raft write hook — writes on that node
+  would bypass consensus until restarted. This is a known follow-up to fix before
+  production cluster use.
+- **Writes only on the leader.** Followers reject writes with `MOVED` / `ERR not
+  leader`. Reads are accepted on any node.

docs/cluster-quickstart.md

@@ -232,14 +232,6 @@ pub trait Cmd: Send + Sync {
     fn get_sub_cmd(&self, _cmd_name: &str) -> Option<&dyn Cmd> {
         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

@@ -25,6 +25,7 @@ use std::pin::Pin;
 use std::sync::Arc;
 use std::time::Duration;
 
+use cmd::CmdFlags;
 use executor::CmdExecutor;
 use log::{debug, error, warn};
 use resp::RespData;
@@ -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,15 @@ 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)
-        .await
+    crate::network_handle::process_network_connection(
+        client,
+        storage_client,
+        cmd_table,
+        executor,
+        leader_gate,
+    )
+    .await
 }

src/net/src/handle.rs

@@ -57,15 +57,18 @@ 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) {
-                Ok(server) => Some(Box::new(server) as Box<dyn ServerTrait>),
-                Err(e) => {
-                    log::error!("Failed to create NetworkServer: {}", e);
-                    None
+            "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);
+                        None
+                    }
                 }
-            },
+            }
             #[cfg(unix)]
             "unix" => match unix::UnixServer::new(addr, None) {
                 Ok(server) => Some(Box::new(server) as Box<dyn ServerTrait>),
@@ -113,6 +116,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 +136,13 @@ 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;