fix(site-replication): keep recovery replay live

rustfs/src/admin/handlers/site_replication.rs

@@ -9569,10 +9569,6 @@ mod tests {
     fn test_peer_timeout_constants_bound_unreachable_peer_probes() {
         assert_eq!(SITE_REPLICATION_PEER_REQUEST_TIMEOUT, Duration::from_secs(10));
         assert_eq!(SITE_REPLICATION_PEER_CONNECT_TIMEOUT, Duration::from_secs(3));
-        assert!(
-            SITE_REPLICATION_RETRY_DRAIN_BATCH_TIMEOUT < SITE_REPLICATION_LIFECYCLE_LOCK_TIMEOUT,
-            "the entire retry drain batch must finish before lifecycle waiters time out"
-        );
         assert!(
             SITE_REPLICATION_LIFECYCLE_LOCK_TIMEOUT >= SITE_REPLICATION_PEER_REQUEST_TIMEOUT,
             "a waiter must not give up before the holder's single wedged peer probe can finish"

rustfs/src/site_replication/hooks.rs

@@ -393,30 +393,33 @@ pub(crate) async fn broadcast_site_replication_make_bucket(
     broadcast_site_replication_json_using_runtime(runtime, &configure_path, &serde_json::json!({})).await
 }
 
-async fn broadcast_site_replication_destructive_bucket_op(
-    runtime: &SiteReplicationRuntime,
-    peers: &[PeerInfo],
-    path: &str,
-) -> S3Result<()> {
-    let sends = peers.iter().map(|peer| async move {
-        let result = async {
-            let transport = PeerTransport::for_runtime_peer(peer).await?;
-            PeerAdminRequest::put(&transport.connection, path, &runtime.state.service_account_access_key)
-                .with_client(&transport.client)
-                .send(&runtime.service_account_secret_key, &serde_json::json!({}))
-                .await
+async fn broadcast_site_replication_destructive_bucket_op(runtime: &SiteReplicationRuntime, path: &str) -> S3Result<()> {
+    let sends = runtime.state.peers.values().filter_map(|peer| {
+        if peer.deployment_id == runtime.local_peer.deployment_id
+            || same_identity_endpoint(&peer.endpoint, &runtime.local_peer.endpoint)
+        {
+            return None;
         }
-        .await;
-        match result {
-            Ok(_) => {
-                dequeue_site_replication_retry_event(peer, path).await;
-                None
+        Some(async move {
+            let result = async {
+                let transport = PeerTransport::for_runtime_peer(peer).await?;
+                PeerAdminRequest::put(&transport.connection, path, &runtime.state.service_account_access_key)
+                    .with_client(&transport.client)
+                    .send(&runtime.service_account_secret_key, &serde_json::json!({}))
+                    .await
             }
-            Err(err) => {
-                enqueue_site_replication_retry_event(peer, path, &err).await;
-                Some(err)
+            .await;
+            match result {
+                Ok(_) => {
+                    dequeue_site_replication_retry_event(peer, path).await;
+                    None
+                }
+                Err(err) => {
+                    enqueue_site_replication_retry_event(peer, path, &err).await;
+                    Some(err)
+                }
             }
-        }
+        })
     });
     futures::future::join_all(sends)
         .await
@@ -454,17 +457,19 @@ pub async fn site_replication_delete_bucket_hook(bucket: &str, force_delete: boo
         })
         .cloned()
         .collect::<Vec<_>>();
-    prequeue_site_replication_destructive_events(&retry_peers, &path).await?;
-    let locked_peers = retry_peers;
+    let retry_path = path.clone();
     let result =
         with_config_object_write_lock(store, SITE_REPLICATION_REPAIR_EXECUTION_LOCK_PATH.to_string(), move || async move {
-            broadcast_site_replication_destructive_bucket_op(&runtime, &locked_peers, &path).await
+            broadcast_site_replication_destructive_bucket_op(&runtime, &path).await
         })
         .await;
     match result {
         Ok(result) => result,
         Err(err) => {
             let err: S3Error = ApiError::from(err).into();
+            for peer in &retry_peers {
+                enqueue_site_replication_retry_event(peer, &retry_path, &err).await;
+            }
             Err(err)
         }
     }

rustfs/src/site_replication/retry.rs

@@ -16,8 +16,6 @@ use super::*;
 
 pub(crate) const SITE_REPLICATION_RETRY_QUEUE_LIMIT: usize = 256;
 
-pub(crate) const SITE_REPLICATION_RETRY_DRAIN_BATCH_TIMEOUT: Duration = Duration::from_secs(25);
-
 /// Attempts before an entry reports as `failed` in retryStats. Visibility
 /// only: a `failed` entry stays drain-eligible, and the reachability probe
 /// short-circuits its backoff once the peer answers again — so an early
@@ -63,13 +61,6 @@ pub(crate) fn retry_event_matches(event: &SiteReplicationRetryEvent, peer: &Peer
     (event.peer_deployment_id == peer.deployment_id || event.peer_endpoint == peer.endpoint) && event.path == path
 }
 
-pub(crate) fn retry_event_is_destructive_bucket_op(event: &SiteReplicationRetryEvent) -> bool {
-    matches!(
-        retry_bucket_operation(&event.path).as_deref(),
-        Some("delete-bucket" | "force-delete-bucket" | "purge-deleted-bucket")
-    )
-}
-
 pub(crate) const SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH: &str = "internal:retry-snapshot:iam";
 
 pub(crate) const SITE_REPLICATION_RETRY_BUCKET_METADATA_SNAPSHOT_PATH: &str = "internal:retry-snapshot:bucket-metadata";
@@ -248,15 +239,8 @@ pub(crate) fn upsert_site_replication_retry_event(
         deletions_recorded: false,
     });
     if queue.len() > SITE_REPLICATION_RETRY_QUEUE_LIMIT {
-        while queue.len() > SITE_REPLICATION_RETRY_QUEUE_LIMIT {
-            let Some(index) = queue.iter().position(|event| !retry_event_is_destructive_bucket_op(event)) else {
-                // Destructive rows are a durable outbox, not a replay cache.
-                // Preserve every unacknowledged peer even if that temporarily
-                // exceeds the soft limit so none becomes a silent orphan.
-                break;
-            };
-            queue.remove(index);
-        }
+        let overflow = queue.len() - SITE_REPLICATION_RETRY_QUEUE_LIMIT;
+        queue.drain(0..overflow);
     }
 }
 
@@ -290,37 +274,6 @@ pub(crate) async fn enqueue_site_replication_retry_event(peer: &PeerInfo, path:
     enqueue_site_replication_retry_event_for_generation(peer, path, error, None).await
 }
 
-pub(crate) fn prequeue_site_replication_destructive_events_in_state(
-    state: &mut SiteReplicationState,
-    peers: &[PeerInfo],
-    path: &str,
-) {
-    for peer in peers {
-        if state.peers.contains_key(&peer.deployment_id) {
-            upsert_site_replication_retry_event(
-                &mut state.retry_queue,
-                peer,
-                path,
-                "destructive site replication bucket operation pending delivery",
-                None,
-            );
-        }
-    }
-}
-
-pub(crate) async fn prequeue_site_replication_destructive_events(peers: &[PeerInfo], path: &str) -> S3Result<()> {
-    if peers.is_empty() {
-        return Ok(());
-    }
-    let peers = peers.to_vec();
-    let path = path.to_string();
-    update_site_replication_state(move |state| {
-        prequeue_site_replication_destructive_events_in_state(state, &peers, &path);
-        Ok(())
-    })
-    .await
-}
-
 pub(crate) async fn enqueue_site_replication_retry_event_for_generation(
     peer: &PeerInfo,
     path: &str,
@@ -1340,17 +1293,11 @@ pub(crate) async fn drain_site_replication_retry_queue_inner() -> S3Result<()> {
         let now = OffsetDateTime::now_utc();
         let actionable = actionable_site_replication_retry_events(&runtime.state, now);
         let deferred = deferred_site_replication_retry_events(&runtime.state, now);
-        let drain = async {
-            promote_reachable_deferred_retry_events(&runtime, &actionable, deferred).await?;
-            if actionable.is_empty() {
-                return Ok(());
-            }
-            drain_site_replication_retry_queue_locked(runtime, actionable).await
-        };
-        match tokio::time::timeout(SITE_REPLICATION_RETRY_DRAIN_BATCH_TIMEOUT, drain).await {
-            Ok(result) => result,
-            Err(_) => Ok(()),
+        promote_reachable_deferred_retry_events(&runtime, &actionable, deferred).await?;
+        if actionable.is_empty() {
+            return Ok(());
         }
+        drain_site_replication_retry_queue_locked(runtime, actionable).await
     })
     .await
     .map_err(ApiError::from)?

rustfs/src/site_replication/tests.rs

@@ -793,65 +793,6 @@ fn test_bucket_make_retry_without_matching_configure_fails_closed() {
     assert_eq!(err.code(), &S3ErrorCode::InternalError);
 }
 
-#[test]
-fn test_destructive_bucket_op_prequeues_every_remote_peer() {
-    let peer_b = PeerInfo {
-        deployment_id: "peer-b".to_string(),
-        ..peer("peer-b", "https://peer-b.example.com")
-    };
-    let peer_c = PeerInfo {
-        deployment_id: "peer-c".to_string(),
-        ..peer("peer-c", "https://peer-c.example.com")
-    };
-    let mut state = SiteReplicationState::default();
-    state.peers.insert(peer_b.deployment_id.clone(), peer_b.clone());
-    state.peers.insert(peer_c.deployment_id.clone(), peer_c.clone());
-    let path = "/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=delete-bucket";
-
-    prequeue_site_replication_destructive_events_in_state(&mut state, &[peer_b, peer_c], path);
-
-    assert_eq!(state.retry_queue.len(), 2);
-    assert!(state.retry_queue.iter().all(|event| event.path == path));
-    assert_eq!(
-        state
-            .retry_queue
-            .iter()
-            .map(|event| event.peer_deployment_id.as_str())
-            .collect::<BTreeSet<_>>(),
-        BTreeSet::from(["peer-b", "peer-c"])
-    );
-}
-
-#[test]
-fn test_destructive_retry_intents_survive_the_soft_queue_limit() {
-    let mut queue = (0..SITE_REPLICATION_RETRY_QUEUE_LIMIT)
-        .map(|index| {
-            drain_event(
-                &format!("peer-{index}"),
-                &format!("/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=bucket-{index}&operation=delete-bucket"),
-                1,
-                None,
-            )
-        })
-        .collect::<Vec<_>>();
-    let ordinary_peer = PeerInfo {
-        deployment_id: "ordinary".to_string(),
-        ..peer("ordinary", "https://ordinary.example.com")
-    };
-    upsert_site_replication_retry_event(&mut queue, &ordinary_peer, SITE_REPLICATION_PEER_EDIT_PATH, "ordinary retry", None);
-    assert_eq!(queue.len(), SITE_REPLICATION_RETRY_QUEUE_LIMIT);
-    assert!(queue.iter().all(retry_event_is_destructive_bucket_op));
-
-    let extra_peer = PeerInfo {
-        deployment_id: "extra".to_string(),
-        ..peer("extra", "https://extra.example.com")
-    };
-    let extra_path = "/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=extra&operation=force-delete-bucket";
-    upsert_site_replication_retry_event(&mut queue, &extra_peer, extra_path, "pending destructive retry", None);
-    assert_eq!(queue.len(), SITE_REPLICATION_RETRY_QUEUE_LIMIT + 1);
-    assert!(queue.iter().any(|event| event.path == extra_path));
-}
-
 #[test]
 fn test_reachable_probe_promotion_is_fenced_by_the_observed_event() {
     let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp");