syncing auth table issues and https node syncing issue fixed

Author: test

src/network/broadcaster.rs

@@ -1,9 +1,31 @@
 use std::collections::HashMap;
+use std::env;
 use std::time::Duration;
 use reqwest;
 use actix_web::{web, HttpResponse, Responder};
 use serde_json::json;
-use crate::storage::engine::{current_timestamp, ClusterData, NodeInfo, NodeStatus};
+use crate::storage::engine::{
+    current_timestamp, ClusterData, NodeInfo, NodeStatus,
+};
+
+/// Read DYNA_MODE and return either "https" or "http".
+fn get_scheme() -> &'static str {
+    match env::var("DYNA_MODE").map(|m| m.to_lowercase()).as_deref() {
+        Ok("https") => "https",
+        _ => "http",
+    }
+}
+
+/// Strip any existing scheme/trailing slash from `node` and build a full URL.
+fn build_url(node: &str, endpoint: &str) -> String {
+    let scheme = get_scheme();
+    let host = node
+        .trim_start_matches("http://")
+        .trim_start_matches("https://")
+        .trim_end_matches('/');
+    let ep = endpoint.trim_start_matches('/');
+    format!("{}://{}/{}", scheme, host, ep)
+}
 
 //
 // MEMBERSHIP BACKGROUND TASK
@@ -21,7 +43,13 @@ pub async fn membership_sync(
         tokio::time::sleep(Duration::from_secs(interval_sec)).await;
 
         // Run health checks and update node statuses
-        check_node_health(&cluster_data, &client, &current_addr, interval_sec).await;
+        check_node_health(
+            &cluster_data,
+            &client,
+            &current_addr,
+            interval_sec,
+        )
+            .await;
 
         // Distribute membership information to other nodes
         gossip_membership(&cluster_data, &client, &current_addr).await;
@@ -36,27 +64,27 @@ async fn check_node_health(
     interval_sec: u64,
 ) {
     let mut nodes_guard = cluster_data.nodes.write().await;
-    let timeout_threshold = interval_sec as u128 * 2000; // 2x interval in milliseconds
+    // 2x interval in milliseconds
+    let timeout_threshold = (interval_sec as u128) * 2000;
 
     for (node, info) in nodes_guard.iter_mut() {
         // Skip self-check
         if node == current_addr {
             continue;
         }
-        let url = if node.starts_with("http://") || node.starts_with("https://") {
-            // Remove any trailing slash for consistency, then append the endpoint.
-            format!("{}/heartbeat", node.trim_end_matches('/'))
-        } else {
-            // If no scheme is present, use http as the default.
-            format!("http://{}/heartbeat", node)
-        };
-
-        match client.get(&url).timeout(Duration::from_secs(2)).send().await {
+        let url = build_url(node, "heartbeat");
+
+        match client
+            .get(&url)
+            .timeout(Duration::from_secs(2))
+            .send()
+            .await
+        {
             Ok(resp) if resp.status().is_success() => {
                 // Successful heartbeat - mark node as active
                 info.last_heartbeat = current_timestamp();
                 info.status = NodeStatus::Active;
-            },
+            }
             _ => {
                 // Failed heartbeat - update status based on current state
                 update_node_status(info, timeout_threshold);
@@ -73,20 +101,20 @@ fn update_node_status(info: &mut NodeInfo, timeout_threshold: u128) {
         NodeStatus::Active => {
             // First failure - mark as suspect
             info.status = NodeStatus::Suspect;
-        },
+        }
         NodeStatus::Suspect => {
             // Check if it's been down too long
             let now = current_timestamp();
             if now - info.last_heartbeat > timeout_threshold {
                 info.status = NodeStatus::Down;
             }
-        },
+        }
         _ => {} // For other states, do nothing
     }
 }
 
 /// Send membership information to all other nodes
-async fn gossip_membership(
+pub async fn gossip_membership(
     cluster_data: &web::Data<ClusterData>,
     client: &reqwest::Client,
     current_addr: &str,
@@ -103,15 +131,21 @@ async fn gossip_membership(
             continue;
         }
 
-        let gossip_url = format!("http://{}/update_membership", node);
+        let gossip_url = build_url(node, "update_membership");
         let membership_clone = nodes_snapshot.clone();
         let client_clone = client.clone();
         let node_clone = node.clone();
 
         // Send update asynchronously
         tokio::spawn(async move {
-            if let Err(e) = send_membership_update(&client_clone, &gossip_url, &membership_clone).await {
-                eprintln!("Error gossiping membership to {}: {}", node_clone, e);
+            if let Err(e) =
+                send_membership_update(&client_clone, &gossip_url, &membership_clone)
+                    .await
+            {
+                eprintln!(
+                    "Error gossiping membership to {}: {}",
+                    node_clone, e
+                );
             }
         });
     }
@@ -130,7 +164,7 @@ async fn send_membership_update(
             } else {
                 Err(format!("Failed with status: {}", resp.status()))
             }
-        },
+        }
         Err(e) => Err(e.to_string()),
     }
 }
@@ -141,7 +175,10 @@ async fn send_membership_update(
 
 /// Respond to heartbeat requests from other nodes
 pub async fn heartbeat() -> impl Responder {
-    HttpResponse::Ok().json(json!({"status": "ok", "timestamp": current_timestamp()}))
+    HttpResponse::Ok().json(json!({
+        "status": "ok",
+        "timestamp": current_timestamp(),
+    }))
 }
 
 /// Process membership updates received from other nodes
@@ -185,8 +222,11 @@ pub async fn get_membership(
     // Create a response with additional metadata
     let response = json!({
         "nodes": nodes.clone(),
-        "active_count": nodes.values().filter(|n| n.status == NodeStatus::Active).count(),
-        "timestamp": current_timestamp()
+        "active_count": nodes
+            .values()
+            .filter(|n| n.status == NodeStatus::Active)
+            .count(),
+        "timestamp": current_timestamp(),
     });
 
     HttpResponse::Ok().json(response)

src/security/authentication.rs

@@ -6,7 +6,6 @@ use sha2::{Digest, Sha256};
 use hex;
 use chrono::{Utc, Duration};
 use std::collections::HashMap;
-use std::env;
 use jsonwebtoken::{encode, Header, EncodingKey};
 
 use crate::storage::engine::{get_active_nodes, get_replication_nodes, AppState, ClusterData, VersionedValue};

src/storage/engine.rs

@@ -11,7 +11,8 @@ use tokio::sync::RwLock;
 use web::Json;
 use crate::storage::subscription::{KeyEvent, SubscriptionManager};
 use jsonwebtoken::{decode, DecodingKey, Validation};
-
+use tokio::time;
+use crate::network::broadcaster::{gossip_membership};
 //
 // CONSTANTS & TYPES
 //
@@ -559,6 +560,7 @@ pub async fn join_cluster(
     }
 
     let new_node = request.node.clone();
+    let cluster2 = cluster.clone();
     let mut nodes_guard = cluster.nodes.write().await;
     if !nodes_guard.contains_key(&new_node) {
         nodes_guard.insert(
@@ -568,6 +570,14 @@ pub async fn join_cluster(
                 last_heartbeat: current_timestamp(),
             },
         );
+
+        tokio::spawn(async move {
+                time::sleep(std::time::Duration::from_secs(5)).await;
+                let args: Vec<String> = env::args().collect();
+                let current_node = args[1].clone();
+                let client = reqwest::Client::new();
+                gossip_membership(&cluster2, &client, &*current_node).await;
+        });
     }
     HttpResponse::Ok().json(nodes_guard.clone())
 }