diff --git a/rustfs/src/admin/handlers/site_replication.rs b/rustfs/src/admin/handlers/site_replication.rs index bacc5093c..8ebf3c02a 100644 --- a/rustfs/src/admin/handlers/site_replication.rs +++ b/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" diff --git a/rustfs/src/site_replication/hooks.rs b/rustfs/src/site_replication/hooks.rs index 76ef318c8..fc8b1764f 100644 --- a/rustfs/src/site_replication/hooks.rs +++ b/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::>(); - 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) } } diff --git a/rustfs/src/site_replication/retry.rs b/rustfs/src/site_replication/retry.rs index 2d269276d..990230b14 100644 --- a/rustfs/src/site_replication/retry.rs +++ b/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)? diff --git a/rustfs/src/site_replication/tests.rs b/rustfs/src/site_replication/tests.rs index 78cb55d52..44b1a17e1 100644 --- a/rustfs/src/site_replication/tests.rs +++ b/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::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::>(); - 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");