Fix

Author: Istrate Andrei-Eduard

README.md

@@ -2,7 +2,7 @@
 
 DynaRust is a high-performance, distributed key–JSON‑value store built in Rust. It's designed for massive concurrency, reliable consistency, and seamless scalability 🔄.
 
-It combines **lock-free concurrent storage**, **binary internal replication**, **causal consistency via Vector Clocks**, and a **SWIM-inspired gossip protocol** to deliver a fault‑tolerant, horizontally scalable datastore.
+It combines **lock-free concurrent storage**, **robust JSON internal replication**, **causal consistency via Vector Clocks**, and a **SWIM-inspired gossip protocol** to deliver a fault‑tolerant, horizontally scalable datastore.
 
 With its advanced real‑time update capabilities, DynaRust pushes live changes with latencies below 5 ms 🚀. Optimized for modern hardware, a single node can comfortably sustain peak traffic of up to **10,000+ live connections** 🔥—and you can increase capacity linearly by adding more nodes to your cluster.
 
@@ -11,7 +11,7 @@ With its advanced real‑time update capabilities, DynaRust pushes live changes
 | **Metric**             | **Value**                                     |
 |------------------------|-----------------------------------------------|
 | Storage Engine         | DashMap (Granular Locking)                    |
-| Internal Protocol      | Bincode (Binary Serialization)                |
+| Internal Protocol      | JSON (Reliable Serialization)                 |
 | Consistency Model      | Eventual Consistency w/ Vector Clocks         |
 | Replication Strategy   | Consistent Hashing Ring                       |
 | Reliability            | Read Repair + Exponential Backoff Retries     |

src/main.rs

@@ -94,7 +94,20 @@ async fn main() -> std::io::Result<()> {
 
     // (Optionally) Ensure the default table exists in memory.
     state.store.entry("default".to_string()).or_default();
-    start_snapshot_task(state.clone());
+    start_snapshot_task(&state);
+
+    // Initialize cluster data with dynamic membership.
+    let mut initial_nodes = HashMap::new();
+    initial_nodes.insert(
+        current_node.clone(),
+        NodeInfo {
+            status: NodeStatus::Active,
+            last_heartbeat: current_timestamp(),
+        },
+    );
+    let cluster_data = web::Data::new(ClusterData {
+        nodes: Arc::new(RwLock::new(initial_nodes)),
+    });
 
     // If a join node is provided, join its cluster and merge its global state.
     if args.len() >= 3 {
@@ -116,6 +129,10 @@ async fn main() -> std::io::Result<()> {
                 if response.status().is_success() {
                     if let Ok(nodes) = response.json::<HashMap<String, NodeInfo>>().await {
                         println!("Joined cluster: {:?}", nodes);
+                        let mut nodes_guard = cluster_data.nodes.write().await;
+                        for (node, info) in nodes {
+                            nodes_guard.insert(node, info);
+                        }
                     }
                 } else {
                     println!("Failed to join cluster (status = {})", response.status());
@@ -167,23 +184,10 @@ async fn main() -> std::io::Result<()> {
     // Spawn the periodic cold save task.
     tokio::spawn(cold_save(state.clone(), 30));
 
-    // Initialize cluster data with dynamic membership.
-    let mut initial_nodes = HashMap::new();
-    initial_nodes.insert(
-        current_node.clone(),
-        NodeInfo {
-            status: NodeStatus::Active,
-            last_heartbeat: current_timestamp(),
-        },
-    );
-    let cluster_data = web::Data::new(ClusterData {
-        nodes: Arc::new(RwLock::new(initial_nodes)),
-    });
-
     // Spawn the membership synchronization (heartbeat) task.
     let cluster_clone = cluster_data.clone();
     let current_clone = current_node.clone();
-    tokio::spawn(membership_sync(cluster_clone, current_clone, 60));
+    tokio::spawn(membership_sync(cluster_clone, current_clone, 5));
     let subscription_manager = web::Data::new(SubscriptionManager::new());
 
     // Initialize metrics collector and middleware

src/network/broadcaster.rs

@@ -217,8 +217,12 @@ async fn send_membership_update(
     url: &str,
     membership: &HashMap<String, NodeInfo>,
 ) -> Result<(), String> {
+    let cluster_token = env::var("CLUSTER_SECRET")
+        .unwrap_or_else(|_| "default_secret".to_string());
+
     match client.post(url)
         .header("Content-Type", "application/octet-stream")
+        .header("X-Cluster-Token", &cluster_token)
         .body(bincode::serialize(membership).unwrap())
         .send().await {
         Ok(resp) => {
@@ -256,7 +260,7 @@ pub async fn update_membership(
     // 1. Authentication Check
     // In production, load this once at startup rather than on every request
     let expected_token = env::var("CLUSTER_SECRET")
-        .unwrap_or_else(|_| "default_insecure_secret".to_string());
+        .unwrap_or_else(|_| "default_secret".to_string());
 
     let is_authenticated = req.headers()
         .get("X-Cluster-Token")
@@ -277,9 +281,9 @@ pub async fn update_membership(
     let mut updated = false;
 
     for (node, info) in incoming.into_iter() {
-        // 2. Input Validation (Basic check to ensure it looks like a URL)
-        if !node.starts_with("http://") && !node.starts_with("https://") {
-            continue; // Ignore malformed node addresses
+        // 2. Input Validation (ensure it's not empty and has no spaces)
+        if node.is_empty() || node.contains(' ') {
+            continue;
         }
 
         // For nodes not in our membership, add them safely

src/storage/admin.rs

@@ -68,7 +68,10 @@ fn validate_admin_token(req: &HttpRequest) -> bool {
                     &Validation::default(),
                 ) {
                     Ok(token_data) => token_data.claims.admin,
-                    Err(_) => false,
+                    Err(e) => {
+                        eprintln!("Admin token validation failed: {}", e);
+                        false
+                    },
                 };
             }
         }
@@ -243,8 +246,7 @@ async fn replicate_admin_change(
                 .put(&url)
                 .header("X-Internal-Request", "true")
                 .header("SECRET", &secret_clone)
-                .header("Content-Type", "application/octet-stream")
-                .body(bincode::serialize(&payload).unwrap())
+                .json(&payload)
                 .send()
                 .await;
         });

src/storage/engine.rs

@@ -85,15 +85,6 @@ impl VersionedValue {
         }
         strictly_newer
     }
-
-    pub fn merge_clock(&mut self, other: &HashMap<String, u64>) {
-        for (node, other_v) in other {
-            let self_v = self.vector_clock.entry(node.clone()).or_insert(0);
-            if *other_v > *self_v {
-                *self_v = *other_v;
-            }
-        }
-    }
 }
 
 /// The local in-memory database state.
@@ -384,15 +375,12 @@ pub async fn get_value(
             tokio::spawn(async move {
                 let _permit = permit;
                 let url = format!("http://{}/internal/{}/key/{}", target, table_repair, key_repair);
-                if let Ok(payload) = bincode::serialize(&latest_repair) {
-                    let _ = cli.put(&url)
-                        .header("X-Internal-Request", "true")
-                        .header("SECRET", &secret_repair)
-                        .header("Content-Type", "application/octet-stream")
-                        .body(payload)
-                        .send()
-                        .await;
-                }
+                let _ = cli.put(&url)
+                    .header("X-Internal-Request", "true")
+                    .header("SECRET", &secret_repair)
+                    .json(&latest_repair)
+                    .send()
+                    .await;
             });
         }
     }
@@ -484,8 +472,7 @@ pub async fn patch_value(
                     .put(&url)
                     .header("X-Internal-Request", "true")
                     .header("SECRET", &secret_clone)
-                    .header("Content-Type", "application/octet-stream")
-                    .body(bincode::serialize(&payload).unwrap())
+                    .json(&payload)
                     .send()
                     .await
                 {
@@ -583,8 +570,7 @@ pub async fn put_value(
                     .put(&url)
                     .header("X-Internal-Request", "true")
                     .header("SECRET", &secret_clone)
-                    .header("Content-Type", "application/octet-stream")
-                    .body(bincode::serialize(&payload).unwrap())
+                    .json(&payload)
                     .send()
                     .await
                 {
@@ -610,22 +596,20 @@ pub async fn put_value_internal(
     path: web::Path<(String, String)>,
     state: web::Data<AppState>,
     sub_manager: web::Data<SubscriptionManager>,
-    body: web::Bytes,
+    body: web::Json<VersionedValue>,
 ) -> impl Responder {
     let (table_name, key_val) = path.into_inner();
 
     // 1️⃣ Cluster‐secret check
-    let cluster_secret = env::var("CLUSTER_SECRET").unwrap_or_default();
-    match req.headers().get("SECRET") {
-        Some(h) if h.to_str().unwrap_or("") == cluster_secret => {}
-        _ => return HttpResponse::Unauthorized().finish(),
+    let cluster_secret = env::var("CLUSTER_SECRET").unwrap_or_else(|_| "default_secret".to_string());
+    let provided_secret = req.headers().get("SECRET").and_then(|h| h.to_str().ok()).unwrap_or("");
+    
+    if provided_secret != cluster_secret {
+        return HttpResponse::Unauthorized().finish();
     }
 
     // 2️⃣ Merge the incoming VersionedValue by version number
-    let incoming: VersionedValue = match bincode::deserialize(&body) {
-        Ok(v) => v,
-        Err(_) => return HttpResponse::BadRequest().finish(),
-    };
+    let incoming = body.into_inner();
     let table = state.store.entry(table_name.clone()).or_default();
 
     let cached = table

src/storage/snapshoting.rs

@@ -46,13 +46,13 @@ async fn clean_old_snapshots(dir: &PathBuf, max_snapshots: usize)
 /// 2. Encrypt it with `encrypt()`
 /// 3. Write to `./snapshots/snapshot_<ts>.json.enc`
 /// 4. Clean up old snapshots
-pub fn start_snapshot_task(state: Data<AppState>) {
+pub fn start_snapshot_task(states: &Data<AppState>) {
     let snap_limit = env::var("SNAP_LIMIT")
         .ok()
         .and_then(|s| s.parse().ok())
         .unwrap_or(100);
     let snapshot_dir = PathBuf::from("./snapshots");
-
+    let state = states.clone();
     tokio::spawn(async move {
         if let Err(e) = fs::create_dir_all(&snapshot_dir).await {
             eprintln!(