fix(site-replication): persist destructive retry intents

rustfs/src/admin/handlers/site_replication.rs

@@ -9570,8 +9570,8 @@ mod tests {
         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_PROBE_BATCH_TIMEOUT < SITE_REPLICATION_LIFECYCLE_LOCK_TIMEOUT,
+            SITE_REPLICATION_RETRY_DRAIN_BATCH_TIMEOUT < SITE_REPLICATION_LIFECYCLE_LOCK_TIMEOUT,
-            "the entire retry probe batch must finish before lifecycle waiters time out"
+            "the entire retry drain batch must finish before lifecycle waiters time out"
         );
         assert!(
             SITE_REPLICATION_LIFECYCLE_LOCK_TIMEOUT >= SITE_REPLICATION_PEER_REQUEST_TIMEOUT,

rustfs/src/site_replication/hooks.rs

@@ -393,6 +393,39 @@ 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, path: &str) -> S3Result<()> {
+    let peers = runtime
+        .state
+        .peers
+        .values()
+        .filter(|peer| {
+            peer.deployment_id != runtime.local_peer.deployment_id
+                && !same_identity_endpoint(&peer.endpoint, &runtime.local_peer.endpoint)
+        })
+        .cloned()
+        .collect::<Vec<_>>();
+    prequeue_site_replication_destructive_events(&peers, path).await?;
+    let mut first_error = None;
+    for peer in &peers {
+        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
+        }
+        .await;
+        match result {
+            Ok(_) => dequeue_site_replication_retry_event(peer, path).await,
+            Err(err) => {
+                enqueue_site_replication_retry_event(peer, path, &err).await;
+                first_error.get_or_insert(err);
+            }
+        }
+    }
+    first_error.map_or(Ok(()), Err)
+}
+
 pub async fn site_replication_delete_bucket_hook(bucket: &str, force_delete: bool) -> S3Result<()> {
     let operation = if force_delete {
         "force-delete-bucket"
@@ -424,7 +457,7 @@ pub async fn site_replication_delete_bucket_hook(bucket: &str, force_delete: boo
     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_json_with_runtime(&runtime, &path, &serde_json::json!({})).await
+            broadcast_site_replication_destructive_bucket_op(&runtime, &path).await
         })
         .await;
     match result {

rustfs/src/site_replication/retry.rs

@@ -16,9 +16,7 @@ use super::*;
 
 pub(crate) const SITE_REPLICATION_RETRY_QUEUE_LIMIT: usize = 256;
 
-pub(crate) const SITE_REPLICATION_RETRY_PROBE_CONCURRENCY: usize = 4;
+pub(crate) const SITE_REPLICATION_RETRY_DRAIN_BATCH_TIMEOUT: Duration = Duration::from_secs(20);
-
-pub(crate) const SITE_REPLICATION_RETRY_PROBE_BATCH_TIMEOUT: Duration = Duration::from_secs(20);
 
 /// Attempts before an entry reports as `failed` in retryStats. Visibility
 /// only: a `failed` entry stays drain-eligible, and the reachability probe
@@ -278,6 +276,37 @@ 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,
@@ -1109,8 +1138,6 @@ pub(crate) async fn promote_reachable_deferred_retry_events(
     actionable: &mut Vec<SiteReplicationRetryEvent>,
     deferred: Vec<SiteReplicationRetryEvent>,
 ) {
-    use futures::StreamExt as _;
-
     let due_peers: HashSet<String> = actionable.iter().map(|event| event.peer_deployment_id.clone()).collect();
     let mut deferred_by_peer: BTreeMap<String, Vec<SiteReplicationRetryEvent>> = BTreeMap::new();
     for event in deferred {
@@ -1136,10 +1163,7 @@ pub(crate) async fn promote_reachable_deferred_retry_events(
             (deployment_id, peer.endpoint.clone(), events, reachable)
         })
     });
-    let mut probes = futures::stream::iter(probes).buffer_unordered(SITE_REPLICATION_RETRY_PROBE_CONCURRENCY);
+    for (deployment_id, peer_endpoint, events, reachable) in futures::future::join_all(probes).await {
-    let deadline = tokio::time::Instant::now() + SITE_REPLICATION_RETRY_PROBE_BATCH_TIMEOUT;
-    while let Ok(Some((deployment_id, peer_endpoint, events, reachable))) = tokio::time::timeout_at(deadline, probes.next()).await
-    {
         if reachable {
             info!(
                 component = LOG_COMPONENT_ADMIN,
@@ -1269,11 +1293,17 @@ pub(crate) async fn drain_site_replication_retry_queue_inner() -> S3Result<()> {
         let now = OffsetDateTime::now_utc();
         let mut actionable = actionable_site_replication_retry_events(&runtime.state, now);
         let deferred = deferred_site_replication_retry_events(&runtime.state, now);
-        promote_reachable_deferred_retry_events(&runtime, &mut actionable, deferred).await;
+        let drain = async {
-        if actionable.is_empty() {
+            promote_reachable_deferred_retry_events(&runtime, &mut actionable, deferred).await;
-            return Ok(());
+            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(()),
         }
-        drain_site_replication_retry_queue_locked(runtime, actionable).await
     })
     .await
     .map_err(ApiError::from)?

rustfs/src/site_replication/tests.rs

@@ -793,6 +793,35 @@ 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_retry_snapshot_fingerprint_detects_concurrent_iam_change() {
     let old = SRIAMItem {