Support for bigger clusters.

src/handlers.rs

@@ -4,7 +4,15 @@ use axum::{
     Json,
 };
 use chrono::Utc;
-use crate::tls::{create_join_token, issue_leaf_cert, JoinClusterRequest, JoinClusterResponse};
+use crate::{
+    db::{meta_get, meta_set},
+    tls::{
+        create_join_token, create_mtls_client, issue_leaf_cert, JoinClusterRequest,
+        JoinClusterResponse,
+    },
+};
+use std::time::Duration;
+use serde::Deserialize;
 
 use rusqlite::params;
 use serde_json::Value;
@@ -36,7 +44,7 @@ pub async fn create_token_handler(
     State(state): State<AppState>,
     payload: Option<Json<Value>>,
 ) -> Result<Json<Value>, ApiError> {
-    if state.role == Role::Replica {
+    if state.is_replica() {
         return Err(api_error(StatusCode::FORBIDDEN, "Only Primary can generate join tokens"));
     }
 
@@ -61,7 +69,7 @@ pub async fn join_cluster_handler(
     State(state): State<AppState>,
     Json(req): Json<JoinClusterRequest>,
 ) -> Result<Json<JoinClusterResponse>, ApiError> {
-    if state.role == Role::Replica {
+    if state.is_replica() {
         return Err(api_error(StatusCode::FORBIDDEN, "Only Primary can issue certificates"));
     }
 
@@ -71,7 +79,9 @@ pub async fn join_cluster_handler(
         .ok_or_else(|| internal_error("CA cert unavailable"))?;
     let ca_key_pem = state
         .ca_key_pem
-        .as_ref()
+        .read()
+        .map_err(|_| internal_error("CA key lock poisoned"))?
+        .clone()
         .ok_or_else(|| internal_error("CA key unavailable"))?;
 
     // 1. Validate & burn the token
@@ -106,7 +116,7 @@ pub async fn join_cluster_handler(
 
     let (cert_pem, key_pem) = issue_leaf_cert(
         ca_cert_pem,
-        ca_key_pem,
+        &ca_key_pem,
         &req.node_id.to_string(),
         sans,
     )
@@ -118,6 +128,7 @@ pub async fn join_cluster_handler(
         ca_pem: ca_cert_pem.to_string(),
         cert_pem,
         key_pem,
+        ca_key_pem: Some(ca_key_pem),
     }))
 }
 
@@ -141,12 +152,13 @@ pub async fn create_record(
 ) -> Result<(StatusCode, Json<ApiResponse>), ApiError> {
     info!(endpoint = "/api/records", "Create request received");
 
-   if state.role == Role::Replica {
+    if state.is_replica() {
+        let p_url = state.get_primary_url();
         return Err(api_error(
             StatusCode::FORBIDDEN,
             format!(
                 "This instance is a read-only replica; write to {}",
-                state.primary_url.as_deref().map(String::as_str).unwrap_or("the primary")
+                p_url.as_deref().unwrap_or("the primary")
             ),
         ));
     }
@@ -285,3 +297,338 @@ pub async fn reload_config(State(state): State<AppState>) -> Result<Json<Value>,
         "fields": field_count
     })))
 }
+
+#[derive(Debug, Deserialize, Default)]
+pub struct StepDownRequest {
+    pub primary_url: Option<String>,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct RepointRequest {
+    pub primary_url: String,
+    pub reset_source: Option<bool>,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct SwitchoverRequest {
+    pub replica_url: String,
+    pub timeout_secs: Option<u64>,
+}
+
+/// Admin endpoint: Demote this primary to read-only replica.
+pub async fn step_down_handler(
+    State(state): State<AppState>,
+    payload: Option<Json<StepDownRequest>>,
+) -> Result<Json<Value>, ApiError> {
+    info!("Step-down requested");
+
+    // 1. Update role in memory
+    {
+        let mut r = state
+            .role
+            .write()
+            .map_err(|_| internal_error("role lock poisoned"))?;
+        *r = Role::Replica;
+    }
+
+    // 2. Stop any existing replica loop
+    state.stop_replica_sync();
+
+    // 3. Update local head seq as baseline for last_applied_seq
+    let local_head: i64 = {
+        let db = state.db.lock().map_err(|_| internal_error("db poisoned"))?;
+        db.query_row("SELECT COALESCE(MAX(seq), 0) FROM records", [], |r| {
+            r.get(0)
+        })
+        .map_err(|e| db_error("get local head", e))?
+    };
+
+    {
+        let db = state.db.lock().map_err(|_| internal_error("db poisoned"))?;
+        meta_set(&db, "role", "replica").map_err(|e| db_error("save role", e))?;
+        meta_set(&db, "last_applied_seq", &local_head.to_string())
+            .map_err(|e| db_error("save last_applied_seq", e))?;
+        let _ = db.execute("DELETE FROM meta WHERE key = 'source_instance_id'", []);
+    }
+
+    if let Ok(mut s) = state.sync.write() {
+        s.state = "idle".into();
+        s.last_applied_seq = local_head;
+        s.source_instance_id = None;
+        s.primary_head_seq = Some(local_head);
+        s.last_error = None;
+    }
+
+    if let Some(Json(req)) = payload {
+        if let Some(p_url) = req.primary_url {
+            let trimmed = p_url.trim_end_matches('/').to_string();
+            if !trimmed.is_empty() {
+                state.set_primary_url(Some(trimmed.clone()));
+                let db = state.db.lock().map_err(|_| internal_error("db poisoned"))?;
+                let _ = meta_set(&db, "primary_url", &trimmed);
+                drop(db);
+                state.start_replica_sync();
+            }
+        }
+    }
+
+    info!(role = "replica", last_applied_seq = local_head, "Node stepped down to replica");
+    Ok(Json(serde_json::json!({
+        "status": "ok",
+        "role": "replica",
+        "primary_url": state.get_primary_url(),
+        "last_applied_seq": local_head,
+    })))
+}
+
+/// Admin endpoint: Promote this replica to Primary.
+pub async fn promote_handler(
+    State(state): State<AppState>,
+) -> Result<Json<Value>, ApiError> {
+    info!("Promotion to Primary requested");
+
+    // 1. Stop pulling changes
+    state.stop_replica_sync();
+
+    // 2. Set role to Primary
+    {
+        let mut r = state
+            .role
+            .write()
+            .map_err(|_| internal_error("role lock poisoned"))?;
+        *r = Role::Primary;
+    }
+
+    let local_head: i64 = {
+        let db = state.db.lock().map_err(|_| internal_error("db poisoned"))?;
+        db.query_row("SELECT COALESCE(MAX(seq), 0) FROM records", [], |r| {
+            r.get(0)
+        })
+        .map_err(|e| db_error("get local head", e))?
+    };
+
+    let epoch = {
+        let db = state.db.lock().map_err(|_| internal_error("db poisoned"))?;
+        meta_set(&db, "role", "primary").map_err(|e| db_error("save role", e))?;
+        let cur_epoch: i64 = meta_get(&db, "cluster_epoch")
+            .ok()
+            .flatten()
+            .and_then(|v| v.parse().ok())
+            .unwrap_or(1);
+        let next_epoch = cur_epoch + 1;
+        meta_set(&db, "cluster_epoch", &next_epoch.to_string())
+            .map_err(|e| db_error("save epoch", e))?;
+        next_epoch
+    };
+
+    if let Ok(mut s) = state.sync.write() {
+        s.state = "primary".into();
+        s.primary_head_seq = None;
+        s.last_error = None;
+    }
+
+    info!(role = "primary", head_seq = local_head, epoch, "Node successfully promoted to Primary");
+    Ok(Json(serde_json::json!({
+        "status": "ok",
+        "role": "primary",
+        "instance_id": state.instance_id,
+        "head_seq": local_head,
+        "epoch": epoch,
+    })))
+}
+
+/// Admin endpoint: Repoint replica to a new primary URL.
+pub async fn repoint_handler(
+    State(state): State<AppState>,
+    Json(req): Json<RepointRequest>,
+) -> Result<Json<Value>, ApiError> {
+    info!(primary_url = %req.primary_url, "Repoint requested");
+
+    let p_url = req.primary_url.trim_end_matches('/').to_string();
+    state.set_primary_url(Some(p_url.clone()));
+
+    {
+        let db = state.db.lock().map_err(|_| internal_error("db poisoned"))?;
+        meta_set(&db, "primary_url", &p_url).map_err(|e| db_error("save primary_url", e))?;
+        if req.reset_source.unwrap_or(true) {
+            let _ = db.execute("DELETE FROM meta WHERE key = 'source_instance_id'", []);
+        }
+    }
+
+    if req.reset_source.unwrap_or(true) {
+        if let Ok(mut s) = state.sync.write() {
+            s.source_instance_id = None;
+            s.last_error = None;
+        }
+    }
+
+    if state.is_replica() {
+        state.start_replica_sync();
+    }
+
+    Ok(Json(serde_json::json!({
+        "status": "ok",
+        "role": state.get_role().as_str(),
+        "primary_url": p_url,
+    })))
+}
+
+/// Admin endpoint: Initiates coordinated graceful switchover from Primary to a chosen Replica.
+pub async fn switchover_handler(
+    State(state): State<AppState>,
+    Json(req): Json<SwitchoverRequest>,
+) -> Result<Json<Value>, ApiError> {
+    info!(target = %req.replica_url, "Coordinated cluster role switchover initiated");
+
+    if !state.is_primary() {
+        return Err(api_error(
+            StatusCode::BAD_REQUEST,
+            "Switchover can only be initiated on the primary node",
+        ));
+    }
+
+    let replica_url = req.replica_url.trim_end_matches('/').to_string();
+    let timeout_secs = req.timeout_secs.unwrap_or(15);
+
+    // Build mTLS client to interact with the replica
+    let ca_pem = state
+        .ca_cert_pem
+        .as_ref()
+        .ok_or_else(|| internal_error("ca_cert_pem unavailable"))?;
+    let cert_pem = state
+        .node_cert_pem
+        .as_ref()
+        .ok_or_else(|| internal_error("node_cert_pem unavailable"))?;
+    let key_pem = state
+        .node_key_pem
+        .as_ref()
+        .ok_or_else(|| internal_error("node_key_pem unavailable"))?;
+
+    let client = create_mtls_client(ca_pem, cert_pem, key_pem)
+        .map_err(|e| internal_error(&format!("failed to create mtls client: {e}")))?;
+
+    // Step 1: Quiesce this primary by demoting to replica (stops accepting new writes)
+    {
+        let mut r = state
+            .role
+            .write()
+            .map_err(|_| internal_error("role lock poisoned"))?;
+        *r = Role::Replica;
+    }
+
+    let local_head: i64 = {
+        let db = state.db.lock().map_err(|_| internal_error("db poisoned"))?;
+        db.query_row("SELECT COALESCE(MAX(seq), 0) FROM records", [], |r| {
+            r.get(0)
+        })
+        .map_err(|e| db_error("get local head", e))?
+    };
+
+    // Step 2: Poll replica status until last_applied_seq >= local_head
+    info!(local_head, "Waiting for replica to catch up to head_seq...");
+    let start = std::time::Instant::now();
+    let mut caught_up = false;
+
+    while start.elapsed() < Duration::from_secs(timeout_secs) {
+        let status_url = format!("{replica_url}/api/replication/status");
+        match client.get(&status_url).send().await {
+            Ok(resp) if resp.status().is_success() => {
+                if let Ok(json) = resp.json::<Value>().await {
+                    let repl_applied = json
+                        .get("sync")
+                        .and_then(|s| s.get("last_applied_seq"))
+                        .and_then(|v| v.as_i64())
+                        .unwrap_or(0);
+
+                    if repl_applied >= local_head {
+                        caught_up = true;
+                        break;
+                    }
+                }
+            }
+            Ok(resp) => {
+                warn!(status = %resp.status(), "Replica returned non-success during status check");
+            }
+            Err(e) => {
+                warn!(error = %e, "Failed to query replica status during switchover");
+            }
+        }
+        tokio::time::sleep(Duration::from_millis(200)).await;
+    }
+
+    if !caught_up {
+        // Rollback to primary
+        error!("Replica did not catch up within timeout; rolling back role to Primary");
+        let mut r = state
+            .role
+            .write()
+            .map_err(|_| internal_error("role lock poisoned"))?;
+        *r = Role::Primary;
+        return Err(api_error(
+            StatusCode::GATEWAY_TIMEOUT,
+            format!(
+                "Replica at {replica_url} failed to catch up to seq {local_head} within {timeout_secs}s"
+            ),
+        ));
+    }
+
+    // Step 3: Promote replica to Primary
+    let promote_url = format!("{replica_url}/api/admin/promote");
+    let resp = match client.post(&promote_url).send().await {
+        Ok(r) => r,
+        Err(e) => {
+            error!(error = %e, "Failed to call promote on replica; rolling back");
+            let mut r = state
+                .role
+                .write()
+                .map_err(|_| internal_error("role lock poisoned"))?;
+            *r = Role::Primary;
+            return Err(api_error(
+                StatusCode::BAD_GATEWAY,
+                format!("Failed to reach replica for promotion: {e}"),
+            ));
+        }
+    };
+
+    if !resp.status().is_success() {
+        error!(status = %resp.status(), "Replica promotion failed; rolling back");
+        let mut r = state
+            .role
+            .write()
+            .map_err(|_| internal_error("role lock poisoned"))?;
+        *r = Role::Primary;
+        return Err(api_error(
+            StatusCode::BAD_GATEWAY,
+            format!("Replica rejected promotion with status {}", resp.status()),
+        ));
+    }
+
+    // Step 4: Configure this node as a replica of the new primary
+    {
+        let db = state.db.lock().map_err(|_| internal_error("db poisoned"))?;
+        let _ = meta_set(&db, "role", "replica");
+        let _ = meta_set(&db, "primary_url", &replica_url);
+        let _ = meta_set(&db, "last_applied_seq", &local_head.to_string());
+        let _ = db.execute("DELETE FROM meta WHERE key = 'source_instance_id'", []);
+    }
+
+    if let Ok(mut s) = state.sync.write() {
+        s.state = "idle".into();
+        s.last_applied_seq = local_head;
+        s.source_instance_id = None;
+        s.primary_head_seq = Some(local_head);
+        s.last_error = None;
+    }
+
+    state.set_primary_url(Some(replica_url.clone()));
+    state.start_replica_sync();
+
+    info!(new_primary = %replica_url, "Graceful switchover succeeded");
+    Ok(Json(serde_json::json!({
+        "status": "ok",
+        "message": "Role switchover completed successfully",
+        "old_primary_role": "replica",
+        "new_primary_url": replica_url,
+        "switched_at_seq": local_head,
+    })))
+}

src/main.rs

@@ -28,11 +28,12 @@ use crate::{
     db::{init_db, load_or_create_instance_id, meta_get, meta_set},
     handlers::{
         create_record, create_token_handler, get_cluster_ca, get_config, get_info,
-        join_cluster_handler, list_records, reload_config,
+        join_cluster_handler, list_records, promote_handler, reload_config, repoint_handler,
+        step_down_handler, switchover_handler,
     },
-    replication::{replica_loop, replication_changes, replication_status, SyncConfig},
+    replication::{replication_changes, replication_status},
     schema::load_config,
-    state::{AppState, Role, SyncStatus},
+    state::{AppState, Role, SyncConfig, SyncStatus},
     tls::{
         bootstrap_replica_join, create_mtls_server_config, generate_ca, issue_leaf_cert,
         JoinToken,
@@ -137,18 +138,24 @@ async fn main() {
     } else {
         // Replica: Check if dynamic bootstrap enrollment is needed
         if node_cert_pem.is_none() {
-            let p_url = primary_url
-                .as_ref()
-                .expect("APP_PRIMARY_URL required for replica enrollment");
+            // APP_JOIN_URL lets the replication source (APP_PRIMARY_URL) differ from the
+            // CA enrollment target (APP_JOIN_URL). This is required in chain topologies
+            // where intermediate replicas serve data but cannot issue certificates.
+            // Falls back to APP_PRIMARY_URL when not set.
+            let join_url = std::env::var("APP_JOIN_URL")
+                .ok()
+                .or_else(|| primary_url.clone())
+                .expect("APP_PRIMARY_URL (or APP_JOIN_URL) required for replica enrollment");
             let token = std::env::var("APP_JOIN_TOKEN")
                 .expect("APP_JOIN_TOKEN required for initial replica enrollment");
 
-            info!(primary = %p_url, "Bootstrapping mTLS credentials via join token...");
-            bootstrap_replica_join(p_url, &token, instance_id, &db)
+            info!(join_url = %join_url, "Bootstrapping mTLS credentials via join token...");
+            bootstrap_replica_join(&join_url, &token, instance_id, &db)
                 .await
                 .expect("Dynamic join failed");
 
             ca_cert_pem = meta_get(&db, "tls_ca_pem").ok().flatten();
+            ca_key_pem = meta_get(&db, "tls_ca_key_pem").ok().flatten();
             node_cert_pem = meta_get(&db, "tls_cert_pem").ok().flatten();
             node_key_pem = meta_get(&db, "tls_key_pem").ok().flatten();
         }
@@ -171,13 +178,19 @@ async fn main() {
         "Instance initialised"
     );
 
+    let sync_cfg = SyncConfig {
+        poll_interval: Duration::from_secs(env_parse("APP_SYNC_INTERVAL_SECS", 5)),
+        max_backoff: Duration::from_secs(env_parse("APP_SYNC_MAX_BACKOFF_SECS", 60)),
+        batch: env_parse("APP_SYNC_BATCH", 200i64).clamp(1, 1000),
+    };
+
     let state = AppState {
         db: Arc::new(Mutex::new(db)),
         schema: Arc::new(RwLock::new(schema)),
         config_path: Arc::new(config_path),
         instance_id,
-        role,
-        primary_url: primary_url.clone().map(Arc::new),
+        role: Arc::new(RwLock::new(role)),
+        primary_url: Arc::new(RwLock::new(primary_url)),
         repl_token: repl_token.clone().map(Arc::new),
         sync: Arc::new(RwLock::new(SyncStatus {
             state: if role == Role::Replica { "idle" } else { "primary" }.into(),
@@ -187,27 +200,17 @@ async fn main() {
             last_success: None,
             last_error: None,
         })),
+        sync_cancel: Arc::new(Mutex::new(None)),
+        sync_cfg,
         ca_cert_pem: ca_cert_pem.clone().map(Arc::new),
-        ca_key_pem: ca_key_pem.map(Arc::new),
+        ca_key_pem: Arc::new(RwLock::new(ca_key_pem)),
+        node_cert_pem: node_cert_pem.clone().map(Arc::new),
+        node_key_pem: node_key_pem.clone().map(Arc::new),
         join_tokens,
     };
 
     if role == Role::Replica {
-        let sync_cfg = SyncConfig {
-            poll_interval: Duration::from_secs(env_parse("APP_SYNC_INTERVAL_SECS", 5)),
-            max_backoff: Duration::from_secs(env_parse("APP_SYNC_MAX_BACKOFF_SECS", 60)),
-            batch: env_parse("APP_SYNC_BATCH", 200i64).clamp(1, 1000),
-        };
-
-        tokio::spawn(replica_loop(
-            state.clone(),
-            primary_url.expect("checked above"),
-            repl_token.unwrap_or_else(|| "".into()),
-            sync_cfg,
-            ca_cert_pem.clone().unwrap(),
-            node_cert_pem.clone().unwrap(),
-            node_key_pem.clone().unwrap(),
-        ));
+        state.start_replica_sync();
     }
 
     let app = Router::new()
@@ -216,6 +219,10 @@ async fn main() {
         .route("/api/records", get(list_records).post(create_record))
         .route("/api/admin/config", get(get_config))
         .route("/api/admin/reload-config", post(reload_config))
+        .route("/api/admin/step-down", post(step_down_handler))
+        .route("/api/admin/promote", post(promote_handler))
+        .route("/api/admin/repoint", post(repoint_handler))
+        .route("/api/admin/switchover", post(switchover_handler))
         .route("/api/replication/changes", get(replication_changes))
         .route("/api/replication/status", get(replication_status))
         // Dynamic Peer Discovery Routes:

src/replication.rs

@@ -18,7 +18,7 @@ use uuid::Uuid;
 
 use crate::{
     db::meta_set,
-    state::{api_error, db_error, internal_error, ApiError, AppState, SyncStatus},
+    state::{api_error, db_error, internal_error, ApiError, AppState, SyncConfig, SyncStatus},
 };
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -43,13 +43,6 @@ pub struct ChangesQuery {
     pub limit: Option<i64>,
 }
 
-#[derive(Debug, Clone)]
-pub struct SyncConfig {
-    pub poll_interval: Duration,
-    pub max_backoff: Duration,
-    pub batch: i64,
-}
-
 enum SyncError {
     Transient(String),
     Fatal(String),
@@ -165,10 +158,13 @@ pub async fn replication_status(State(state): State<AppState>) -> Result<Json<Va
         .primary_head_seq
         .map(|head| (head - status.last_applied_seq).max(0));
 
+    let current_role = state.get_role();
+    let primary_url = state.get_primary_url();
+
     Ok(Json(serde_json::json!({
         "instance_id": state.instance_id,
-        "role": state.role.as_str(),
-        "primary_url": state.primary_url.as_deref(),
+        "role": current_role.as_str(),
+        "primary_url": primary_url,
         "local_head_seq": local_head,
         "lag_records": lag,
         "sync": status,
@@ -335,6 +331,7 @@ pub async fn replica_loop(
     ca_pem: String,
     cert_pem: String,
     key_pem: String,
+    mut cancel_rx: tokio::sync::watch::Receiver<bool>,
 ) {
     let client = match create_mtls_client(&ca_pem, &cert_pem, &key_pem) {
         Ok(c) => c,
@@ -348,30 +345,61 @@ pub async fn replica_loop(
     info!(primary = %primary_url, "Replica mTLS sync loop started");
 
     loop {
-        match sync_once(&client, &state, &primary_url, &token, cfg.batch).await {
-            Ok(more) => {
-                backoff = Duration::from_secs(1);
-                if !more {
-                    tokio::time::sleep(jitter(cfg.poll_interval)).await;
+        if *cancel_rx.borrow() {
+            info!("Replica sync loop stopped via cancellation token");
+            break;
+        }
+
+        tokio::select! {
+            _ = cancel_rx.changed() => {
+                if *cancel_rx.borrow() {
+                    info!("Replica sync loop received cancellation");
+                    break;
                 }
             }
-            Err(SyncError::Transient(message)) => {
-                let wait = jitter(backoff);
-                warn!(reason = %message, retry_in = ?wait, "Replication pull failed");
-                update_status(&state, |s| {
-                    s.state = "retrying".into();
-                    s.last_error = Some(message);
-                });
-                tokio::time::sleep(wait).await;
-                backoff = (backoff * 2).min(cfg.max_backoff);
-            }
-            Err(SyncError::Fatal(message)) => {
-                error!(reason = %message, "Replication halted; operator action required");
-                update_status(&state, |s| {
-                    s.state = "halted".into();
-                    s.last_error = Some(message);
-                });
-                return;
+            res = sync_once(&client, &state, &primary_url, &token, cfg.batch) => {
+                match res {
+                    Ok(more) => {
+                        backoff = Duration::from_secs(1);
+                        if !more {
+                            tokio::select! {
+                                _ = cancel_rx.changed() => {
+                                    if *cancel_rx.borrow() {
+                                        info!("Replica sync loop cancelled while waiting for next poll");
+                                        break;
+                                    }
+                                }
+                                _ = tokio::time::sleep(jitter(cfg.poll_interval)) => {}
+                            }
+                        }
+                    }
+                    Err(SyncError::Transient(message)) => {
+                        let wait = jitter(backoff);
+                        warn!(reason = %message, retry_in = ?wait, "Replication pull failed");
+                        update_status(&state, |s| {
+                            s.state = "retrying".into();
+                            s.last_error = Some(message);
+                        });
+                        tokio::select! {
+                            _ = cancel_rx.changed() => {
+                                if *cancel_rx.borrow() {
+                                    info!("Replica sync loop cancelled during backoff sleep");
+                                    break;
+                                }
+                            }
+                            _ = tokio::time::sleep(wait) => {}
+                        }
+                        backoff = (backoff * 2).min(cfg.max_backoff);
+                    }
+                    Err(SyncError::Fatal(message)) => {
+                        error!(reason = %message, "Replication halted; operator action required");
+                        update_status(&state, |s| {
+                            s.state = "halted".into();
+                            s.last_error = Some(message);
+                        });
+                        return;
+                    }
+                }
             }
         }
     }

src/state.rs

@@ -5,30 +5,137 @@ use serde_json::Value;
 use std::{
     collections::HashMap,
     sync::{Arc, Mutex, RwLock},
+    time::Duration,
 };
-use tracing::error;
+use tracing::{error, info, warn};
 use uuid::Uuid;
 
 use crate::{schema::SchemaConfig, tls::JoinToken};
 
+#[derive(Debug, Clone)]
+pub struct SyncConfig {
+    pub poll_interval: Duration,
+    pub max_backoff: Duration,
+    pub batch: i64,
+}
+
 #[derive(Clone)]
 pub struct AppState {
     pub db: Arc<Mutex<Connection>>,
     pub schema: Arc<RwLock<SchemaConfig>>,
     pub config_path: Arc<String>,
     pub instance_id: Uuid,
-    pub role: Role,
-    pub primary_url: Option<Arc<String>>,
+    pub role: Arc<RwLock<Role>>,
+    pub primary_url: Arc<RwLock<Option<String>>>,
     pub repl_token: Option<Arc<String>>,
     pub sync: Arc<RwLock<SyncStatus>>,
 
+    // Replication task management:
+    pub sync_cancel: Arc<Mutex<Option<tokio::sync::watch::Sender<bool>>>>,
+    pub sync_cfg: SyncConfig,
+
     // mTLS & Discovery additions:
     pub ca_cert_pem: Option<Arc<String>>,
-    pub ca_key_pem: Option<Arc<String>>,
+    pub ca_key_pem: Arc<RwLock<Option<String>>>,
+    pub node_cert_pem: Option<Arc<String>>,
+    pub node_key_pem: Option<Arc<String>>,
     pub join_tokens: Arc<RwLock<HashMap<String, JoinToken>>>,
 }
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+impl AppState {
+    pub fn get_role(&self) -> Role {
+        *self.role.read().unwrap()
+    }
+
+    pub fn is_primary(&self) -> bool {
+        self.get_role() == Role::Primary
+    }
+
+    pub fn is_replica(&self) -> bool {
+        self.get_role() == Role::Replica
+    }
+
+    pub fn get_primary_url(&self) -> Option<String> {
+        self.primary_url.read().unwrap().clone()
+    }
+
+    pub fn set_primary_url(&self, url: Option<String>) {
+        let mut p = self.primary_url.write().unwrap();
+        *p = url;
+    }
+
+    pub fn stop_replica_sync(&self) {
+        let mut cancel = self.sync_cancel.lock().unwrap();
+        if let Some(tx) = cancel.take() {
+            let _ = tx.send(true);
+            info!("Sent cancellation signal to replica sync loop");
+        }
+    }
+
+    pub fn start_replica_sync(&self) {
+        self.stop_replica_sync();
+
+        let primary_url = match self.get_primary_url() {
+            Some(u) if !u.is_empty() => u,
+            _ => {
+                warn!("Cannot start replica sync loop: primary_url is not set");
+                return;
+            }
+        };
+
+        let token = self
+            .repl_token
+            .as_ref()
+            .map(|t| t.as_str().to_string())
+            .unwrap_or_default();
+        let ca_pem = match self.ca_cert_pem.as_ref() {
+            Some(ca) => ca.as_str().to_string(),
+            None => {
+                warn!("Cannot start replica sync loop: ca_cert_pem missing");
+                return;
+            }
+        };
+        let cert_pem = match self.node_cert_pem.as_ref() {
+            Some(c) => c.as_str().to_string(),
+            None => {
+                warn!("Cannot start replica sync loop: node_cert_pem missing");
+                return;
+            }
+        };
+        let key_pem = match self.node_key_pem.as_ref() {
+            Some(k) => k.as_str().to_string(),
+            None => {
+                warn!("Cannot start replica sync loop: node_key_pem missing");
+                return;
+            }
+        };
+
+        let (cancel_tx, cancel_rx) = tokio::sync::watch::channel(false);
+        {
+            let mut cancel = self.sync_cancel.lock().unwrap();
+            *cancel = Some(cancel_tx);
+        }
+
+        let state_clone = self.clone();
+        let sync_cfg = self.sync_cfg.clone();
+
+        tokio::spawn(async move {
+            crate::replication::replica_loop(
+                state_clone,
+                primary_url,
+                token,
+                sync_cfg,
+                ca_pem,
+                cert_pem,
+                key_pem,
+                cancel_rx,
+            )
+            .await;
+        });
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 pub enum Role {
     Primary,
     Replica,

src/tls.rs

@@ -34,6 +34,8 @@ pub struct JoinClusterResponse {
     pub ca_pem: String,
     pub cert_pem: String,
     pub key_pem: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub ca_key_pem: Option<String>,
 }
 
 // ---------------------------------------------------------------------------
@@ -213,6 +215,9 @@ pub async fn bootstrap_replica_join(
     meta_set(db, "tls_ca_pem", &payload.ca_pem).map_err(|e| e.to_string())?;
     meta_set(db, "tls_cert_pem", &payload.cert_pem).map_err(|e| e.to_string())?;
     meta_set(db, "tls_key_pem", &payload.key_pem).map_err(|e| e.to_string())?;
+    if let Some(ref ca_key) = payload.ca_key_pem {
+        meta_set(db, "tls_ca_key_pem", ca_key).map_err(|e| e.to_string())?;
+    }
 
     info!("Replica successfully joined cluster and persisted mTLS certificates");
     Ok(())