From 516f7fecc148abddbcc62bff3be701e9d070b43c Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 26 Jul 2026 01:25:37 +0800 Subject: [PATCH] fix(tier): converge config after peer recovery (#5240) Retry peer tier config reloads after committed mutations so recovered nodes converge without requiring a second admin change. Co-authored-by: heihutu --- .../src/inline_fast_path_cluster_test.rs | 38 +- crates/ecstore/src/cluster/rpc/mod.rs | 1 + .../src/cluster/rpc/peer_rest_client.rs | 191 +++++++- .../ecstore/src/services/notification_sys.rs | 453 +++++++++++++++++- rustfs/src/admin/handlers/tier.rs | 26 +- 5 files changed, 670 insertions(+), 39 deletions(-) diff --git a/crates/e2e_test/src/inline_fast_path_cluster_test.rs b/crates/e2e_test/src/inline_fast_path_cluster_test.rs index 30132a5e0..c3a1d9ff4 100644 --- a/crates/e2e_test/src/inline_fast_path_cluster_test.rs +++ b/crates/e2e_test/src/inline_fast_path_cluster_test.rs @@ -966,6 +966,15 @@ fn unique_tier_name() -> String { } async fn add_rustfs_tier(hot: &RustFSTestClusterEnvironment, cold: &RustFSTestEnvironment, tier_name: &str) -> TestResult { + let response = submit_rustfs_tier(hot, cold, tier_name).await?; + wait_for_tier_verifiable(hot, tier_name, &response).await +} + +async fn submit_rustfs_tier( + hot: &RustFSTestClusterEnvironment, + cold: &RustFSTestEnvironment, + tier_name: &str, +) -> TestResult { let body = serde_json::json!({ "type": "rustfs", "rustfs": { @@ -994,8 +1003,7 @@ async fn add_rustfs_tier(hot: &RustFSTestClusterEnvironment, cold: &RustFSTestEn .await?; let attempt = format!("status={status}, body={}", compact_body(&response)); if status.is_success() { - wait_for_tier_verifiable(hot, tier_name, &format!("status={status}, body={response}")).await?; - return Ok(()); + return Ok(format!("status={status}, body={response}")); } attempts.push(attempt); if Instant::now() >= deadline { @@ -1549,7 +1557,7 @@ async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls() -> Te #[tokio::test] #[serial] -async fn four_node_add_tier_committed_replay_converges() -> TestResult { +async fn four_node_add_tier_converges() -> TestResult { init_logging(); let mut cold = RustFSTestEnvironment::new().await?; @@ -1563,7 +1571,29 @@ async fn four_node_add_tier_committed_replay_converges() -> TestResult { let tier_name = unique_tier_name(); add_rustfs_tier(&hot, &cold, &tier_name).await?; - wait_for_tier_converged(&hot, &tier_name, "committed AddTier replay").await + wait_for_tier_converged(&hot, &tier_name, "AddTier convergence").await +} + +#[tokio::test] +#[serial] +async fn four_node_add_tier_converges_after_offline_node_restart_without_second_mutation() -> TestResult { + init_logging(); + + let mut cold = RustFSTestEnvironment::new().await?; + cold.access_key = "inlineofflinecoldadmin".to_string(); + cold.secret_key = "inlineofflinecoldsecret".to_string(); + cold.start_rustfs_server_without_cleanup(vec![]).await?; + cold.create_s3_client().create_bucket().bucket(TIER_BUCKET).send().await?; + + let mut hot = RustFSTestClusterEnvironment::new(4).await?; + hot.start().await?; + + let tier_name = unique_tier_name(); + hot.stop_node(3)?; + let add_tier_response = submit_rustfs_tier(&hot, &cold, &tier_name).await?; + hot.start_node(3).await?; + + wait_for_tier_converged(&hot, &tier_name, &add_tier_response).await } #[tokio::test] diff --git a/crates/ecstore/src/cluster/rpc/mod.rs b/crates/ecstore/src/cluster/rpc/mod.rs index cae151517..10e158275 100644 --- a/crates/ecstore/src/cluster/rpc/mod.rs +++ b/crates/ecstore/src/cluster/rpc/mod.rs @@ -37,6 +37,7 @@ pub use http_auth::{ #[cfg(test)] pub(crate) use internode_data_transport::TcpHttpInternodeDataTransport; pub use internode_data_transport::build_internode_data_transport_from_env; +pub(crate) use peer_rest_client::TierConfigReloadOutcome; pub use peer_rest_client::{ PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, PeerRestClient, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, ScannerPeerActivity, diff --git a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs index 7fcc1ae47..9be63f6a4 100644 --- a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs +++ b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs @@ -1631,29 +1631,121 @@ impl PeerRestClient { } pub async fn load_transition_tier_config(&self) -> Result<()> { - let result = self.load_transition_tier_config_inner().await; - if let Err(err) = &result - && Self::is_network_like_error(err) - { - self.prepare_retry().await; - return self.finalize_result(self.load_transition_tier_config_inner().await).await; + match self.load_transition_tier_config_outcome().await { + TierConfigReloadOutcome::Success => Ok(()), + TierConfigReloadOutcome::TransientReconnect(err) | TierConfigReloadOutcome::TransientRetrySameChannel(err) => { + self.finalize_result(Err(err)).await + } + TierConfigReloadOutcome::Terminal(err) => Err(err), } - self.finalize_result(result).await } - async fn load_transition_tier_config_inner(&self) -> Result<()> { - let mut client = self.get_client().await?; - let request = Request::new(LoadTransitionTierConfigRequest {}); + pub(crate) async fn load_transition_tier_config_outcome(&self) -> TierConfigReloadOutcome { + let outcome = self.load_transition_tier_config_single_attempt_outcome().await; + if outcome.is_transient() { + return self.load_transition_tier_config_once_outcome().await; + } + outcome + } - let response = client.load_transition_tier_config(request).await?.into_inner(); + pub(crate) async fn load_transition_tier_config_single_attempt_outcome(&self) -> TierConfigReloadOutcome { + let outcome = self.load_transition_tier_config_once_outcome().await; + if outcome.requires_reconnect() { + self.prepare_retry().await; + } + outcome + } + + pub(crate) async fn load_transition_tier_config_once_outcome(&self) -> TierConfigReloadOutcome { + let mut client = match self.get_client().await { + Ok(client) => client, + Err(err) => return tier_config_reload_connection_outcome(err), + }; + let mut request = Request::new(LoadTransitionTierConfigRequest {}); + request.set_timeout(rustfs_protos::heal_control_execution_timeout()); + + let response = match client.load_transition_tier_config(request).await { + Ok(response) => response.into_inner(), + Err(status) => return tier_config_reload_status_outcome(status), + }; if !response.success { - if let Some(msg) = response.error_info { - return Err(Error::other(msg)); - } - return Err(Error::other("")); + return tier_config_reload_remote_failure(response.error_info); } - Ok(()) + TierConfigReloadOutcome::Success + } +} + +pub(crate) enum TierConfigReloadOutcome { + Success, + TransientReconnect(Error), + TransientRetrySameChannel(Error), + Terminal(Error), +} + +impl TierConfigReloadOutcome { + fn is_transient(&self) -> bool { + matches!(self, Self::TransientReconnect(_) | Self::TransientRetrySameChannel(_)) + } + + fn requires_reconnect(&self) -> bool { + matches!(self, Self::TransientReconnect(_)) + } +} + +fn tier_config_reload_connection_outcome(err: Error) -> TierConfigReloadOutcome { + if is_tier_config_reload_connection_failure(&err) { + TierConfigReloadOutcome::TransientReconnect(err) + } else { + TierConfigReloadOutcome::Terminal(err) + } +} + +fn is_tier_config_reload_connection_failure(err: &Error) -> bool { + let message = err.to_string().to_ascii_lowercase(); + if message + .split_once("can not get client, err:") + .is_some_and(|(_, local_error)| local_error.contains("unavailable")) + { + return true; + } + [ + "temporarily offline", + "transport error", + "error trying to connect", + "connection refused", + "connection reset", + "connection closed", + "connection aborted", + "broken pipe", + "not connected", + "unexpected eof", + "timed out", + "deadline has elapsed", + "tcp connect error", + ] + .iter() + .any(|needle| message.contains(needle)) +} + +fn tier_config_reload_remote_failure(error_info: Option) -> TierConfigReloadOutcome { + let error_info = error_info.unwrap_or_default(); + if matches!(error_info.as_str(), "errServerNotInitialized" | "ServerNotInitialized") { + TierConfigReloadOutcome::TransientRetrySameChannel(Error::other(error_info)) + } else { + TierConfigReloadOutcome::Terminal(Error::other(error_info)) + } +} + +fn tier_config_reload_status_outcome(status: tonic::Status) -> TierConfigReloadOutcome { + use tonic::Code; + + if matches!(status.code(), Code::Unavailable | Code::DeadlineExceeded) { + TierConfigReloadOutcome::TransientReconnect(status.into()) + } else if status.code() == Code::Unknown && status.message().starts_with("Service was not ready:") { + TierConfigReloadOutcome::TransientRetrySameChannel(status.into()) + } else { + TierConfigReloadOutcome::Terminal(status.into()) } } @@ -2059,6 +2151,73 @@ mod tests { assert!(!PeerRestClient::is_network_like_error(&Error::NotImplemented)); } + #[test] + fn tier_config_reload_outcome_keeps_tonic_and_remote_errors_typed() { + assert!(matches!( + tier_config_reload_status_outcome(tonic::Status::unavailable("peer offline")), + TierConfigReloadOutcome::TransientReconnect(_) + )); + assert!(matches!( + tier_config_reload_status_outcome(tonic::Status::deadline_exceeded("peer timeout")), + TierConfigReloadOutcome::TransientReconnect(_) + )); + assert!(matches!( + tier_config_reload_status_outcome(tonic::Status::permission_denied("bad signature")), + TierConfigReloadOutcome::Terminal(_) + )); + assert!(matches!( + tier_config_reload_status_outcome(tonic::Status::unknown("Service was not ready: test client")), + TierConfigReloadOutcome::TransientRetrySameChannel(_) + )); + assert!(matches!( + tier_config_reload_status_outcome(tonic::Status::unknown("peer response unknown")), + TierConfigReloadOutcome::Terminal(_) + )); + assert!(matches!( + tier_config_reload_status_outcome(tonic::Status::cancelled("request cancelled")), + TierConfigReloadOutcome::Terminal(_) + )); + assert!(matches!( + tier_config_reload_remote_failure(Some("backend unavailable".to_string())), + TierConfigReloadOutcome::Terminal(_) + )); + assert!(matches!( + tier_config_reload_remote_failure(Some("errServerNotInitialized".to_string())), + TierConfigReloadOutcome::TransientRetrySameChannel(_) + )); + assert!(matches!( + tier_config_reload_connection_outcome(Error::other("backend unavailable")), + TierConfigReloadOutcome::Terminal(_) + )); + assert!(matches!( + tier_config_reload_connection_outcome(Error::other("can not get client, err: connection unavailable")), + TierConfigReloadOutcome::TransientReconnect(_) + )); + } + + #[tokio::test] + async fn tier_config_reload_single_attempt_clears_offline_gate_without_redial() { + let client = test_peer_client(); + client.offline.store(true, Ordering::Release); + + let outcome = client.load_transition_tier_config_single_attempt_outcome().await; + + assert!(matches!(outcome, TierConfigReloadOutcome::TransientReconnect(_))); + assert!(!client.offline.load(Ordering::Acquire)); + } + + #[test] + fn tier_config_reload_readiness_retry_does_not_require_reconnect() { + let client = test_peer_client(); + client.offline.store(true, Ordering::Release); + + let outcome = tier_config_reload_status_outcome(tonic::Status::unknown("Service was not ready: startup")); + + assert!(matches!(outcome, TierConfigReloadOutcome::TransientRetrySameChannel(_))); + assert!(!outcome.requires_reconnect()); + assert!(client.offline.load(Ordering::Acquire)); + } + #[tokio::test] async fn peer_rest_client_fast_fails_when_marked_offline() { let client = test_peer_client(); diff --git a/crates/ecstore/src/services/notification_sys.rs b/crates/ecstore/src/services/notification_sys.rs index 890e71a6f..13ef3e09a 100644 --- a/crates/ecstore/src/services/notification_sys.rs +++ b/crates/ecstore/src/services/notification_sys.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::cluster::rpc::{PeerRestClient, ScannerPeerActivity}; +use crate::cluster::rpc::{PeerRestClient, ScannerPeerActivity, TierConfigReloadOutcome}; use crate::diagnostics::admin_server_info::get_commit_id; use crate::disk::DiskAPI; use crate::error::{Error, Result}; @@ -34,7 +34,8 @@ use std::future::Future; use std::hash::{Hash, Hasher}; use std::sync::{Arc, Mutex, OnceLock}; use std::time::{Duration, SystemTime}; -use tokio::time::timeout; +use tokio::time::{sleep, timeout}; +use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, warn}; use uuid::Uuid; @@ -44,6 +45,8 @@ const LOG_COMPONENT_ECSTORE: &str = "ecstore"; const LOG_SUBSYSTEM_NOTIFICATION: &str = "notification"; const EVENT_NOTIFICATION_PEER_PROPAGATION: &str = "notification_peer_propagation"; const SCANNER_ACTIVITY_PROBE_TIMEOUT: Duration = Duration::from_secs(5); +const TIER_CONFIG_RELOAD_RETRY_BASE: Duration = Duration::from_millis(100); +const TIER_CONFIG_RELOAD_RETRY_CAP: Duration = Duration::from_secs(5); /// Cached result from the last successful admin call to a peer. struct PeerAdminCache { @@ -57,6 +60,16 @@ struct PeerAdminCache { last_server_success: Option, } +#[derive(Default)] +struct TierConfigReloadWorkers { + peers: HashMap, +} + +enum TierConfigReloadFinish { + Completed, + Pending, +} + impl PeerAdminCache { fn new() -> Self { Self { @@ -97,6 +110,7 @@ pub struct NotificationSys { pub all_peer_clients: Vec>, peer_topology_hosts: Vec, peer_admin_caches: Vec>, + tier_config_reload_workers: Arc>, } impl NotificationSys { @@ -108,6 +122,7 @@ impl NotificationSys { all_peer_clients, peer_topology_hosts, peer_admin_caches, + tier_config_reload_workers: Default::default(), } } } @@ -1127,6 +1142,124 @@ impl NotificationSys { join_all(futures).await } + /// Starts one immediate configuration reload worker per peer. Concurrent + /// tier mutations share the existing worker for that peer. + pub fn spawn_transition_tier_config_reload_workers(self: &Arc) { + self.spawn_transition_tier_config_reload_workers_with_cancel_token(runtime_sources::background_services_cancel_token()); + } + + fn spawn_transition_tier_config_reload_workers_with_cancel_token(self: &Arc, cancel_token: Option) { + let Some(cancel_token) = cancel_token else { + warn!( + event = EVENT_NOTIFICATION_PEER_PROPAGATION, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_NOTIFICATION, + action = "reload_transition_tier_config", + result = "background_service_unavailable", + "notification peer propagation" + ); + return; + }; + for (peer_index, client) in self.peer_clients.iter().enumerate() { + let Some(client) = client.clone() else { + warn!( + event = EVENT_NOTIFICATION_PEER_PROPAGATION, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_NOTIFICATION, + action = "reload_transition_tier_config", + peer_index, + result = "peer_unreachable", + "notification peer propagation" + ); + continue; + }; + let host = client.grid_host.clone(); + if !self.reserve_tier_config_reload_worker(&host) { + debug!( + event = EVENT_NOTIFICATION_PEER_PROPAGATION, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_NOTIFICATION, + action = "reload_transition_tier_config", + host, + result = "coalesced", + "notification peer propagation" + ); + continue; + } + let sys = Arc::clone(self); + let cancel_token = cancel_token.clone(); + tokio::spawn(async move { + run_tier_config_reload_worker(sys, host, cancel_token, move || { + let client = client.clone(); + async move { client.load_transition_tier_config_single_attempt_outcome().await } + }) + .await; + }); + } + } + + fn reserve_tier_config_reload_worker(&self, host: &str) -> bool { + let mut workers = self + .tier_config_reload_workers + .lock() + .expect("tier config reload worker state must not be poisoned"); + match workers.peers.get_mut(host) { + Some(pending) => { + *pending = true; + false + } + None => { + workers.peers.insert(host.to_string(), false); + true + } + } + } + + fn take_tier_config_reload_pending(&self, host: &str) -> bool { + let mut workers = self + .tier_config_reload_workers + .lock() + .expect("tier config reload worker state must not be poisoned"); + let Some(pending) = workers.peers.get_mut(host) else { + return false; + }; + let pending_reload = *pending; + *pending = false; + pending_reload + } + + fn finish_tier_config_reload_worker(&self, host: &str) -> TierConfigReloadFinish { + let mut workers = self + .tier_config_reload_workers + .lock() + .expect("tier config reload worker state must not be poisoned"); + let Some(pending) = workers.peers.get_mut(host) else { + return TierConfigReloadFinish::Completed; + }; + if *pending { + *pending = false; + return TierConfigReloadFinish::Pending; + } + workers.peers.remove(host); + TierConfigReloadFinish::Completed + } + + fn cancel_tier_config_reload_worker(&self, host: &str) { + let mut workers = self + .tier_config_reload_workers + .lock() + .expect("tier config reload worker state must not be poisoned"); + workers.peers.remove(host); + } + + fn tier_config_reload_worker_active(&self, host: &str) -> bool { + self.tier_config_reload_workers + .lock() + .expect("tier config reload worker state must not be poisoned") + .peers + .contains_key(host) + } + pub async fn prepare_tier_mutation(&self, mutation_id: Uuid, canonical_payload: Bytes) -> Vec { let mut futures = Vec::with_capacity(self.peer_clients.len()); for client in self.peer_clients.iter().cloned() { @@ -1172,6 +1305,113 @@ impl NotificationSys { } } +async fn run_tier_config_reload_worker( + sys: Arc, + host: String, + cancel_token: CancellationToken, + mut reload: F, +) where + F: FnMut() -> Fut, + Fut: Future, +{ + let mut retry_attempt = 0; + loop { + if cancel_token.is_cancelled() { + sys.cancel_tier_config_reload_worker(&host); + return; + } + let result = tokio::select! { + _ = cancel_token.cancelled() => { + sys.cancel_tier_config_reload_worker(&host); + return; + } + result = reload() => result, + }; + + match result { + TierConfigReloadOutcome::Success => match sys.finish_tier_config_reload_worker(&host) { + TierConfigReloadFinish::Completed => { + debug!( + event = EVENT_NOTIFICATION_PEER_PROPAGATION, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_NOTIFICATION, + action = "reload_transition_tier_config", + host, + result = "success", + "notification peer propagation" + ); + return; + } + TierConfigReloadFinish::Pending => retry_attempt = 0, + }, + TierConfigReloadOutcome::Terminal(_) => match sys.finish_tier_config_reload_worker(&host) { + TierConfigReloadFinish::Completed => { + warn!( + event = EVENT_NOTIFICATION_PEER_PROPAGATION, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_NOTIFICATION, + action = "reload_transition_tier_config", + host, + outcome = "terminal", + "tier configuration reload stopped after a terminal outcome" + ); + return; + } + TierConfigReloadFinish::Pending => retry_attempt = 0, + }, + TierConfigReloadOutcome::TransientReconnect(_) | TierConfigReloadOutcome::TransientRetrySameChannel(_) => { + let delay = tier_config_reload_retry_delay(retry_attempt); + retry_attempt = retry_attempt.saturating_add(1); + if sys.take_tier_config_reload_pending(&host) { + retry_attempt = 0; + continue; + } + if retry_attempt == 1 { + warn!( + event = EVENT_NOTIFICATION_PEER_PROPAGATION, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_NOTIFICATION, + action = "reload_transition_tier_config", + host, + retry_attempt, + retry_delay_ms = delay.as_millis(), + outcome = "transient", + "tier configuration reload failed; retrying" + ); + } else if retry_attempt.is_power_of_two() { + debug!( + event = EVENT_NOTIFICATION_PEER_PROPAGATION, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_NOTIFICATION, + action = "reload_transition_tier_config", + host, + retry_attempt, + retry_delay_ms = delay.as_millis(), + outcome = "transient", + "tier configuration reload retry failed" + ); + } + + tokio::select! { + _ = cancel_token.cancelled() => { + sys.cancel_tier_config_reload_worker(&host); + return; + } + _ = sleep(delay) => {} + } + } + } + } +} + +fn tier_config_reload_retry_delay(retry_attempt: u32) -> Duration { + let multiplier = 1_u32 << retry_attempt.min(6); + TIER_CONFIG_RELOAD_RETRY_BASE + .checked_mul(multiplier) + .unwrap_or(TIER_CONFIG_RELOAD_RETRY_CAP) + .min(TIER_CONFIG_RELOAD_RETRY_CAP) +} + async fn scanner_activity_with_timeout(timeout_duration: Duration, host: &str, activity: F) -> Result where F: Future>, @@ -1722,6 +1962,7 @@ mod tests { ))], peer_topology_hosts: Vec::new(), peer_admin_caches: Vec::new(), + tier_config_reload_workers: Default::default(), }; let client = sys @@ -1766,6 +2007,7 @@ mod tests { all_peer_clients: Vec::new(), peer_topology_hosts: vec!["node-a:9000".to_string()], peer_admin_caches: vec![Mutex::new(PeerAdminCache::new())], + tier_config_reload_workers: Default::default(), }; let err = sys @@ -1786,6 +2028,7 @@ mod tests { all_peer_clients: vec![None, None], peer_topology_hosts: vec!["node-a:9000".to_string()], peer_admin_caches: vec![Mutex::new(PeerAdminCache::new())], + tier_config_reload_workers: Default::default(), }; let err = sys @@ -1803,6 +2046,7 @@ mod tests { all_peer_clients: Vec::new(), peer_topology_hosts: Vec::new(), peer_admin_caches: Vec::new(), + tier_config_reload_workers: Default::default(), }; let err = sys @@ -1824,6 +2068,7 @@ mod tests { all_peer_clients: vec![None], peer_topology_hosts: vec!["127.0.0.1:9000".to_string()], peer_admin_caches: vec![Mutex::new(PeerAdminCache::new())], + tier_config_reload_workers: Default::default(), }; let err = sys @@ -1841,6 +2086,7 @@ mod tests { all_peer_clients: vec![None, None], peer_topology_hosts: vec!["node-a:9000".to_string()], peer_admin_caches: vec![Mutex::new(PeerAdminCache::new())], + tier_config_reload_workers: Default::default(), }; let servers = sys.server_info().await; @@ -1899,6 +2145,7 @@ mod tests { peer_clients: Vec::new(), all_peer_clients: Vec::new(), peer_admin_caches: Vec::new(), + tier_config_reload_workers: Default::default(), peer_topology_hosts: Vec::new(), }; let missing = sys @@ -1972,6 +2219,7 @@ mod tests { all_peer_clients: Vec::new(), peer_topology_hosts: vec!["node-a:9000".to_string()], peer_admin_caches: vec![Mutex::new(PeerAdminCache::new())], + tier_config_reload_workers: Default::default(), }; let err = sys @@ -1985,6 +2233,205 @@ mod tests { assert!(msg.contains("peer[0]")); } + #[test] + fn tier_config_reload_retry_delay_is_exponentially_capped() { + assert_eq!(tier_config_reload_retry_delay(0), Duration::from_millis(100)); + assert_eq!(tier_config_reload_retry_delay(1), Duration::from_millis(200)); + assert_eq!(tier_config_reload_retry_delay(5), Duration::from_millis(3200)); + assert_eq!(tier_config_reload_retry_delay(6), TIER_CONFIG_RELOAD_RETRY_CAP); + assert_eq!(tier_config_reload_retry_delay(u32::MAX), TIER_CONFIG_RELOAD_RETRY_CAP); + } + + #[tokio::test] + async fn tier_config_reload_worker_retries_only_network_failures() { + let sys = Arc::new(NotificationSys { + peer_clients: Vec::new(), + all_peer_clients: Vec::new(), + peer_topology_hosts: Vec::new(), + peer_admin_caches: Vec::new(), + tier_config_reload_workers: Default::default(), + }); + assert!(sys.reserve_tier_config_reload_worker("node-a:9000")); + let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let calls_for_reload = Arc::clone(&calls); + + run_tier_config_reload_worker(Arc::clone(&sys), "node-a:9000".to_string(), CancellationToken::new(), move || { + let attempt = calls_for_reload.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + async move { + if attempt == 0 { + TierConfigReloadOutcome::TransientReconnect(Error::other("connection refused")) + } else { + TierConfigReloadOutcome::Success + } + } + }) + .await; + + assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 2); + assert!(!sys.tier_config_reload_worker_active("node-a:9000")); + } + + #[tokio::test] + async fn tier_config_reload_worker_converges_after_readiness_unknown() { + let sys = Arc::new(NotificationSys { + peer_clients: Vec::new(), + all_peer_clients: Vec::new(), + peer_topology_hosts: Vec::new(), + peer_admin_caches: Vec::new(), + tier_config_reload_workers: Default::default(), + }); + assert!(sys.reserve_tier_config_reload_worker("node-a:9000")); + let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let calls_for_reload = Arc::clone(&calls); + + run_tier_config_reload_worker(Arc::clone(&sys), "node-a:9000".to_string(), CancellationToken::new(), move || { + let attempt = calls_for_reload.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + async move { + if attempt == 0 { + TierConfigReloadOutcome::TransientRetrySameChannel(Error::other("Service was not ready: test client")) + } else { + TierConfigReloadOutcome::Success + } + } + }) + .await; + + assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 2); + assert!(!sys.tier_config_reload_worker_active("node-a:9000")); + } + + #[tokio::test] + async fn tier_config_reload_worker_stops_on_terminal_failure() { + let sys = Arc::new(NotificationSys { + peer_clients: Vec::new(), + all_peer_clients: Vec::new(), + peer_topology_hosts: Vec::new(), + peer_admin_caches: Vec::new(), + tier_config_reload_workers: Default::default(), + }); + assert!(sys.reserve_tier_config_reload_worker("node-a:9000")); + let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let calls_for_reload = Arc::clone(&calls); + + run_tier_config_reload_worker(Arc::clone(&sys), "node-a:9000".to_string(), CancellationToken::new(), move || { + calls_for_reload.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + async { TierConfigReloadOutcome::Terminal(Error::NotImplemented) } + }) + .await; + + assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 1); + assert!(!sys.tier_config_reload_worker_active("node-a:9000")); + } + + #[tokio::test] + async fn tier_config_reload_worker_reloads_once_after_success_with_pending_mutation() { + let sys = Arc::new(NotificationSys { + peer_clients: Vec::new(), + all_peer_clients: Vec::new(), + peer_topology_hosts: Vec::new(), + peer_admin_caches: Vec::new(), + tier_config_reload_workers: Default::default(), + }); + assert!(sys.reserve_tier_config_reload_worker("node-a:9000")); + let sys_for_reload = Arc::clone(&sys); + let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let calls_for_reload = Arc::clone(&calls); + + run_tier_config_reload_worker(Arc::clone(&sys), "node-a:9000".to_string(), CancellationToken::new(), move || { + let attempt = calls_for_reload.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + let sys = Arc::clone(&sys_for_reload); + async move { + if attempt == 0 { + assert!(!sys.reserve_tier_config_reload_worker("node-a:9000")); + TierConfigReloadOutcome::Success + } else { + TierConfigReloadOutcome::Success + } + } + }) + .await; + + assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 2); + assert!(!sys.tier_config_reload_worker_active("node-a:9000")); + } + + #[test] + fn tier_config_reload_none_peer_does_not_start_a_worker() { + let sys = Arc::new(NotificationSys { + peer_clients: vec![None], + all_peer_clients: Vec::new(), + peer_topology_hosts: vec!["node-a:9000".to_string()], + peer_admin_caches: vec![Mutex::new(PeerAdminCache::new())], + tier_config_reload_workers: Default::default(), + }); + + sys.spawn_transition_tier_config_reload_workers_with_cancel_token(Some(CancellationToken::new())); + + assert!( + sys.tier_config_reload_workers + .lock() + .expect("tier config reload worker state must not be poisoned") + .peers + .is_empty() + ); + } + + #[test] + fn tier_config_reload_without_background_token_does_not_reserve_a_worker() { + let client = PeerRestClient::new( + "127.0.0.1:9000".to_string().try_into().expect("peer host should parse"), + "http://127.0.0.1:9000".to_string(), + ); + let sys = Arc::new(NotificationSys { + peer_clients: vec![Some(client)], + all_peer_clients: Vec::new(), + peer_topology_hosts: vec!["127.0.0.1:9000".to_string()], + peer_admin_caches: vec![Mutex::new(PeerAdminCache::new())], + tier_config_reload_workers: Default::default(), + }); + + sys.spawn_transition_tier_config_reload_workers_with_cancel_token(None); + + assert!( + sys.tier_config_reload_workers + .lock() + .expect("tier config reload worker state must not be poisoned") + .peers + .is_empty() + ); + } + + #[tokio::test] + async fn tier_config_reload_cancellation_during_transient_backoff_releases_state() { + let sys = Arc::new(NotificationSys { + peer_clients: Vec::new(), + all_peer_clients: Vec::new(), + peer_topology_hosts: Vec::new(), + peer_admin_caches: Vec::new(), + tier_config_reload_workers: Default::default(), + }); + assert!(sys.reserve_tier_config_reload_worker("node-a:9000")); + let cancel_token = CancellationToken::new(); + let cancel_for_reload = cancel_token.clone(); + let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let calls_for_reload = Arc::clone(&calls); + + run_tier_config_reload_worker(Arc::clone(&sys), "node-a:9000".to_string(), cancel_token, move || { + let attempt = calls_for_reload.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + let cancel_token = cancel_for_reload.clone(); + async move { + if attempt == 0 { + cancel_token.cancel(); + } + TierConfigReloadOutcome::TransientReconnect(Error::other("connection refused")) + } + }) + .await; + + assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 1); + assert!(!sys.tier_config_reload_worker_active("node-a:9000")); + } + #[tokio::test] async fn load_transition_tier_config_reports_unreachable_peers() { let sys = NotificationSys { @@ -1992,6 +2439,7 @@ mod tests { all_peer_clients: Vec::new(), peer_topology_hosts: vec!["node-a:9000".to_string()], peer_admin_caches: vec![Mutex::new(PeerAdminCache::new())], + tier_config_reload_workers: Default::default(), }; let results = sys.load_transition_tier_config().await; @@ -2008,6 +2456,7 @@ mod tests { all_peer_clients: Vec::new(), peer_topology_hosts: vec!["node-a:9000".to_string()], peer_admin_caches: vec![Mutex::new(PeerAdminCache::new())], + tier_config_reload_workers: Default::default(), }; let mutation_id = Uuid::from_u128(1); diff --git a/rustfs/src/admin/handlers/tier.rs b/rustfs/src/admin/handlers/tier.rs index 641fee008..72443d051 100644 --- a/rustfs/src/admin/handlers/tier.rs +++ b/rustfs/src/admin/handlers/tier.rs @@ -28,7 +28,6 @@ use crate::{ }, auth::{check_key_valid, get_session_token}, server::{ADMIN_PREFIX, RemoteAddr}, - storage::request_context::spawn_traced, }; use http::{HeaderMap, StatusCode, Uri}; use hyper::Method; @@ -89,22 +88,15 @@ fn wasabi_payload_name(config: &TierConfig) -> S3Result { fn spawn_transition_tier_config_propagation(action: &'static str) { if let Some(notification_sys) = current_notification_system() { - spawn_traced(async move { - for peer_result in notification_sys.load_transition_tier_config().await { - if let Some(err) = peer_result.err { - warn!( - event = EVENT_ADMIN_TIER_STATE, - component = LOG_COMPONENT_ADMIN, - subsystem = LOG_SUBSYSTEM_TIER, - action = action, - host = if peer_result.host.is_empty() { "" } else { peer_result.host.as_str() }, - result = "propagation_failed", - error = %err, - "admin tier state" - ); - } - } - }); + debug!( + event = EVENT_ADMIN_TIER_STATE, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_TIER, + action, + result = "propagation_started", + "admin tier state" + ); + notification_sys.spawn_transition_tier_config_reload_workers(); } }