Huge design change (#471)

Author: akiradeveloper

Rakefile

@@ -0,0 +1,4 @@
+task :doc do
+    sh "cargo doc"
+    sh "python3 -m http.server --directory target/doc 3000"
+end

book/src/client-interaction.md

@@ -13,7 +13,7 @@ You can send a R/W command to the cluster with this API.
 
 ```proto
 message WriteRequest {
-  uint32 shard_id = 1;
+  uint32 shard_index = 1;
   bytes message = 2;
   string request_id = 3;
 }
@@ -41,7 +41,7 @@ processed locally.
 
 ```proto
 message ReadRequest {
-  uint32 shard_id = 1;
+  uint32 shard_index = 1;
   bytes message = 2;
   bool read_locally = 3;
 }

book/src/cluster-management.md

@@ -13,12 +13,12 @@ From the admin client, the following API can add or remove a Raft process in the
 
 ```proto
 message AddServerRequest {
-  uint32 shard_id = 1;
+  uint32 shard_index = 1;
   string server_id = 2;
 }
 
 message RemoveServerRequest {
-  uint32 shard_id = 1;
+  uint32 shard_index = 1;
   string server_id = 2;
 }
 ```

book/src/leadership.md

@@ -38,6 +38,6 @@ any remaining Raft process to forcibly start a new election (by promoting to a c
 
 ```proto
 message TimeoutNow {
-  uint32 shard_id = 1;
+  uint32 shard_index = 1;
 }
 ```

sorock-check/README.md

@@ -6,7 +6,7 @@ https://github.com/user-attachments/assets/9aff6794-778b-48fa-bfbd-838e63b3e5c8
 
 ## Usage
 
-`sorock-check connect $URL $SHARD_ID`. (e.g. `sorock-check connect http://node5:50051 34`)
+`sorock-check connect $URL $shard_index`. (e.g. `sorock-check connect http://node5:50051 34`)
 
 Once connected to any node in a cluster,
 the program will automatically connect to all nodes in the cluster.

sorock-check/src/main.rs

@@ -30,7 +30,7 @@ mod proto {
 #[derive(Parser)]
 enum Sub {
     #[clap(about = "Start monitoring a cluster by connecting to a node.")]
-    Monitor { addr: Uri, shard_id: u32 },
+    Monitor { addr: Uri, shard_index: u32 },
     #[clap(about = "Embedded test. 0 -> Static data, 1 -> Mock servers")]
     TestMonitor { number: u8 },
 }
@@ -46,8 +46,8 @@ async fn main() -> Result<()> {
     let args = Args::parse();
 
     let model = match args.sub {
-        Sub::Monitor { addr, shard_id } => {
-            let node = real::connect_real_node(addr, shard_id);
+        Sub::Monitor { addr, shard_index } => {
+            let node = real::connect_real_node(addr, shard_index);
             model::Model::new(node).await
         }
         Sub::TestMonitor { number: 0 } => model::Model::test(),

sorock-check/src/real.rs

@@ -4,21 +4,26 @@ use futures::StreamExt;
 use std::{pin::Pin, time::Duration};
 use tonic::transport::Uri;
 
-pub fn connect_real_node(uri: Uri, shard_id: u32) -> impl model::stream::Node {
+pub fn connect_real_node(uri: Uri, shard_index: u32) -> impl model::stream::Node {
     let chan = Endpoint::from(uri).connect_lazy();
     let client = proto::raft_client::RaftClient::new(chan);
-    RealNode { client, shard_id }
+    RealNode {
+        client,
+        shard_index,
+    }
 }
 
 struct RealNode {
     client: proto::raft_client::RaftClient<Channel>,
-    shard_id: u32,
+    shard_index: u32,
 }
 
 #[async_trait::async_trait]
 impl model::stream::Node for RealNode {
     async fn watch_membership(&self) -> Pin<Box<dyn Stream<Item = proto::Membership> + Send>> {
-        let shard = proto::Shard { id: self.shard_id };
+        let shard = proto::Shard {
+            id: self.shard_index,
+        };
         let mut client = self.client.clone();
         let st = async_stream::stream! {
             loop {
@@ -42,7 +47,9 @@ impl model::stream::Node for RealNode {
         &self,
         _: Uri,
     ) -> Pin<Box<dyn Stream<Item = proto::LogMetrics> + Send>> {
-        let shard = proto::Shard { id: self.shard_id };
+        let shard = proto::Shard {
+            id: self.shard_index,
+        };
         let mut client = self.client.clone();
         let st = async_stream::stream! {
             match client.watch_log_metrics(shard).await {

sorock/Cargo.toml

@@ -28,7 +28,6 @@ rand.workspace = true
 redb.workspace = true
 serde.workspace = true
 serde_bytes.workspace = true
-shrinkwraprs = "0.3"
 spin.workspace = true
 thiserror.workspace = true
 tokio = { workspace = true, features = ["rt"] }

sorock/proto/sorock.proto

@@ -10,15 +10,15 @@ package sorock;
 // Client may send a write requests twice but they are executed only once as long as
 // they have the same `request_id`.
 message WriteRequest {
-  uint32 shard_id = 1;
+  uint32 shard_index = 1;
   bytes message = 2;
   string request_id = 3;
 }
 
 // Read-only request to the `RaftApp`.
 // This type of request is processed in optimized path.
 message ReadRequest {
-  uint32 shard_id = 1;
+  uint32 shard_index = 1;
   bytes message = 2;
   bool read_locally = 3;
 }
@@ -34,12 +34,12 @@ message Clock {
 }
 
 message KernRequest {
-  uint32 shard_id = 1;
+  uint32 shard_index = 1;
   bytes message = 2;
 }
 
 message ReplicationStreamHeader {
-  uint32 shard_id = 1;
+  uint32 shard_index = 1;
   string sender_id = 2;
   Clock prev_clock = 3;
 }
@@ -60,7 +60,7 @@ message ReplicationStreamResponse {
 }
 
 message GetSnapshotRequest {
-  uint32 shard_id = 1;
+  uint32 shard_index = 1;
   uint64 index = 2;
 }
 
@@ -69,7 +69,7 @@ message SnapshotChunk {
 }
 
 message VoteRequest {
-  uint32 shard_id = 1;
+  uint32 shard_index = 1;
   uint64 vote_term = 2;
   string candidate_id = 3;
   Clock candidate_clock = 4;
@@ -92,13 +92,13 @@ message Heartbeat {
 
 // Request to add a Raft process with `server_id` to a shard.
 message AddServerRequest {
-  uint32 shard_id = 1;
+  uint32 shard_index = 1;
   string server_id = 2;
 }
 
 // Request to remove a Raft process with `server_id` from a shard.
 message RemoveServerRequest {
-  uint32 shard_id = 1;
+  uint32 shard_index = 1;
   string server_id = 2;
 }
 
@@ -114,7 +114,7 @@ message Membership {
 // to become a leader disregarding the election timeout.
 // You can use this request to rebalance the leaders in the cluster.
 message TimeoutNow {
-  uint32 shard_id = 1;
+  uint32 shard_index = 1;
 }
 
 message LogMetrics {

sorock/src/backend/redb/ballot.rs

@@ -5,10 +5,10 @@ use redb::ReadableDatabase;
 mod value {
     use super::*;
 
-    #[derive(serde::Deserialize, serde::Serialize)]
+    #[derive(Deserialize, Serialize)]
     struct OnDiskStruct {
         term: u64,
-        voted_for: Option<sorock::NodeId>,
+        voted_for: Option<sorock::NodeAddress>,
     }
 
     pub fn ser(x: Ballot) -> Vec<u8> {
@@ -39,8 +39,8 @@ pub struct BallotStore {
 }
 
 impl BallotStore {
-    pub fn new(db: Arc<Database>, shard_id: u32) -> Result<Self> {
-        let space = format!("ballot-{shard_id}");
+    pub fn new(db: Arc<Database>, shard_index: u32) -> Result<Self> {
+        let space = format!("ballot-{shard_index}");
 
         // Insert the initial value if not exists.
         let tx = db.begin_write()?;