feat(db): add db cluster logic and tests

ahnlich/Cargo.toml

@@ -54,6 +54,7 @@ prost = "0.13"
 pulp = "0.21.4"
 smallvec = "1.13"
 openraft = { version = "0.9", features = ["serde", "storage-v2"] }
+bitcode = { version = "0.6", features = ["serde"] }
 
 [profile.release]
 lto = true

ahnlich/db/Cargo.toml

@@ -42,6 +42,10 @@ pulp.workspace = true
 papaya.workspace = true
 ahnlich_types = { path = "../types", version = "*"  }
 tonic.workspace = true
+prost.workspace = true
+openraft.workspace = true
+ahnlich-replication = { path = "../replication", version = "*" }
+bitcode.workspace = true
 
 [dev-dependencies]
 futures.workspace = true
@@ -54,7 +58,6 @@ dhat = "0.3"
 rcgen = "0.13"
 tempfile = "3"
 ahnlich_client_rs = { path = "../client", version = "*" }
-bitcode = { version = "0.6", features = ["serde"] }
 chrono = { version = "0.4", default-features = false, features = ["std", "clock"] }
 
 [[bench]]

ahnlich/db/src/cli/server.rs

@@ -1,3 +1,4 @@
+use ahnlich_replication::config::RaftStorageEngine;
 use clap::{Args, Parser, Subcommand};
 use utils::cli::CommandLineConfig;
 
@@ -18,6 +19,20 @@ pub enum Commands {
 pub struct ServerConfig {
     #[arg(long, default_value_t = 1369)]
     pub port: u16,
+    #[arg(long)]
+    pub cluster_addr: Option<std::net::SocketAddr>,
+    #[arg(long, default_value_t = false, conflicts_with = "cluster_join")]
+    pub cluster_bootstrap: bool,
+    #[arg(long, conflicts_with = "cluster_bootstrap")]
+    pub cluster_join: Option<std::net::SocketAddr>,
+    #[arg(long, value_enum, default_value_t = RaftStorageEngine::RocksDb)]
+    pub cluster_storage: RaftStorageEngine,
+    #[arg(long, requires_if("rocksdb", "cluster_storage"))]
+    pub cluster_data_dir: Option<std::path::PathBuf>,
+    #[arg(long, default_value_t = 1000)]
+    pub cluster_snapshot_logs: u64,
+    #[arg(long, default_value_t = 300_000)]
+    pub cluster_snapshot_interval: u64,
     #[clap(flatten)]
     pub common: CommandLineConfig,
 }
@@ -26,6 +41,13 @@ impl Default for ServerConfig {
     fn default() -> Self {
         Self {
             port: 1369,
+            cluster_addr: None,
+            cluster_bootstrap: false,
+            cluster_join: None,
+            cluster_storage: RaftStorageEngine::RocksDb,
+            cluster_data_dir: None,
+            cluster_snapshot_logs: 1000,
+            cluster_snapshot_interval: 300_000,
             common: CommandLineConfig::default(),
         }
     }
@@ -38,6 +60,39 @@ impl ServerConfig {
         self
     }
 
+    pub fn is_clustered(&self) -> bool {
+        self.cluster_addr.is_some()
+    }
+
+    pub fn cluster_addr(mut self, addr: std::net::SocketAddr) -> Self {
+        self.cluster_addr = Some(addr);
+        self
+    }
+
+    pub fn cluster_bootstrap(mut self, addr: std::net::SocketAddr) -> Self {
+        self.cluster_addr = Some(addr);
+        self.cluster_bootstrap = true;
+        self.cluster_join = None;
+        self
+    }
+
+    pub fn cluster_join(mut self, addr: std::net::SocketAddr, join: std::net::SocketAddr) -> Self {
+        self.cluster_addr = Some(addr);
+        self.cluster_bootstrap = false;
+        self.cluster_join = Some(join);
+        self
+    }
+
+    pub fn cluster_data_dir(mut self, dir: std::path::PathBuf) -> Self {
+        self.cluster_data_dir = Some(dir);
+        self
+    }
+
+    pub fn cluster_storage(mut self, storage: RaftStorageEngine) -> Self {
+        self.cluster_storage = storage;
+        self
+    }
+
     pub fn persist_location(mut self, location: std::path::PathBuf) -> Self {
         self.common.persist_location = Some(location);
         self

ahnlich/db/src/lib.rs

@@ -3,6 +3,7 @@ mod algorithm;
 pub mod cli;
 pub mod engine;
 pub mod errors;
+pub mod replication;
 pub mod server;
 
 #[cfg(test)]