feat(cluster): implement Raft consensus with tracing and HTTP support

- Add Raft log and snapshot mechanisms for distributed consensus
- Integrate hyper HTTP server and client libraries for network communication
- Enhance tracing capabilities with structured logging and spans
- Add dependency tracking for new consensus-related crates
- Implement snapshot storage with serialization and persistence
- Add remote repository synchronization via Raft commands
- Include comprehensive tracing instrumentation across services

actor/mod.rs

@@ -1,15 +1,21 @@
 pub mod handler;
 pub mod message;
+pub mod raft_log;
 pub mod server;
+pub mod snapshot;
 pub mod sync;
 
 pub use handler::find_primary_in_cluster;
 pub use handler::{
-    GitNodeActor, GitNodeArgs, RepoEntry, broadcast_ref_update, broadcast_role_changed,
-    get_category_members, get_cluster_nodes, list_all_groups, route_group_for, start_node_actor,
+    broadcast_append_entries, broadcast_ref_update, broadcast_role_changed,
+    get_category_members, get_cluster_nodes, is_leader_lease_valid, list_all_groups,
+    renew_leader_lease, route_group_for, start_node_actor, GitNodeActor, GitNodeArgs, RepoEntry,
 };
 pub use message::{
-    ElectionRequest, ElectionResult, GitNodeMessage, NodeHealth, ROLE_PRIMARY, ROLE_REPLICA,
-    RefUpdateEvent, RepoActorMessage, RoleChangedEvent, RouteDecision,
+    AppendEntriesRequest, AppendEntriesResponse, ElectionRequest, ElectionResult,
+    GitNodeMessage, NodeHealth, ReadIndexRequest, ReadIndexResponse, RefUpdateEvent,
+    RepoActorMessage, RoleChangedEvent, RouteDecision, SerializedRaftEntry,
+    ROLE_PRIMARY, ROLE_REPLICA, RAFT_MSG_VERSION,
 };
+pub use raft_log::{Command as RaftCommand, LogEntry as RaftLogEntry, RaftLog};
 pub use server::init_actor_cluster;

actor/server.rs

@@ -2,14 +2,16 @@ use crate::actor::handler::start_node_actor;
 use crate::actor::message::GitNodeMessage;
 use crate::server::GitksService;
 use ractor::ActorRef;
+use std::path::PathBuf;
 
 pub async fn init_actor_cluster(
     service: GitksService,
     storage_name: String,
     grpc_addr: String,
+    data_dir: PathBuf,
 ) -> Result<(ActorRef<GitNodeMessage>, tokio::task::JoinHandle<()>), ractor::SpawnErr> {
     tracing::info!(storage_name = %storage_name, grpc_addr = %grpc_addr, "initializing actor cluster");
-    let result = start_node_actor(service, storage_name.clone(), grpc_addr).await?;
+    let result = start_node_actor(service, storage_name.clone(), grpc_addr, data_dir).await?;
     tracing::info!(storage_name = %storage_name, "actor cluster ready");
     Ok(result)
 }

actor/snapshot.rs

@@ -0,0 +1,202 @@
+//! Raft log snapshot mechanism for log compaction.
+//!
+//! When the Raft log grows beyond the size threshold, a snapshot is created
+//! that captures the current state of all repositories. Old log entries before
+//! the snapshot are then discarded.
+//!
+//! Snapshot format:
+//! - `raft-snapshot.dat`: Contains the serialized state at a given log index
+//!
+//! The snapshot includes:
+//! - All repository entries (path, role, last_commit)
+//! - The log index at which the snapshot was taken
+//! - The term at which the snapshot was taken
+
+use std::collections::HashMap;
+use std::io::{BufReader, BufWriter, Read, Write};
+use std::path::{Path, PathBuf};
+
+use crate::actor::handler::RepoEntry;
+use crate::error::{GitError, GitResult};
+
+/// Snapshot metadata and data.
+#[derive(Debug, Clone)]
+pub struct RaftSnapshot {
+    /// The log index at which this snapshot was taken.
+    pub last_included_index: u64,
+    /// The term at which this snapshot was taken.
+    pub last_included_term: u64,
+    /// All repository entries at the time of the snapshot.
+    pub repos: HashMap<String, RepoEntry>,
+}
+
+impl RaftSnapshot {
+    /// Create a new snapshot from the current state.
+    pub fn new(
+        last_included_index: u64,
+        last_included_term: u64,
+        repos: HashMap<String, RepoEntry>,
+    ) -> Self {
+        Self {
+            last_included_index,
+            last_included_term,
+            repos,
+        }
+    }
+
+    /// Serialize the snapshot to bytes.
+    pub fn encode(&self) -> Vec<u8> {
+        let mut buf = Vec::new();
+        // Header
+        buf.extend(self.last_included_index.to_be_bytes());
+        buf.extend(self.last_included_term.to_be_bytes());
+        // Repository count
+        buf.extend((self.repos.len() as u32).to_be_bytes());
+        // Each repository entry
+        for (path, entry) in &self.repos {
+            encode_string(&mut buf, path);
+            encode_string(&mut buf, &entry.role);
+            encode_string(&mut buf, &entry.last_commit);
+            buf.push(if entry.read_only { 1 } else { 0 });
+        }
+        buf
+    }
+
+    /// Deserialize a snapshot from bytes.
+    pub fn decode(data: &[u8]) -> Option<Self> {
+        if data.len() < 20 {
+            return None;
+        }
+        let mut offset = 0;
+        let last_included_index = read_u64(data, &mut offset)?;
+        let last_included_term = read_u64(data, &mut offset)?;
+        let repo_count = read_u32(data, &mut offset)? as usize;
+
+        let mut repos = HashMap::with_capacity(repo_count);
+        for _ in 0..repo_count {
+            let path = read_string(data, &mut offset)?;
+            let role = read_string(data, &mut offset)?;
+            let last_commit = read_string(data, &mut offset)?;
+            let read_only = data.get(offset).copied().unwrap_or(0) == 1;
+            offset += 1;
+            repos.insert(path, RepoEntry {
+                role,
+                last_commit,
+                read_only,
+            });
+        }
+
+        Some(Self {
+            last_included_index,
+            last_included_term,
+            repos,
+        })
+    }
+}
+
+/// Snapshot storage manager.
+pub struct SnapshotStorage {
+    snapshot_path: PathBuf,
+}
+
+impl SnapshotStorage {
+    pub fn new(data_dir: &Path) -> Self {
+        Self {
+            snapshot_path: data_dir.join("raft-snapshot.dat"),
+        }
+    }
+
+    /// Save a snapshot to disk.
+    pub fn save(&self, snapshot: &RaftSnapshot) -> GitResult<()> {
+        let data = snapshot.encode();
+        let file = std::fs::File::create(&self.snapshot_path).map_err(GitError::Io)?;
+        let mut writer = BufWriter::new(file);
+        writer.write_all(&data).map_err(GitError::Io)?;
+        writer.flush().map_err(GitError::Io)?;
+
+        tracing::info!(
+            index = snapshot.last_included_index,
+            term = snapshot.last_included_term,
+            repos = snapshot.repos.len(),
+            "raft snapshot saved"
+        );
+        Ok(())
+    }
+
+    /// Load a snapshot from disk.
+    pub fn load(&self) -> GitResult<Option<RaftSnapshot>> {
+        if !self.snapshot_path.exists() {
+            return Ok(None);
+        }
+
+        let file = std::fs::File::open(&self.snapshot_path).map_err(GitError::Io)?;
+        let mut reader = BufReader::new(file);
+        let mut data = Vec::new();
+        reader.read_to_end(&mut data).map_err(GitError::Io)?;
+
+        match RaftSnapshot::decode(&data) {
+            Some(snapshot) => {
+                tracing::info!(
+                    index = snapshot.last_included_index,
+                    term = snapshot.last_included_term,
+                    repos = snapshot.repos.len(),
+                    "raft snapshot loaded"
+                );
+                Ok(Some(snapshot))
+            }
+            None => {
+                tracing::warn!("failed to decode raft snapshot, ignoring");
+                Ok(None)
+            }
+        }
+    }
+
+    /// Check if a snapshot exists.
+    pub fn exists(&self) -> bool {
+        self.snapshot_path.exists()
+    }
+
+    /// Delete the snapshot file.
+    pub fn delete(&self) -> GitResult<()> {
+        if self.snapshot_path.exists() {
+            std::fs::remove_file(&self.snapshot_path).map_err(GitError::Io)?;
+        }
+        Ok(())
+    }
+}
+
+// ── Helper functions ─────────────────────────────────────────
+
+fn encode_string(buf: &mut Vec<u8>, s: &str) {
+    let bytes = s.as_bytes();
+    buf.extend((bytes.len() as u32).to_be_bytes());
+    buf.extend(bytes);
+}
+
+fn read_u32(data: &[u8], offset: &mut usize) -> Option<u32> {
+    if *offset + 4 > data.len() {
+        return None;
+    }
+    let val = u32::from_be_bytes(data[*offset..*offset + 4].try_into().ok()?);
+    *offset += 4;
+    Some(val)
+}
+
+fn read_u64(data: &[u8], offset: &mut usize) -> Option<u64> {
+    if *offset + 8 > data.len() {
+        return None;
+    }
+    let val = u64::from_be_bytes(data[*offset..*offset + 8].try_into().ok()?);
+    *offset += 8;
+    Some(val)
+}
+
+fn read_string(data: &[u8], offset: &mut usize) -> Option<String> {
+    let len = read_u32(data, offset)? as usize;
+    if *offset + len > data.len() {
+        return None;
+    }
+    let s = String::from_utf8_lossy(&data[*offset..*offset + len]).into_owned();
+    *offset += len;
+    Some(s)
+}

actor/sync.rs

@@ -355,3 +355,53 @@ fn update_local_ref(repo_path: &Path, ref_name: &str, new_oid: &str) {
         Err(e) => tracing::error!(ref_name = %ref_name, error = %e, "update-ref spawn failed"),
     }
 }
+
+/// Apply a committed Raft command to the local git repository.
+/// This is called on followers when they receive committed entries from the leader.
+pub fn apply_raft_command_to_repo(
+    repo_prefix: &Path,
+    command: &crate::actor::raft_log::Command,
+) {
+    match command {
+        crate::actor::raft_log::Command::RefUpdate {
+            relative_path,
+            ref_name,
+            old_oid: _,
+            new_oid,
+        } => {
+            let repo_path = repo_prefix.join(relative_path);
+            tracing::info!(
+                relative_path = %relative_path,
+                ref_name = %ref_name,
+                new_oid = %new_oid,
+                "applying RefUpdate from Raft log to local repo"
+            );
+            update_local_ref(&repo_path, ref_name, new_oid);
+        }
+        crate::actor::raft_log::Command::RegisterRepo {
+            relative_path,
+            storage_name: _,
+        } => {
+            tracing::info!(
+                relative_path = %relative_path,
+                "RegisterRepo from Raft log (no git action needed)"
+            );
+        }
+        crate::actor::raft_log::Command::RemoveRepo { relative_path } => {
+            tracing::info!(
+                relative_path = %relative_path,
+                "RemoveRepo from Raft log (no git action needed)"
+            );
+        }
+        crate::actor::raft_log::Command::SetPrimary {
+            storage_name,
+            relative_paths,
+        } => {
+            tracing::info!(
+                storage_name = %storage_name,
+                paths = relative_paths.len(),
+                "SetPrimary from Raft log (no git action needed)"
+            );
+        }
+    }
+}