fix(site-replication): wake retry drain after peer recovery

rustfs/src/admin/handlers/site_replication.rs

@@ -434,6 +434,7 @@ pub fn register_site_replication_route(r: &mut S3Router<AdminOperation>) -> 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<Box<dyn std::future::Future<Output = ()> + 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<Box<dyn std::future::Future<Output = ()> + 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()

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<u64>,
+    /// 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<SRRetryStats> {
     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()

rustfs/src/site_replication/tests.rs

@@ -419,6 +419,7 @@ fn drain_event(peer: &str, path: &str, retry_count: u32, updated_at: Option<Offs
         last_error: "remote-operation-failed".to_string(),
         updated_at,
         edit_generation: None,
+        peer_unreachable: false,
         deletions_recorded: false,
     }
 }
@@ -828,6 +829,47 @@ fn test_site_replication_retry_backoff_schedule() {
     assert!(elapsed(30, 86_401));
 }
 
+#[test]
+fn test_retry_error_marks_peer_unreachable_only_for_transport_failures() {
+    let mut queue = Vec::new();
+    let peer = peer("remote", "https://remote.example.com");
+    let bucket_make = "/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=make-with-versioning";
+
+    upsert_site_replication_retry_event(
+        &mut queue,
+        &peer,
+        bucket_make,
+        "peer request to https://remote.example.com failed (connect): connection refused",
+        None,
+    );
+    assert!(queue[0].peer_unreachable);
+
+    upsert_site_replication_retry_event(
+        &mut queue,
+        &peer,
+        bucket_make,
+        "peer request to https://remote.example.com failed with 500 Internal Server Error",
+        None,
+    );
+    assert!(!queue[0].peer_unreachable, "application failures must keep the normal replay backoff");
+}
+
+#[test]
+fn test_retry_event_peer_unreachable_is_legacy_serde_default() {
+    let json = r#"{
+        "id":"evt-legacy",
+        "peer_deployment_id":"remote",
+        "peer_endpoint":"https://remote.example.com",
+        "path":"/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=make-with-versioning",
+        "retry_count":1,
+        "failed":false,
+        "last_error":"legacy"
+    }"#;
+
+    let event: SiteReplicationRetryEvent = serde_json::from_str(json).expect("legacy retry event decodes");
+    assert!(!event.peer_unreachable);
+}
+
 /// The actionable subset respects classification, peer membership and
 /// backoff; everything else stays untouched in the queue.
 #[test]
@@ -915,6 +957,51 @@ fn test_deferred_site_replication_retry_events_partition() {
     assert_eq!(actionable[0].path, "/rustfs/admin/v3/site-replication/peer/bucket-meta");
 }
 
+#[test]
+fn test_deferred_retry_events_probe_fresh_peer_transport_failures() {
+    let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp");
+    let mut state = SiteReplicationState::default();
+    state
+        .peers
+        .insert("remote".to_string(), peer("remote", "https://remote.example.com"));
+
+    let bucket_make = "/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=make-with-versioning";
+    let mut fresh_transport_failure = drain_event("remote", bucket_make, 1, Some(now - time::Duration::seconds(30)));
+    fresh_transport_failure.peer_unreachable = true;
+    state.retry_queue.push(fresh_transport_failure);
+
+    let deferred = deferred_site_replication_retry_events(&state, now);
+    assert_eq!(
+        deferred.len(),
+        1,
+        "fresh transport failures must be eligible for a cheap reachability probe"
+    );
+    assert_eq!(deferred[0].path, bucket_make);
+
+    let actionable = actionable_site_replication_retry_events(&state, now);
+    assert!(actionable.is_empty(), "the event is still protected from direct replay by normal backoff");
+}
+
+#[test]
+fn test_deferred_retry_events_do_not_probe_fresh_application_failures() {
+    let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp");
+    let mut state = SiteReplicationState::default();
+    state
+        .peers
+        .insert("remote".to_string(), peer("remote", "https://remote.example.com"));
+
+    let bucket_make = "/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=make-with-versioning";
+    state
+        .retry_queue
+        .push(drain_event("remote", bucket_make, 1, Some(now - time::Duration::seconds(30))));
+
+    assert!(
+        deferred_site_replication_retry_events(&state, now).is_empty(),
+        "reachable peers that reject an operation must keep the base replay backoff"
+    );
+    assert!(actionable_site_replication_retry_events(&state, now).is_empty());
+}
+
 /// The drain settles a peer-edit success under a freshly allocated
 /// generation; legacy queue entries carry `edit_generation: None` and
 /// must be cleared by that generation-scoped settlement (`(Some, None)`

rustfs/src/site_replication_reconcile.rs

@@ -28,16 +28,19 @@ use std::pin::Pin;
 use std::sync::OnceLock;
 use std::time::Duration;
 
+use tokio::time::Instant;
 use tokio_util::sync::CancellationToken;
 use tracing::warn;
 
 const RECONCILE_INTERVAL: Duration = Duration::from_secs(600);
+const RETRY_DRAIN_INTERVAL: Duration = Duration::from_secs(30);
 
 /// A reconciler reports its own failures; the outcome carries no value because neither
 /// caller can act on one — a site that cannot repair its replication wiring still serves S3.
 type ReconcileHook = fn() -> Pin<Box<dyn Future<Output = ()> + Send>>;
 
 static RECONCILER: OnceLock<ReconcileHook> = OnceLock::new();
+static RETRY_DRAINER: OnceLock<ReconcileHook> = 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<ReconcileHook>,
+    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));
+    }
+}