diff --git a/rustfs/src/admin/handlers/site_replication.rs b/rustfs/src/admin/handlers/site_replication.rs index 253920ae1..bacc5093c 100644 --- a/rustfs/src/admin/handlers/site_replication.rs +++ b/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, - "the entire retry probe batch must finish before lifecycle waiters time out" + 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, diff --git a/rustfs/src/site_replication/hooks.rs b/rustfs/src/site_replication/hooks.rs index 57b3e29c9..8ff6079c1 100644 --- a/rustfs/src/site_replication/hooks.rs +++ b/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::>(); + 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 { diff --git a/rustfs/src/site_replication/retry.rs b/rustfs/src/site_replication/retry.rs index 0139d03e7..e96811554 100644 --- a/rustfs/src/site_replication/retry.rs +++ b/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_PROBE_BATCH_TIMEOUT: Duration = Duration::from_secs(20); +pub(crate) const SITE_REPLICATION_RETRY_DRAIN_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, deferred: Vec, ) { - use futures::StreamExt as _; - let due_peers: HashSet = actionable.iter().map(|event| event.peer_deployment_id.clone()).collect(); let mut deferred_by_peer: BTreeMap> = 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); - 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 - { + for (deployment_id, peer_endpoint, events, reachable) in futures::future::join_all(probes).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; - if actionable.is_empty() { - return Ok(()); + let drain = async { + promote_reachable_deferred_retry_events(&runtime, &mut 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(()), } - drain_site_replication_retry_queue_locked(runtime, actionable).await }) .await .map_err(ApiError::from)? diff --git a/rustfs/src/site_replication/tests.rs b/rustfs/src/site_replication/tests.rs index 26c049356..8960634e1 100644 --- a/rustfs/src/site_replication/tests.rs +++ b/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::from(["peer-b", "peer-c"]) + ); +} + #[test] fn test_retry_snapshot_fingerprint_detects_concurrent_iam_change() { let old = SRIAMItem {