feat: implement cluster health monitoring and chunk replication

Cargo.toml

@@ -33,6 +33,7 @@ async-stream = "0.3"
 tokio-stream = "0.1"
 uuid.workspace = true
 hex.workspace = true
+reqwest = { version = "0.12", features = ["json"] }
 
 [dev-dependencies]
 tower = "0.5"

src/bin/gateway.rs

@@ -1,4 +1,4 @@
-use monoce_gateway::{HttpGateway, AppState};
+use monoce_gateway::{HttpGateway, AppState, cluster::{health_check_loop, replication_loop}};
 use std::net::SocketAddr;
 use tracing_subscriber;
 
@@ -27,6 +27,24 @@ async fn main() -> anyhow::Result<()> {
     // Create app state
     let state = AppState::new(node_id, cluster_id);
 
+    // Spawn health check background task
+    let health_state = state.clone();
+    tokio::spawn(async move {
+        loop {
+            tokio::time::sleep(std::time::Duration::from_secs(2)).await;
+            health_check_loop(&health_state).await;
+        }
+    });
+
+    // Spawn replication reconciliation background task
+    let replication_state = state.clone();
+    tokio::spawn(async move {
+        loop {
+            tokio::time::sleep(std::time::Duration::from_secs(10)).await;
+            replication_loop(&replication_state).await;
+        }
+    });
+
     // Create and start gateway
     let gateway = HttpGateway::new(addr, state);
 

src/cluster.rs

@@ -0,0 +1,132 @@
+use crate::handlers::AppState;
+use std::time::{SystemTime, UNIX_EPOCH};
+
+pub async fn health_check_loop(state: &AppState) {
+    let mut registry = state.node_registry.write().await;
+    let now = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap()
+        .as_millis() as u64;
+
+    for (node_id, record) in registry.nodes.iter_mut() {
+        let is_healthy = check_node_health(&record.address).await;
+        let was_online = record.is_online;
+        record.is_online = is_healthy;
+        if is_healthy {
+            record.last_seen_ms = now;
+        }
+        if was_online != is_healthy {
+            if is_healthy {
+                tracing::info!("Node {} is now ONLINE", node_id);
+            } else {
+                tracing::warn!("Node {} is now OFFLINE", node_id);
+            }
+        }
+    }
+}
+
+async fn check_node_health(address: &str) -> bool {
+    let client = reqwest::Client::builder()
+        .timeout(std::time::Duration::from_secs(2))
+        .build()
+        .ok();
+    if let Some(client) = client {
+        client.get(format!("{}/health", address))
+            .send()
+            .await
+            .map(|r| r.status().is_success())
+            .unwrap_or(false)
+    } else {
+        false
+    }
+}
+
+pub async fn replication_loop(state: &AppState) {
+    // Get current chunk locations and node states
+    let registry = state.node_registry.read().await;
+    let locations = state.chunk_locations.read().await;
+
+    let online_nodes: Vec<String> = registry.nodes.values()
+        .filter(|n| n.is_online)
+        .map(|n| n.address.clone())
+        .collect();
+
+    if online_nodes.is_empty() {
+        return;
+    }
+
+    // Collect chunks that need replication
+    let mut replication_tasks: Vec<(String, String, String)> = Vec::new();
+
+    for (chunk_cid, node_addrs) in locations.iter() {
+        let online_replicas: Vec<&String> = node_addrs.iter()
+            .filter(|addr| online_nodes.contains(addr))
+            .collect();
+
+        // If under-replicated (less than 2 replicas online), try to replicate
+        if online_replicas.len() < 2 && !online_replicas.is_empty() {
+            // Find a node that doesn't have this chunk
+            let missing_nodes: Vec<&String> = online_nodes.iter()
+                .filter(|n| !node_addrs.contains(n))
+                .collect();
+
+            if let Some(dest_node) = missing_nodes.first() {
+                if let Some(src_node) = online_replicas.first() {
+                    replication_tasks.push((
+                        chunk_cid.clone(),
+                        (*src_node).clone(),
+                        (*dest_node).clone(),
+                    ));
+                }
+            }
+        }
+    }
+
+    // Release read locks before performing replications
+    drop(locations);
+    drop(registry);
+
+    // Perform replications and update locations
+    for (chunk_cid, src_node, dest_node) in replication_tasks {
+        tracing::info!(
+            "Replicating chunk {} from {} to {}",
+            chunk_cid, src_node, dest_node
+        );
+
+        match replicate_chunk(&chunk_cid, &src_node, &dest_node).await {
+            Ok(_) => {
+                // Update chunk_locations after successful replication
+                let mut locations = state.chunk_locations.write().await;
+                if let Some(addrs) = locations.get_mut(&chunk_cid) {
+                    if !addrs.contains(&dest_node) {
+                        addrs.push(dest_node.clone());
+                    }
+                }
+                tracing::info!("Successfully replicated chunk {} to {}", chunk_cid, dest_node);
+            }
+            Err(e) => {
+                tracing::error!("Failed to replicate chunk {}: {}", chunk_cid, e);
+            }
+        }
+    }
+}
+
+async fn replicate_chunk(cid: &str, src: &str, dest: &str) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    let client = reqwest::Client::new();
+
+    // Fetch from source
+    let data = client.get(format!("{}/api/v1/chunks/{}", src, cid))
+        .send()
+        .await?
+        .bytes()
+        .await?;
+
+    // Upload to destination
+    client.put(format!("{}/api/v1/chunks/{}", dest, cid))
+        .body(data)
+        .send()
+        .await?
+        .error_for_status()?;
+
+    Ok(())
+}