From 9cd7c1ff9041432f410dd41f1342c0082ac401f6 Mon Sep 17 00:00:00 2001 From: Michael Vlach Date: Wed, 9 Sep 2026 23:47:54 +0200 Subject: [PATCH 1/5] Add log catch-up resync for cluster nodes Introduce a `/cluster/logs` endpoint and log-stream format so followers can catch up missing raft entries before falling back to full snapshot sync. Add append failure tracking in raft (with a force-resync heartbeat flag after repeated append failures) to proactively trigger follower resync when a node is persistently unreachable. Also add `cluster_max_chunk_size` configuration (default 65,536) and wire it into snapshot/log streaming and extraction paths, updating tests and config initializers accordingly. --- agdb_api/src/api_types/config_impl.rs | 6 + agdb_api/src/test_server.rs | 5 +- agdb_api/src/test_server/test_cluster.rs | 2 + agdb_server/src/app.rs | 1 + agdb_server/src/cluster.rs | 144 +++++++++++++- agdb_server/src/cluster_log.rs | 2 + agdb_server/src/config.rs | 13 ++ agdb_server/src/raft.rs | 178 +++++++++++++++++- agdb_server/src/routes/cluster.rs | 88 ++++++++- .../tests/routes/cluster_test_extra.rs | 96 ++++++++++ agdb_server/tests/routes/misc_routes.rs | 5 + agdb_server/tests/tls/mod.rs | 2 + 12 files changed, 526 insertions(+), 16 deletions(-) diff --git a/agdb_api/src/api_types/config_impl.rs b/agdb_api/src/api_types/config_impl.rs index c783a8f1..b9e56bdd 100644 --- a/agdb_api/src/api_types/config_impl.rs +++ b/agdb_api/src/api_types/config_impl.rs @@ -8,6 +8,7 @@ pub const DEFAULT_TOKEN_EXPIRY_SECONDS: u64 = 3600; pub const MIN_TOKEN_EXPIRY_SECONDS: u64 = 60; pub const MAX_TOKEN_EXPIRY_SECONDS: u64 = 86400; pub const DEFAULT_CLUSTER_MAX_LOG_ENTRIES: u64 = 1000; +pub const DEFAULT_CLUSTER_MAX_CHUNK_SIZE: u64 = 65_536; #[derive(Debug)] #[cfg_attr(feature = "api", derive(agdb::TypeDef))] @@ -31,6 +32,7 @@ pub struct ConfigImpl { pub cluster_election_factor_ms: u64, pub cluster: Vec, pub cluster_max_log_entries: u64, + pub cluster_max_chunk_size: u64, pub cluster_node_id: usize, pub start_time: u64, pub token_expiry_seconds: u64, @@ -85,6 +87,10 @@ pub fn config_to_str(config: &ConfigImpl) -> String { "cluster_max_log_entries: {}\n", config.cluster_max_log_entries )); + buffer.push_str(&format!( + "cluster_max_chunk_size: {}\n", + config.cluster_max_chunk_size + )); buffer.push_str(&format!( "token_expiry_seconds: {}\n", config.token_expiry_seconds diff --git a/agdb_api/src/test_server.rs b/agdb_api/src/test_server.rs index 1096a6d7..4254141a 100644 --- a/agdb_api/src/test_server.rs +++ b/agdb_api/src/test_server.rs @@ -6,6 +6,7 @@ use crate::AgdbApi; use crate::QueryAudit; use crate::ReqwestClient; use crate::config_impl::ConfigImpl; +use crate::config_impl::DEFAULT_CLUSTER_MAX_CHUNK_SIZE; use crate::config_impl::DEFAULT_CLUSTER_MAX_LOG_ENTRIES; use crate::config_impl::DEFAULT_LOG_BODY_LIMIT; use crate::config_impl::DEFAULT_REQUEST_BODY_LIMIT; @@ -320,6 +321,7 @@ impl TestServerImpl { cluster_election_factor_ms: 1000, cluster: Vec::new(), cluster_max_log_entries: DEFAULT_CLUSTER_MAX_LOG_ENTRIES, + cluster_max_chunk_size: DEFAULT_CLUSTER_MAX_CHUNK_SIZE, cluster_node_id: 0, start_time: 0, token_expiry_seconds: DEFAULT_TOKEN_EXPIRY_SECONDS, @@ -331,7 +333,8 @@ impl TestServerImpl { } pub fn next_port() -> u16 { - PORT.fetch_add(1, Ordering::Relaxed) + std::process::id() as u16 + PORT.fetch_add(1, Ordering::Relaxed) + .wrapping_add(std::process::id() as u16) } pub fn restart(&mut self) -> Result<(), TestError> { diff --git a/agdb_api/src/test_server/test_cluster.rs b/agdb_api/src/test_server/test_cluster.rs index 7ecb2a56..d23f791d 100644 --- a/agdb_api/src/test_server/test_cluster.rs +++ b/agdb_api/src/test_server/test_cluster.rs @@ -3,6 +3,7 @@ use crate::ClusterStatus; use crate::LogLevelFilter; use crate::ReqwestClient; use crate::config_impl::ConfigImpl; +use crate::config_impl::DEFAULT_CLUSTER_MAX_CHUNK_SIZE; use crate::config_impl::DEFAULT_CLUSTER_MAX_LOG_ENTRIES; use crate::config_impl::DEFAULT_LOG_BODY_LIMIT; use crate::config_impl::DEFAULT_REQUEST_BODY_LIMIT; @@ -160,6 +161,7 @@ async fn create_cluster_impl( cluster_election_factor_ms: 250, cluster: Vec::new(), cluster_max_log_entries: max_log_entries, + cluster_max_chunk_size: DEFAULT_CLUSTER_MAX_CHUNK_SIZE, cluster_node_id: 0, start_time: 0, token_expiry_seconds: DEFAULT_TOKEN_EXPIRY_SECONDS, diff --git a/agdb_server/src/app.rs b/agdb_server/src/app.rs index a0c3be24..3dc582f5 100644 --- a/agdb_server/src/app.rs +++ b/agdb_server/src/app.rs @@ -202,6 +202,7 @@ pub(crate) fn app( "/cluster/admin/user/logout_all", routing::post(routes::cluster::admin_logout_all), ) + .route("/cluster/logs", routing::get(routes::cluster::logs)) .route("/cluster/snapshot", routing::get(routes::cluster::snapshot)) .route("/cluster/status", routing::get(routes::cluster::status)) .route("/user/login", routing::post(routes::user::login)) diff --git a/agdb_server/src/cluster.rs b/agdb_server/src/cluster.rs index 43c7ce4a..b67b8b38 100644 --- a/agdb_server/src/cluster.rs +++ b/agdb_server/src/cluster.rs @@ -285,7 +285,13 @@ async fn start_cluster( Err(e) => crate::warn!( "[{index}] Error sending response to cluster node '{node_index}': {e:?}" ), + } + } else if matches!(request.data, raft::RequestType::Append(_)) { + let fail_response = raft::Response { + target: request.index, + result: raft::ResponseType::CommitError("send failed".into()), }; + let _ = node.responses.send((request, fail_response)); } } else { break; @@ -388,19 +394,34 @@ async fn resync_from_leader(cluster: &Cluster, config: &Config) -> ServerResult< } }; + cluster.resync.store(true, Ordering::Relaxed); + + let from_index = cluster.raft.read().await.storage.log_commit(); + match catchup_logs_from_leader(cluster, config, leader_index, from_index).await { + Ok(()) => { + cluster.resync.store(false, Ordering::Relaxed); + crate::info!("[{}] Resync completed via log catch-up", cluster.index); + return Ok(()); + } + Err(e) => { + crate::info!( + "[{}] Log catch-up failed ({e:?}), falling back to full snapshot", + cluster.index + ); + } + } + let mut snapshot_sources: Vec = (0..cluster.nodes.len()) .filter(|index| *index != cluster.index && *index != leader_index) .collect(); snapshot_sources.push(leader_index); crate::info!( - "[{}] Starting resync, snapshot candidates: {:?}", + "[{}] Starting snapshot resync, candidates: {:?}", cluster.index, snapshot_sources ); - cluster.resync.store(true, Ordering::Relaxed); - let mut result = Err(ServerError::from("no snapshot source available")); for source_index in snapshot_sources { @@ -430,15 +451,112 @@ async fn resync_from_leader(cluster: &Cluster, config: &Config) -> ServerResult< cluster.resync.store(false, Ordering::Relaxed); match &result { - Ok(()) => crate::info!("[{}] Resync completed successfully", cluster.index), + Ok(()) => crate::info!("[{}] Resync completed via snapshot", cluster.index), Err(e) => crate::error!("[{}] Resync failed: {:?}", cluster.index, e), } result } +async fn catchup_logs_from_leader( + cluster: &Cluster, + config: &Config, + leader_index: usize, + from_index: u64, +) -> ServerResult<()> { + let logs_url = format!( + "{}/api/v1/cluster/logs?from_index={from_index}", + cluster.nodes[leader_index].base_url + ); + + let response = cluster.nodes[leader_index] + .client + .client + .get(&logs_url) + .bearer_auth(&config.cluster_token) + .timeout(Duration::from_secs(600)) + .send() + .await + .map_err(|e| ServerError::from(format!("log catch-up request failed: {e:?}")))?; + + let status = response.status().as_u16(); + if status != 200 { + let body = response.text().await.unwrap_or_default(); + return Err(ServerError::from(format!( + "log catch-up endpoint returned {status}: {body}", + ))); + } + + let body = response + .bytes() + .await + .map_err(|e| ServerError::from(format!("log catch-up stream error: {e:?}")))?; + + if body.len() < 16 { + return Err(ServerError::from( + "log catch-up response too small (missing header)", + )); + } + + let entry_count = u64::from_le_bytes(body[0..8].try_into().unwrap()); + let commit_index = u64::from_le_bytes(body[8..16].try_into().unwrap()); + + const MAX_CATCHUP_ENTRIES: u64 = 100_000; + if entry_count > MAX_CATCHUP_ENTRIES { + return Err(ServerError::from(format!( + "log catch-up entry count {entry_count} exceeds limit {MAX_CATCHUP_ENTRIES}", + ))); + } + + let mut offset = 16usize; + let mut entries = Vec::with_capacity(entry_count as usize); + + for _ in 0..entry_count { + if offset + 8 > body.len() { + return Err(ServerError::from("log catch-up truncated (json_len)")); + } + let json_len = u64::from_le_bytes(body[offset..offset + 8].try_into().unwrap()) as usize; + offset += 8; + + if offset + json_len > body.len() { + return Err(ServerError::from("log catch-up truncated (json_bytes)")); + } + let log: Log = serde_json::from_slice(&body[offset..offset + json_len]) + .map_err(|e| ServerError::from(format!("log catch-up deserialization error: {e}")))?; + entries.push(log); + offset += json_len; + } + + if let Some(first) = entries.first() + && first.index > from_index + 1 + { + return Err(ServerError::from(format!( + "log catch-up gap: need index {}, first available is {}", + from_index + 1, + first.index + ))); + } + + let mut raft = cluster.raft.write().await; + for log in &entries { + raft.storage.append(log.clone(), None).await?; + } + if commit_index > raft.storage.commit { + raft.storage.commit(commit_index).await?; + } + raft.refresh_local_from_storage(); + + crate::info!( + "[{}] Log catch-up complete: {} entries applied, commit_index={}", + cluster.index, + entries.len(), + commit_index + ); + + Ok(()) +} + const SNAPSHOT_PARTIAL_TTL_SECS: u64 = 3600; -const SNAPSHOT_EXTRACT_CHUNK: usize = 65536; async fn validate_snapshot_header(partial_file: &Path, current_commit: u64) -> ServerResult<()> { let mut file = tokio::fs::File::open(partial_file).await?; @@ -477,7 +595,13 @@ async fn do_resync(cluster: &Cluster, config: &Config, node_index: usize) -> Ser let current_commit = cluster.raft.read().await.storage.commit; validate_snapshot_header(&partial_file, current_commit).await?; - if let Err(e) = extract_snapshot_binary(&partial_file, &install_dir).await { + if let Err(e) = extract_snapshot_binary( + &partial_file, + &install_dir, + config.cluster_max_chunk_size as usize, + ) + .await + { let _ = std::fs::remove_dir_all(&install_dir); return Err(e); } @@ -598,7 +722,11 @@ async fn download_snapshot_to_partial( Ok(()) } -async fn extract_snapshot_binary(partial_file: &Path, install_dir: &Path) -> ServerResult<()> { +async fn extract_snapshot_binary( + partial_file: &Path, + install_dir: &Path, + chunk_size: usize, +) -> ServerResult<()> { std::fs::create_dir_all(install_dir)?; let mut file = tokio::fs::File::open(partial_file).await?; @@ -619,7 +747,7 @@ async fn extract_snapshot_binary(partial_file: &Path, install_dir: &Path) -> Ser } let canonical_install = install_dir.canonicalize()?; - let mut buf = vec![0u8; SNAPSHOT_EXTRACT_CHUNK]; + let mut buf = vec![0u8; chunk_size]; for _ in 0..file_count { let mut len_buf = [0u8; 4]; diff --git a/agdb_server/src/cluster_log.rs b/agdb_server/src/cluster_log.rs index adb6cf96..7bbbaaad 100644 --- a/agdb_server/src/cluster_log.rs +++ b/agdb_server/src/cluster_log.rs @@ -380,6 +380,7 @@ mod tests { use agdb::SyncMode; use agdb_api::LogLevelFilter; use agdb_api::config_impl::ConfigImpl; + use agdb_api::config_impl::DEFAULT_CLUSTER_MAX_CHUNK_SIZE; use agdb_api::config_impl::DEFAULT_CLUSTER_MAX_LOG_ENTRIES; use agdb_api::config_impl::DEFAULT_LOG_BODY_LIMIT; use agdb_api::config_impl::DEFAULT_REQUEST_BODY_LIMIT; @@ -435,6 +436,7 @@ mod tests { cluster_election_factor_ms: 1000, cluster: vec![], cluster_max_log_entries: DEFAULT_CLUSTER_MAX_LOG_ENTRIES, + cluster_max_chunk_size: DEFAULT_CLUSTER_MAX_CHUNK_SIZE, cluster_node_id: 0, start_time: SystemTime::now() .duration_since(UNIX_EPOCH) diff --git a/agdb_server/src/config.rs b/agdb_server/src/config.rs index bf00b5a9..843a4d94 100644 --- a/agdb_server/src/config.rs +++ b/agdb_server/src/config.rs @@ -1,6 +1,7 @@ use agdb::SyncMode; use agdb_api::LogLevelFilter; use agdb_api::config_impl::ConfigImpl; +use agdb_api::config_impl::DEFAULT_CLUSTER_MAX_CHUNK_SIZE; use agdb_api::config_impl::DEFAULT_CLUSTER_MAX_LOG_ENTRIES; use agdb_api::config_impl::DEFAULT_LOG_BODY_LIMIT; use agdb_api::config_impl::DEFAULT_REQUEST_BODY_LIMIT; @@ -174,6 +175,14 @@ pub(crate) fn from_str(content: &str) -> Result { .parse() .map_err(|e| format!("Invalid cluster_max_log_entries: {e:?}"))? } + "cluster_max_chunk_size" => { + config.cluster_max_chunk_size = value + .parse() + .map_err(|e| format!("Invalid cluster_max_chunk_size: {e:?}"))?; + if config.cluster_max_chunk_size == 0 { + config.cluster_max_chunk_size = DEFAULT_CLUSTER_MAX_CHUNK_SIZE; + } + } "token_expiry_seconds" => { config.token_expiry_seconds = value .parse() @@ -241,6 +250,7 @@ fn default_config() -> ConfigImpl { cluster_election_factor_ms: 1000, cluster: vec![], cluster_max_log_entries: DEFAULT_CLUSTER_MAX_LOG_ENTRIES, + cluster_max_chunk_size: DEFAULT_CLUSTER_MAX_CHUNK_SIZE, cluster_node_id: 0, start_time: 0, token_expiry_seconds: DEFAULT_TOKEN_EXPIRY_SECONDS, @@ -306,6 +316,7 @@ mod tests { cluster_election_factor_ms: 1000, cluster: vec![], cluster_max_log_entries: DEFAULT_CLUSTER_MAX_LOG_ENTRIES, + cluster_max_chunk_size: DEFAULT_CLUSTER_MAX_CHUNK_SIZE, cluster_node_id: 0, start_time: 0, token_expiry_seconds: DEFAULT_TOKEN_EXPIRY_SECONDS, @@ -347,6 +358,7 @@ mod tests { cluster_election_factor_ms: 1000, cluster: vec![], cluster_max_log_entries: DEFAULT_CLUSTER_MAX_LOG_ENTRIES, + cluster_max_chunk_size: DEFAULT_CLUSTER_MAX_CHUNK_SIZE, cluster_node_id: 0, start_time: 0, token_expiry_seconds: DEFAULT_TOKEN_EXPIRY_SECONDS, @@ -387,6 +399,7 @@ mod tests { cluster_election_factor_ms: 1000, cluster: vec![], cluster_max_log_entries: DEFAULT_CLUSTER_MAX_LOG_ENTRIES, + cluster_max_chunk_size: DEFAULT_CLUSTER_MAX_CHUNK_SIZE, cluster_node_id: 0, start_time: 0, token_expiry_seconds: DEFAULT_TOKEN_EXPIRY_SECONDS, diff --git a/agdb_server/src/raft.rs b/agdb_server/src/raft.rs index f850abbe..844a14af 100644 --- a/agdb_server/src/raft.rs +++ b/agdb_server/src/raft.rs @@ -66,7 +66,9 @@ pub(crate) struct Request { log_term: u64, log_commit: u64, prune_index: u64, - data: RequestType, + #[serde(default)] + force_resync: bool, + pub(crate) data: RequestType, } #[derive(Debug, Serialize, Deserialize)] @@ -75,6 +77,8 @@ pub(crate) struct Response { pub(crate) result: ResponseType, } +const APPEND_FAILURE_THRESHOLD: u32 = 3; + struct Node { index: u64, log_index: u64, @@ -82,6 +86,8 @@ struct Node { log_commit: u64, timer: Instant, voted: bool, + append_failures: u32, + force_resync: bool, } pub(crate) trait Storage { @@ -150,6 +156,8 @@ impl> Cluster { }, timer: Instant::now(), voted: i == settings.index, + append_failures: 0, + force_resync: false, }) .collect(), hash: settings.hash, @@ -204,6 +212,7 @@ impl> Cluster { log_term: self.local().log_term, log_commit: self.local().log_commit, prune_index, + force_resync: false, data: RequestType::Append(vec![log.clone()]), }) .collect(); @@ -307,7 +316,14 @@ impl> Cluster { match (&self.state, &request.data, &response.result) { (Election, PreVote, OK) => Ok(self.pre_vote_received(request)), (Candidate, Vote, OK) => Ok(self.vote_received(request)), - (Leader, Heartbeat | Append(_), OK) => self.commit(request).await, + (Leader, Heartbeat | Append(_), OK) => { + self.node_mut(request.target).append_failures = 0; + if request.force_resync { + Ok(None) + } else { + self.commit(request).await + } + } (Leader, Heartbeat | Append(_), LogMismatch(mismatch)) => { self.reconcile(request, mismatch).await } @@ -322,6 +338,28 @@ impl> Cluster { } Ok(None) } + (Leader, Append(_), ResponseType::CommitError(e)) => { + self.node_mut(request.target).append_failures += 1; + let failures = self.node(request.target).append_failures; + crate::info!( + "[{}] Node {} append failed ({}/{}): {}", + self.index, + request.target, + failures, + APPEND_FAILURE_THRESHOLD, + e + ); + if failures >= APPEND_FAILURE_THRESHOLD { + self.node_mut(request.target).force_resync = true; + self.node_mut(request.target).append_failures = 0; + crate::warn!( + "[{}] Node {} exceeded append failure threshold, forcing resync", + self.index, + request.target, + ); + } + Ok(None) + } (_, _, ResponseType::CommitError(e)) => { crate::info!( "[{}] Node {} commit error: {}", @@ -371,6 +409,7 @@ impl> Cluster { log_term: self.local().log_term, log_commit: self.local().log_commit, prune_index: self.storage.prune_index(), + force_resync: false, data: RequestType::Append(logs), }])) } @@ -394,6 +433,7 @@ impl> Cluster { log_term: self.local().log_term, log_commit: self.local().log_commit, prune_index: 0, + force_resync: false, data: RequestType::PreVote, }) .collect() @@ -461,6 +501,7 @@ impl> Cluster { log_term: self.local().log_term, log_commit: self.local().log_commit, prune_index: 0, + force_resync: false, data: RequestType::Vote, }) .collect() @@ -552,7 +593,9 @@ impl> Cluster { fn heartbeat(&mut self) -> Vec> { let prune_index = self.storage.prune_index(); - self.nodes + let log_commit = self.local().log_commit; + let requests: Vec> = self + .nodes .iter() .filter(|node| { self.index != node.index @@ -565,11 +608,18 @@ impl> Cluster { term: self.term, log_index: self.local().log_index, log_term: self.local().log_term, - log_commit: self.local().log_commit, + log_commit, prune_index, + force_resync: node.force_resync, data: RequestType::Heartbeat, }) - .collect() + .collect(); + + for req in &requests { + self.node_mut(req.target).force_resync = false; + } + + requests } fn heartbeat_no_timer(&mut self) -> Vec> { @@ -587,6 +637,7 @@ impl> Cluster { log_term: self.local().log_term, log_commit: self.local().log_commit, prune_index, + force_resync: false, data: RequestType::Heartbeat, }) .collect(); @@ -603,6 +654,14 @@ impl> Cluster { self.validate_term(request)?; self.become_follower(request); + if request.force_resync + && (self.local().log_index != request.log_index + || self.local().log_term != request.log_term) + { + self.needs_resync = true; + return Self::ok(request); + } + if request.prune_index > 0 && request.prune_index > self.storage.log_commit() { self.needs_resync = true; return Self::ok(request); @@ -1442,6 +1501,7 @@ mod test { log_term: 1, log_commit: 10, prune_index: 5, + force_resync: false, data: RequestType::Heartbeat, }; @@ -1508,6 +1568,7 @@ mod test { log_term: 1, log_commit: 10, prune_index: 5, + force_resync: false, data: RequestType::Heartbeat, }; @@ -1560,6 +1621,7 @@ mod test { log_term: 1, log_commit: 10, prune_index: 50, // leader has pruned up to 50 — way past our commit + force_resync: false, data: RequestType::Heartbeat, }; @@ -1570,4 +1632,110 @@ mod test { Ok(()) } + + #[tokio::test] + async fn force_resync_flag_triggers_resync() -> anyhow::Result<()> { + // Follower at log_index=10, log_commit=10 + let storage = TestStorage { + logs: (1..=10) + .map(|i| Log { + db_id: None, + index: i, + term: 1, + data: i as u8, + }) + .collect(), + commit: 10, + prune_index: 0, + }; + let settings = ClusterSettings { + index: 1, + size: 3, + hash: 123, + election_factor_ms: 1000, + heartbeat_timeout: Duration::from_secs(1), + term_timeout: Duration::from_secs(3), + max_log_entries: 1000, + }; + let mut cluster: Cluster = Cluster::new(storage, settings); + cluster.state = ClusterState::Follower(0); + cluster.term = 1; + + assert!(!cluster.needs_resync()); + + // Heartbeat with force_resync=true — follower's log_index (10) differs + // from leader's (20), so the flag triggers needs_resync. + let request = Request { + hash: 123, + index: 0, + target: 1, + term: 1, + log_index: 20, + log_term: 1, + log_commit: 20, + prune_index: 0, + force_resync: true, + data: RequestType::Heartbeat, + }; + + let response = cluster.request(&request).await; + assert_eq!(response.result, ResponseType::Ok); + assert!(cluster.needs_resync()); + + Ok(()) + } + + #[tokio::test] + async fn force_resync_skipped_when_already_caught_up() -> anyhow::Result<()> { + // Follower already at log_index=10, log_commit=10 + let storage = TestStorage { + logs: (1..=10) + .map(|i| Log { + db_id: None, + index: i, + term: 1, + data: i as u8, + }) + .collect(), + commit: 10, + prune_index: 0, + }; + let settings = ClusterSettings { + index: 1, + size: 3, + hash: 123, + election_factor_ms: 1000, + heartbeat_timeout: Duration::from_secs(1), + term_timeout: Duration::from_secs(3), + max_log_entries: 1000, + }; + let mut cluster: Cluster = Cluster::new(storage, settings); + cluster.state = ClusterState::Follower(0); + cluster.term = 1; + + assert!(!cluster.needs_resync()); + + // Heartbeat with force_resync=true, but follower's log state matches the + // leader's (same log_index and log_term) — the follower caught up through + // normal reconciliation before the force_resync heartbeat arrived. + // Resync should NOT trigger. + let request = Request { + hash: 123, + index: 0, + target: 1, + term: 1, + log_index: 10, + log_term: 1, + log_commit: 10, + prune_index: 0, + force_resync: true, + data: RequestType::Heartbeat, + }; + + let response = cluster.request(&request).await; + assert_eq!(response.result, ResponseType::Ok); + assert!(!cluster.needs_resync()); + + Ok(()) + } } diff --git a/agdb_server/src/routes/cluster.rs b/agdb_server/src/routes/cluster.rs index dd0b8758..2285b260 100644 --- a/agdb_server/src/routes/cluster.rs +++ b/agdb_server/src/routes/cluster.rs @@ -284,7 +284,11 @@ pub(crate) async fn status( } const SNAPSHOT_STAGING_TTL_SECS: u64 = 600; -const SNAPSHOT_STREAM_CHUNK: usize = 65536; + +#[derive(serde::Deserialize)] +pub(crate) struct LogsQuery { + from_index: u64, +} pub(crate) async fn snapshot( _cluster_id: ClusterId, @@ -361,6 +365,7 @@ pub(crate) async fn snapshot( log_term, log_commit, cluster.snapshot_in_flight.clone(), + config.cluster_max_chunk_size as usize, ) { Ok(b) => b, Err(e) => { @@ -556,6 +561,7 @@ fn snapshot_body_stream( log_term: u64, log_commit: u64, in_flight: Arc, + chunk_size: usize, ) -> ServerResult { let files = list_snapshot_files(&staging_dir)?; Ok(Body::from_stream(snapshot_file_stream( @@ -565,6 +571,7 @@ fn snapshot_body_stream( log_term, log_commit, in_flight, + chunk_size, ))) } @@ -575,6 +582,7 @@ fn snapshot_file_stream( log_term: u64, log_commit: u64, in_flight: Arc, + chunk_size: usize, ) -> impl futures::Stream> + Send + 'static { let file_count = files.len() as u64; let guard = SnapshotInFlightGuard(in_flight.clone()); @@ -596,7 +604,7 @@ fn snapshot_file_stream( } pos = 32; - let mut buf = vec![0u8; SNAPSHOT_STREAM_CHUNK]; + let mut buf = vec![0u8; chunk_size]; for (rel_path, abs_path) in files { use tokio::io::AsyncReadExt; @@ -671,6 +679,82 @@ fn collect_staging_files( Ok(()) } +pub(crate) async fn logs( + _cluster_id: ClusterId, + State(cluster): State, + State(config): State, + Query(params): Query, +) -> ServerResult { + if cluster.resync.load(Ordering::Acquire) { + return axum::response::Response::builder() + .status(StatusCode::SERVICE_UNAVAILABLE) + .body(Body::from("resyncing")) + .map_err(|e| ServerError::from(e.to_string())); + } + + let raft = cluster.raft.read().await; + let entries = raft + .storage + .cluster_log + .logs_since(params.from_index) + .await?; + let commit_index = raft.storage.commit; + drop(raft); + + if entries.is_empty() { + return axum::response::Response::builder() + .status(StatusCode::NOT_FOUND) + .body(Body::from("no entries available")) + .map_err(|e| ServerError::from(e.to_string())); + } + + let chunk_size = config.cluster_max_chunk_size as usize; + let body = Body::from_stream(log_entries_byte_stream(entries, commit_index, chunk_size)); + + axum::response::Response::builder() + .status(StatusCode::OK) + .header("content-type", "application/octet-stream") + .body(body) + .map_err(|e| ServerError::from(e.to_string())) +} + +fn log_entries_byte_stream( + logs: Vec>, + commit_index: u64, + chunk_size: usize, +) -> impl futures::Stream> + Send + 'static { + async_stream::try_stream! { + let entry_count = logs.len() as u64; + + let mut header = [0u8; 16]; + header[0..8].copy_from_slice(&entry_count.to_le_bytes()); + header[8..16].copy_from_slice(&commit_index.to_le_bytes()); + yield bytes::Bytes::copy_from_slice(&header); + + let mut buf = Vec::with_capacity(chunk_size); + + for log in &logs { + let json = serde_json::to_vec(log) + .map_err(std::io::Error::other)?; + let len_bytes = (json.len() as u64).to_le_bytes(); + + if !buf.is_empty() && buf.len() + 8 + json.len() > chunk_size { + yield bytes::Bytes::from(std::mem::take(&mut buf)); + } + + buf.extend_from_slice(&len_bytes); + buf.extend_from_slice(&json); + + if buf.len() >= chunk_size { + yield bytes::Bytes::from(std::mem::take(&mut buf)); + } + } + if !buf.is_empty() { + yield bytes::Bytes::from(buf); + } + } +} + fn cleanup_stale_snapshot_stagings( data_dir: &std::path::Path, current_commit: u64, diff --git a/agdb_server/tests/routes/cluster_test_extra.rs b/agdb_server/tests/routes/cluster_test_extra.rs index 1cbec7e5..ff0d917c 100644 --- a/agdb_server/tests/routes/cluster_test_extra.rs +++ b/agdb_server/tests/routes/cluster_test_extra.rs @@ -170,6 +170,102 @@ async fn rebalance() -> Result<(), TestError> { Ok(()) } +#[tokio::test] +async fn log_catchup_after_append_failures() -> Result<(), TestError> { + // Default max_log_entries (1000): logs are NOT pruned. + // When the follower is down, Append delivery fails repeatedly. + // After APPEND_FAILURE_THRESHOLD consecutive failures the leader sets + // force_resync, causing the next heartbeat to carry an elevated + // prune_index (= leader's log_commit). The follower sees + // prune_index > its own log_commit → needs_resync → + // resync_from_leader tries log catch-up first, which succeeds + // because the entries are still available (no pruning). + let mut servers = create_cluster(3, false).await?; + let mut follower = AgdbApi::new( + ReqwestClient::with_client(reqwest_client()), + &servers[2].address, + ); + follower.cluster_user_login(ADMIN, ADMIN).await?; + follower.admin_shutdown().await?; + servers[2].wait().await?; + + let mut leader = AgdbApi::new( + ReqwestClient::with_client(reqwest_client()), + &servers[0].address, + ); + leader.user_login(ADMIN, ADMIN).await?; + leader + .db_add(ADMIN, "log_catchup_test", DbKind::Mapped) + .await?; + leader + .db_exec_mut( + ADMIN, + "log_catchup_test", + &[QueryBuilder::insert() + .nodes() + .aliases("root") + .values(vec![vec![("key", 1).into()]]) + .query() + .into()], + ) + .await?; + + // Each mutation triggers an Append to the downed follower which fails, + // incrementing append_failures. After 3 failures force_resync is set. + for i in 0..10 { + leader + .db_exec_mut( + ADMIN, + "log_catchup_test", + &[QueryBuilder::insert() + .values(vec![vec![("key", i).into()]]) + .ids("root") + .query() + .into()], + ) + .await?; + } + + servers[2].restart()?; + wait_for_ready(&follower).await?; + + let node1 = AgdbApi::new( + ReqwestClient::with_client(reqwest_client()), + &servers[1].address, + ); + wait_for_leader(&node1).await?; + + let mut synced = false; + + for _ in 0..10 { + if follower.user_login(ADMIN, ADMIN).await.is_ok() + && let Ok(result) = follower + .db_exec( + ADMIN, + "log_catchup_test", + &[QueryBuilder::select() + .values("key") + .ids("root") + .query() + .into()], + ) + .await + && let Ok(value) = result.1[0].elements[0].values[0].value.to_u64() + { + if value == 9 { + synced = true; + break; + } + } + + tokio::time::sleep(std::time::Duration::from_secs(1)).await; + } + + assert!(synced, "follower did not sync after append failures"); + + Ok(()) +} + fn cluster_log_entry_count(data_dir: &str) -> u64 { let log_path = format!("{data_dir}/agdb_server.log"); let db = agdb::Db::new(&log_path).expect("failed to open cluster log"); diff --git a/agdb_server/tests/routes/misc_routes.rs b/agdb_server/tests/routes/misc_routes.rs index f9ff4497..3b137b73 100644 --- a/agdb_server/tests/routes/misc_routes.rs +++ b/agdb_server/tests/routes/misc_routes.rs @@ -5,6 +5,7 @@ use agdb_api::DbKind; use agdb_api::LogLevelFilter; use agdb_api::ReqwestClient; use agdb_api::config_impl::ConfigImpl; +use agdb_api::config_impl::DEFAULT_CLUSTER_MAX_CHUNK_SIZE; use agdb_api::config_impl::DEFAULT_CLUSTER_MAX_LOG_ENTRIES; use agdb_api::config_impl::DEFAULT_LOG_BODY_LIMIT; use agdb_api::config_impl::DEFAULT_REQUEST_BODY_LIMIT; @@ -165,6 +166,7 @@ async fn basepath_test() -> anyhow::Result<()> { cluster_election_factor_ms: 1000, cluster: Vec::new(), cluster_max_log_entries: DEFAULT_CLUSTER_MAX_LOG_ENTRIES, + cluster_max_chunk_size: DEFAULT_CLUSTER_MAX_CHUNK_SIZE, cluster_node_id: 0, start_time: 0, token_expiry_seconds: DEFAULT_TOKEN_EXPIRY_SECONDS, @@ -343,6 +345,7 @@ async fn large_payload() -> anyhow::Result<()> { cluster_election_factor_ms: 1000, cluster: Vec::new(), cluster_max_log_entries: DEFAULT_CLUSTER_MAX_LOG_ENTRIES, + cluster_max_chunk_size: DEFAULT_CLUSTER_MAX_CHUNK_SIZE, cluster_node_id: 0, start_time: 0, token_expiry_seconds: DEFAULT_TOKEN_EXPIRY_SECONDS, @@ -435,6 +438,7 @@ async fn static_files() -> anyhow::Result<()> { cluster_election_factor_ms: 1000, cluster: Vec::new(), cluster_max_log_entries: DEFAULT_CLUSTER_MAX_LOG_ENTRIES, + cluster_max_chunk_size: DEFAULT_CLUSTER_MAX_CHUNK_SIZE, cluster_node_id: 0, start_time: 0, token_expiry_seconds: DEFAULT_TOKEN_EXPIRY_SECONDS, @@ -504,6 +508,7 @@ async fn static_files_with_basepath() -> anyhow::Result<()> { cluster_election_factor_ms: 1000, cluster: Vec::new(), cluster_max_log_entries: DEFAULT_CLUSTER_MAX_LOG_ENTRIES, + cluster_max_chunk_size: DEFAULT_CLUSTER_MAX_CHUNK_SIZE, cluster_node_id: 0, start_time: 0, token_expiry_seconds: DEFAULT_TOKEN_EXPIRY_SECONDS, diff --git a/agdb_server/tests/tls/mod.rs b/agdb_server/tests/tls/mod.rs index 32689058..13128573 100644 --- a/agdb_server/tests/tls/mod.rs +++ b/agdb_server/tests/tls/mod.rs @@ -3,6 +3,7 @@ use agdb_api::AgdbApi; use agdb_api::LogLevelFilter; use agdb_api::ReqwestClient; use agdb_api::config_impl::ConfigImpl; +use agdb_api::config_impl::DEFAULT_CLUSTER_MAX_CHUNK_SIZE; use agdb_api::config_impl::DEFAULT_CLUSTER_MAX_LOG_ENTRIES; use agdb_api::config_impl::DEFAULT_LOG_BODY_LIMIT; use agdb_api::config_impl::DEFAULT_REQUEST_BODY_LIMIT; @@ -39,6 +40,7 @@ async fn https() -> anyhow::Result<()> { cluster_election_factor_ms: 1000, cluster: Vec::new(), cluster_max_log_entries: DEFAULT_CLUSTER_MAX_LOG_ENTRIES, + cluster_max_chunk_size: DEFAULT_CLUSTER_MAX_CHUNK_SIZE, cluster_node_id: 0, start_time: 0, token_expiry_seconds: DEFAULT_TOKEN_EXPIRY_SECONDS, From bdc40f56fa7abd16972f1db40c2a17c0723fe8cc Mon Sep 17 00:00:00 2001 From: Michael Vlach Date: Wed, 9 Sep 2026 23:51:55 +0200 Subject: [PATCH 2/5] Update cluster_test_extra.rs --- agdb_server/tests/routes/cluster_test_extra.rs | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/agdb_server/tests/routes/cluster_test_extra.rs b/agdb_server/tests/routes/cluster_test_extra.rs index ff0d917c..3f170905 100644 --- a/agdb_server/tests/routes/cluster_test_extra.rs +++ b/agdb_server/tests/routes/cluster_test_extra.rs @@ -251,11 +251,10 @@ async fn log_catchup_after_append_failures() -> Result<(), TestError> { ) .await && let Ok(value) = result.1[0].elements[0].values[0].value.to_u64() + && value == 9 { - if value == 9 { - synced = true; - break; - } + synced = true; + break; } tokio::time::sleep(std::time::Duration::from_secs(1)).await; From 6ae1bbda898157f35a77c07e0c4028ec58776651 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 9 Sep 2026 22:10:32 +0000 Subject: [PATCH 3/5] Fix 5 review comments: streaming, contiguity, commit validation, lock, data encapsulation Co-authored-by: michaelvlach <11575751+michaelvlach@users.noreply.github.com> --- agdb_server/src/cluster.rs | 106 +++++++++++++++++++----------- agdb_server/src/raft.rs | 8 ++- agdb_server/src/routes/cluster.rs | 13 ++-- 3 files changed, 81 insertions(+), 46 deletions(-) diff --git a/agdb_server/src/cluster.rs b/agdb_server/src/cluster.rs index b67b8b38..4a96a2b3 100644 --- a/agdb_server/src/cluster.rs +++ b/agdb_server/src/cluster.rs @@ -286,7 +286,7 @@ async fn start_cluster( "[{index}] Error sending response to cluster node '{node_index}': {e:?}" ), } - } else if matches!(request.data, raft::RequestType::Append(_)) { + } else if request.is_append() { let fail_response = raft::Response { target: request.index, result: raft::ResponseType::CommitError("send failed".into()), @@ -487,19 +487,27 @@ async fn catchup_logs_from_leader( ))); } - let body = response - .bytes() - .await - .map_err(|e| ServerError::from(format!("log catch-up stream error: {e:?}")))?; - - if body.len() < 16 { - return Err(ServerError::from( - "log catch-up response too small (missing header)", - )); + let mut stream = response.bytes_stream(); + let mut buf: Vec = Vec::new(); + + while buf.len() < 16 { + match stream.next().await { + Some(Ok(chunk)) => buf.extend_from_slice(&chunk), + Some(Err(e)) => { + return Err(ServerError::from(format!( + "log catch-up stream error: {e:?}" + ))) + } + None => { + return Err(ServerError::from( + "log catch-up response too small (missing header)", + )) + } + } } - let entry_count = u64::from_le_bytes(body[0..8].try_into().unwrap()); - let commit_index = u64::from_le_bytes(body[8..16].try_into().unwrap()); + let entry_count = u64::from_le_bytes(buf[0..8].try_into().unwrap()); + let commit_index = u64::from_le_bytes(buf[8..16].try_into().unwrap()); const MAX_CATCHUP_ENTRIES: u64 = 100_000; if entry_count > MAX_CATCHUP_ENTRIES { @@ -508,39 +516,63 @@ async fn catchup_logs_from_leader( ))); } - let mut offset = 16usize; - let mut entries = Vec::with_capacity(entry_count as usize); + buf.drain(0..16); + + let mut raft = cluster.raft.write().await; + let mut prev_index = from_index; + let mut last_applied_index: Option = None; for _ in 0..entry_count { - if offset + 8 > body.len() { - return Err(ServerError::from("log catch-up truncated (json_len)")); + while buf.len() < 8 { + match stream.next().await { + Some(Ok(chunk)) => buf.extend_from_slice(&chunk), + Some(Err(e)) => { + return Err(ServerError::from(format!( + "log catch-up stream error: {e:?}" + ))) + } + None => return Err(ServerError::from("log catch-up truncated (json_len)")), + } } - let json_len = u64::from_le_bytes(body[offset..offset + 8].try_into().unwrap()) as usize; - offset += 8; - - if offset + json_len > body.len() { - return Err(ServerError::from("log catch-up truncated (json_bytes)")); + let json_len = u64::from_le_bytes(buf[0..8].try_into().unwrap()) as usize; + buf.drain(0..8); + + while buf.len() < json_len { + match stream.next().await { + Some(Ok(chunk)) => buf.extend_from_slice(&chunk), + Some(Err(e)) => { + return Err(ServerError::from(format!( + "log catch-up stream error: {e:?}" + ))) + } + None => return Err(ServerError::from("log catch-up truncated (json_bytes)")), + } } - let log: Log = serde_json::from_slice(&body[offset..offset + json_len]) + let log: Log = serde_json::from_slice(&buf[..json_len]) .map_err(|e| ServerError::from(format!("log catch-up deserialization error: {e}")))?; - entries.push(log); - offset += json_len; - } + buf.drain(0..json_len); - if let Some(first) = entries.first() - && first.index > from_index + 1 - { - return Err(ServerError::from(format!( - "log catch-up gap: need index {}, first available is {}", - from_index + 1, - first.index - ))); + if log.index != prev_index + 1 { + return Err(ServerError::from(format!( + "log catch-up non-contiguous: expected index {}, got {}", + prev_index + 1, + log.index + ))); + } + + prev_index = log.index; + last_applied_index = Some(log.index); + raft.storage.append(log, None).await?; } - let mut raft = cluster.raft.write().await; - for log in &entries { - raft.storage.append(log.clone(), None).await?; + if let Some(last_index) = last_applied_index { + if commit_index > last_index { + return Err(ServerError::from(format!( + "log catch-up commit_index {commit_index} exceeds last applied index {last_index}", + ))); + } } + if commit_index > raft.storage.commit { raft.storage.commit(commit_index).await?; } @@ -549,7 +581,7 @@ async fn catchup_logs_from_leader( crate::info!( "[{}] Log catch-up complete: {} entries applied, commit_index={}", cluster.index, - entries.len(), + entry_count, commit_index ); diff --git a/agdb_server/src/raft.rs b/agdb_server/src/raft.rs index 844a14af..70f72de4 100644 --- a/agdb_server/src/raft.rs +++ b/agdb_server/src/raft.rs @@ -68,7 +68,13 @@ pub(crate) struct Request { prune_index: u64, #[serde(default)] force_resync: bool, - pub(crate) data: RequestType, + data: RequestType, +} + +impl Request { + pub(crate) fn is_append(&self) -> bool { + matches!(self.data, RequestType::Append(_)) + } } #[derive(Debug, Serialize, Deserialize)] diff --git a/agdb_server/src/routes/cluster.rs b/agdb_server/src/routes/cluster.rs index 2285b260..397b0785 100644 --- a/agdb_server/src/routes/cluster.rs +++ b/agdb_server/src/routes/cluster.rs @@ -692,14 +692,11 @@ pub(crate) async fn logs( .map_err(|e| ServerError::from(e.to_string())); } - let raft = cluster.raft.read().await; - let entries = raft - .storage - .cluster_log - .logs_since(params.from_index) - .await?; - let commit_index = raft.storage.commit; - drop(raft); + let (cluster_log, commit_index) = { + let raft = cluster.raft.read().await; + (raft.storage.cluster_log.clone(), raft.storage.commit) + }; + let entries = cluster_log.logs_since(params.from_index).await?; if entries.is_empty() { return axum::response::Response::builder() From 7e9a79ba922ddfb8bba0feb11072e84e1136df2f Mon Sep 17 00:00:00 2001 From: Michael Vlach Date: Thu, 10 Sep 2026 00:14:19 +0200 Subject: [PATCH 4/5] Update cluster.rs --- agdb_server/src/cluster.rs | 20 ++++++++++---------- 1 file changed, 10 insertions(+), 10 deletions(-) diff --git a/agdb_server/src/cluster.rs b/agdb_server/src/cluster.rs index 4a96a2b3..2c93eac1 100644 --- a/agdb_server/src/cluster.rs +++ b/agdb_server/src/cluster.rs @@ -496,12 +496,12 @@ async fn catchup_logs_from_leader( Some(Err(e)) => { return Err(ServerError::from(format!( "log catch-up stream error: {e:?}" - ))) + ))); } None => { return Err(ServerError::from( "log catch-up response too small (missing header)", - )) + )); } } } @@ -529,7 +529,7 @@ async fn catchup_logs_from_leader( Some(Err(e)) => { return Err(ServerError::from(format!( "log catch-up stream error: {e:?}" - ))) + ))); } None => return Err(ServerError::from("log catch-up truncated (json_len)")), } @@ -543,7 +543,7 @@ async fn catchup_logs_from_leader( Some(Err(e)) => { return Err(ServerError::from(format!( "log catch-up stream error: {e:?}" - ))) + ))); } None => return Err(ServerError::from("log catch-up truncated (json_bytes)")), } @@ -565,12 +565,12 @@ async fn catchup_logs_from_leader( raft.storage.append(log, None).await?; } - if let Some(last_index) = last_applied_index { - if commit_index > last_index { - return Err(ServerError::from(format!( - "log catch-up commit_index {commit_index} exceeds last applied index {last_index}", - ))); - } + if let Some(last_index) = last_applied_index + && commit_index > last_index + { + return Err(ServerError::from(format!( + "log catch-up commit_index {commit_index} exceeds last applied index {last_index}", + ))); } if commit_index > raft.storage.commit { From afae8a75e67b3b289f9b4587ea1134097c8f975a Mon Sep 17 00:00:00 2001 From: Michael Vlach Date: Thu, 10 Sep 2026 00:31:30 +0200 Subject: [PATCH 5/5] Update cluster.rs --- agdb_server/src/cluster.rs | 30 ++++++++++++++++++------------ 1 file changed, 18 insertions(+), 12 deletions(-) diff --git a/agdb_server/src/cluster.rs b/agdb_server/src/cluster.rs index 2c93eac1..86731732 100644 --- a/agdb_server/src/cluster.rs +++ b/agdb_server/src/cluster.rs @@ -518,9 +518,8 @@ async fn catchup_logs_from_leader( buf.drain(0..16); - let mut raft = cluster.raft.write().await; + let mut entries = Vec::with_capacity(entry_count as usize); let mut prev_index = from_index; - let mut last_applied_index: Option = None; for _ in 0..entry_count { while buf.len() < 8 { @@ -561,21 +560,28 @@ async fn catchup_logs_from_leader( } prev_index = log.index; - last_applied_index = Some(log.index); - raft.storage.append(log, None).await?; + entries.push(log); } - if let Some(last_index) = last_applied_index - && commit_index > last_index - { - return Err(ServerError::from(format!( - "log catch-up commit_index {commit_index} exceeds last applied index {last_index}", - ))); + let mut raft = cluster.raft.write().await; + let last_index = entries.last().map(|log| log.index); + + for log in entries { + raft.storage.append(log, None).await?; } - if commit_index > raft.storage.commit { - raft.storage.commit(commit_index).await?; + if let Some(last_index) = last_index { + if commit_index > last_index { + return Err(ServerError::from(format!( + "log catch-up commit_index {commit_index} exceeds last applied index {last_index}", + ))); + } + + if commit_index > raft.storage.commit { + raft.storage.commit(commit_index).await?; + } } + raft.refresh_local_from_storage(); crate::info!(