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 <heihutu@gmail.com>

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<String> {
     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]

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,

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<String>) -> 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();

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<SystemTime>,
 }
 
+#[derive(Default)]
+struct TierConfigReloadWorkers {
+    peers: HashMap<String, bool>,
+}
+
+enum TierConfigReloadFinish {
+    Completed,
+    Pending,
+}
+
 impl PeerAdminCache {
     fn new() -> Self {
         Self {
@@ -97,6 +110,7 @@ pub struct NotificationSys {
     pub all_peer_clients: Vec<Option<PeerRestClient>>,
     peer_topology_hosts: Vec<String>,
     peer_admin_caches: Vec<Mutex<PeerAdminCache>>,
+    tier_config_reload_workers: Arc<Mutex<TierConfigReloadWorkers>>,
 }
 
 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>) {
+        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<Self>, cancel_token: Option<CancellationToken>) {
+        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<NotificationPeerErr> {
         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<F, Fut>(
+    sys: Arc<NotificationSys>,
+    host: String,
+    cancel_token: CancellationToken,
+    mut reload: F,
+) where
+    F: FnMut() -> Fut,
+    Fut: Future<Output = TierConfigReloadOutcome>,
+{
+    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<F>(timeout_duration: Duration, host: &str, activity: F) -> Result<ScannerPeerActivity>
 where
     F: Future<Output = Result<ScannerPeerActivity>>,
@@ -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);
 

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<String> {
 
 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() { "<unknown>" } 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();
     }
 }