From dfe0b957239bf8faf94df904d363db6075595783 Mon Sep 17 00:00:00 2001 From: cxymds Date: Fri, 4 Sep 2026 21:58:02 +0800 Subject: [PATCH] fix(site-replication): wake retry drain after peer recovery --- rustfs/src/admin/handlers/site_replication.rs | 57 +++++++++--- rustfs/src/site_replication/retry.rs | 29 +++++-- rustfs/src/site_replication/tests.rs | 87 +++++++++++++++++++ rustfs/src/site_replication_reconcile.rs | 48 +++++++++- 4 files changed, 200 insertions(+), 21 deletions(-) diff --git a/rustfs/src/admin/handlers/site_replication.rs b/rustfs/src/admin/handlers/site_replication.rs index d75e3b850..8ebf3c02a 100644 --- a/rustfs/src/admin/handlers/site_replication.rs +++ b/rustfs/src/admin/handlers/site_replication.rs @@ -434,6 +434,7 @@ pub fn register_site_replication_route(r: &mut S3Router) -> std: // into this module: startup sits below this layer and must not depend upwards. The admin // router is built before startup reconciles, so the hook is always installed in time. crate::site_replication_reconcile::register_site_replication_reconciler(reconcile_site_replication_wiring); + crate::site_replication_reconcile::register_site_replication_retry_drainer(reconcile_site_replication_retry_drain); for (method, path, operation) in [ (Method::PUT, "/v3/site-replication/add", AdminOperation(&SiteReplicationAddHandler {})), @@ -1803,28 +1804,55 @@ async fn reconcile_site_replication_buckets() -> S3Result<()> { /// (`SiteReplicationEditHandler`), so a tick landing between them would rewrite the targets /// from the stale endpoint. The pending marker in the persisted state closes that window. /// Skipping costs nothing — the timer comes back. +async fn site_replication_reconcile_prerequisites_ready() -> bool { + if current_iam_handle().is_none() || current_object_store_handle().is_none() { + return false; + } + if let Err(err) = migrate_collapsed_retry_queue_paths().await { + warn!( + event = EVENT_ADMIN_SITE_REPLICATION_STATE, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SITE_REPLICATION, + result = "retry_queue_migration_failed", + error = ?err, + "admin site replication state" + ); + return false; + } + true +} + +fn reconcile_site_replication_retry_drain() -> std::pin::Pin + Send>> { + Box::pin(async { + let Some(_lifecycle) = SiteReplicationLifecycleGuard::try_acquire() else { + return; + }; + if !site_replication_reconcile_prerequisites_ready().await { + return; + } + match load_site_replication_state().await { + Ok(state) => { + if state.pending_endpoint_refresh.is_some() || state.pending_rotation.is_some() || state.pending_remove.is_some() + { + return; + } + } + Err(_) => return, + } + drain_site_replication_retry_queue().await; + }) +} + fn reconcile_site_replication_wiring() -> std::pin::Pin + Send>> { Box::pin(async { // The scheduler starts before IAM and the object store are guaranteed ready (IAM // bootstrap may still be recovering), so an early tick returns quietly instead of // logging a failure for every reconciler. - if current_iam_handle().is_none() || current_object_store_handle().is_none() { - return; - } - let Some(_lifecycle) = SiteReplicationLifecycleGuard::try_acquire() else { return; }; - if let Err(err) = migrate_collapsed_retry_queue_paths().await { - warn!( - event = EVENT_ADMIN_SITE_REPLICATION_STATE, - component = LOG_COMPONENT_ADMIN, - subsystem = LOG_SUBSYSTEM_SITE_REPLICATION, - result = "retry_queue_migration_failed", - error = ?err, - "admin site replication state" - ); + if !site_replication_reconcile_prerequisites_ready().await { return; } @@ -3046,6 +3074,7 @@ fn set_pending_endpoint_refresh(state: &mut SiteReplicationState, pending: Pendi last_error: "endpoint target refresh pending".to_string(), updated_at: Some(OffsetDateTime::now_utc()), edit_generation: None, + peer_unreachable: false, deletions_recorded: false, }); state.pending_endpoint_refresh = Some(pending); @@ -12791,6 +12820,7 @@ mod tests { last_error: "site replication is not enabled".to_string(), updated_at: Some(OffsetDateTime::now_utc()), edit_generation: None, + peer_unreachable: false, deletions_recorded: false, }], ..Default::default() @@ -12989,6 +13019,7 @@ mod tests { last_error: "peer offline".to_string(), updated_at: Some(OffsetDateTime::now_utc()), edit_generation: None, + peer_unreachable: false, deletions_recorded: false, }], ..Default::default() diff --git a/rustfs/src/site_replication/retry.rs b/rustfs/src/site_replication/retry.rs index 818fbee40..00c1b2b67 100644 --- a/rustfs/src/site_replication/retry.rs +++ b/rustfs/src/site_replication/retry.rs @@ -41,6 +41,12 @@ pub(crate) struct SiteReplicationRetryEvent { /// [`settle_site_replication_retry_events`]. #[serde(default, skip_serializing_if = "Option::is_none")] pub(crate) edit_generation: Option, + /// The latest delivery failure happened before an authenticated peer + /// response was received (connect, timeout, DNS, or TLS). Such failures + /// may bypass the expensive replay backoff only after a cheap devnull + /// reachability probe proves the peer is back. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub(crate) peer_unreachable: bool, /// Whether every failure folded into this collapsed IAM entry had its /// deletion body (if it was a deletion) recorded in /// [`SiteReplicationState::iam_deletion_replays`]. Only then may a @@ -206,11 +212,13 @@ pub(crate) fn upsert_site_replication_retry_event( let path = collapsed_retry_queue_path(path).unwrap_or(path); let now = OffsetDateTime::now_utc(); let detail = summarize_peer_error_detail(error); + let peer_unreachable = retry_error_indicates_peer_unreachable(error); if let Some(event) = queue.iter_mut().find(|event| retry_event_matches(event, peer, path)) { event.retry_count = event.retry_count.saturating_add(1); event.failed = event.retry_count >= SITE_REPLICATION_RETRY_FAILED_AFTER; event.last_error = detail; event.updated_at = Some(now); + event.peer_unreachable = peer_unreachable; // Keep the newest generation: an older delivery that fails afterwards // must not lower the fence and let its own success settle the event. event.edit_generation = event.edit_generation.max(generation); @@ -227,6 +235,7 @@ pub(crate) fn upsert_site_replication_retry_event( last_error: detail, updated_at: Some(now), edit_generation: generation, + peer_unreachable, deletions_recorded: false, }); if queue.len() > SITE_REPLICATION_RETRY_QUEUE_LIMIT { @@ -235,6 +244,14 @@ pub(crate) fn upsert_site_replication_retry_event( } } +pub(crate) fn retry_error_indicates_peer_unreachable(error: &str) -> bool { + let error = error.to_ascii_lowercase(); + error.contains("failed (connect)") + || error.contains("failed (timeout)") + || error.contains("failed (dns resolution)") + || error.contains("failed (tls handshake)") +} + pub(crate) fn retry_stats_for_state(state: &SiteReplicationState) -> Option { if state.retry_queue.is_empty() { return None; @@ -989,10 +1006,10 @@ pub(crate) fn actionable_site_replication_retry_events( /// backoff exists to spare a *dead* peer the expensive replay (plan build, /// snapshot resend) — it must not delay convergence to a peer that has /// already RECOVERED, or a failure window ends in up to a day of silent -/// divergence (backlog#2071). The drain probes each such peer with one cheap -/// request per tick and promotes its backlog when the probe succeeds. The -/// base backoff still floors individual re-attempts so a reachable peer that -/// keeps failing a delivery is not hammered faster than before. +/// divergence (backlog#2071). Transport failures may be probed before the +/// normal replay backoff elapses; application failures still wait at least +/// one base interval so a reachable peer that keeps rejecting a replay is not +/// hammered faster than before. pub(crate) fn deferred_site_replication_retry_events( state: &SiteReplicationState, now: OffsetDateTime, @@ -1005,7 +1022,9 @@ pub(crate) fn deferred_site_replication_retry_events( .filter(|event| !site_replication_retry_backoff_elapsed(event, now)) .filter(|event| { event.updated_at.is_none_or(|updated_at| { - now.unix_timestamp().saturating_sub(updated_at.unix_timestamp()) >= SITE_REPLICATION_RETRY_DRAIN_BASE_BACKOFF_SECS + event.peer_unreachable + || now.unix_timestamp().saturating_sub(updated_at.unix_timestamp()) + >= SITE_REPLICATION_RETRY_DRAIN_BASE_BACKOFF_SECS }) }) .cloned() diff --git a/rustfs/src/site_replication/tests.rs b/rustfs/src/site_replication/tests.rs index e37869a43..bc14ea72f 100644 --- a/rustfs/src/site_replication/tests.rs +++ b/rustfs/src/site_replication/tests.rs @@ -419,6 +419,7 @@ fn drain_event(peer: &str, path: &str, retry_count: u32, updated_at: Option Pin + Send>>; static RECONCILER: OnceLock = OnceLock::new(); +static RETRY_DRAINER: OnceLock = OnceLock::new(); /// Install the admin layer's reconciler. Idempotent: a second call is ignored, which keeps /// repeated router construction (tests, the embedded server) from panicking. @@ -45,6 +48,12 @@ pub(crate) fn register_site_replication_reconciler(reconcile: ReconcileHook) { let _ = RECONCILER.set(reconcile); } +/// Install the admin layer's lightweight retry drain. Idempotent for the same +/// reason as [`register_site_replication_reconciler`]. +pub(crate) fn register_site_replication_retry_drainer(drain: ReconcileHook) { + let _ = RETRY_DRAINER.set(drain); +} + /// Repair drifted site-replication wiring, immediately and then on a timer. /// /// The first pass runs inside the spawned task rather than on the caller's path: it walks @@ -62,16 +71,38 @@ pub(crate) fn spawn_site_replication_reconcile_task(ctx: CancellationToken) { return; } + spawn_reconcile_loop(ctx.clone(), RECONCILE_INTERVAL, &RECONCILER, true); + + if RETRY_DRAINER.get().is_none() { + warn!("site replication retry drainer is not registered; periodic retry drain disabled"); + return; + } + spawn_reconcile_loop(ctx, RETRY_DRAIN_INTERVAL, &RETRY_DRAINER, false); +} + +fn spawn_reconcile_loop( + ctx: CancellationToken, + interval: Duration, + hook: &'static OnceLock, + run_immediately: bool, +) { tokio::spawn(async move { - let mut ticker = tokio::time::interval(RECONCILE_INTERVAL); + let first_tick = if run_immediately { + Instant::now() + } else { + Instant::now() + interval + }; + let mut ticker = tokio::time::interval_at(first_tick, interval); ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); loop { tokio::select! { _ = ctx.cancelled() => break, - // The first tick fires immediately, which is the startup repair pass. + // The heavy reconciler owns the startup repair pass. The lightweight + // retry drain starts on its normal cadence so it cannot steal that + // first lifecycle lock and defer bucket/IAM repair for a full interval. _ = ticker.tick() => { - if let Some(reconcile) = RECONCILER.get() { + if let Some(reconcile) = hook.get() { reconcile().await; } } @@ -79,3 +110,14 @@ pub(crate) fn spawn_site_replication_reconcile_task(ctx: CancellationToken) { } }); } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn retry_drain_runs_faster_than_heavy_reconcile() { + assert!(RETRY_DRAIN_INTERVAL < RECONCILE_INTERVAL); + assert!(RETRY_DRAIN_INTERVAL <= Duration::from_secs(60)); + } +}