diff --git a/rustfs/src/admin/handlers/site_replication.rs b/rustfs/src/admin/handlers/site_replication.rs index 6af0b6cf1..c8ddd39db 100644 --- a/rustfs/src/admin/handlers/site_replication.rs +++ b/rustfs/src/admin/handlers/site_replication.rs @@ -2739,6 +2739,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, + deletions_recorded: false, }); state.pending_endpoint_refresh = Some(pending); Ok(()) @@ -3152,6 +3153,7 @@ fn remove_sites(mut state: SiteReplicationState, req: SRRemoveReq) -> SiteReplic state.peers.clear(); state.resync_status.clear(); state.retry_queue.clear(); + state.iam_deletion_replays.clear(); state.pending_endpoint_refresh = None; state.updated_at = Some(OffsetDateTime::now_utc()); return state; @@ -3162,6 +3164,7 @@ fn remove_sites(mut state: SiteReplicationState, req: SRRemoveReq) -> SiteReplic state.peers.clear(); state.resync_status.clear(); state.retry_queue.clear(); + state.iam_deletion_replays.clear(); state.pending_endpoint_refresh = None; state.updated_at = Some(OffsetDateTime::now_utc()); return state; @@ -3182,6 +3185,11 @@ fn remove_sites(mut state: SiteReplicationState, req: SRRemoveReq) -> SiteReplic .iter() .any(|(deployment_id, endpoint)| &event.peer_deployment_id == deployment_id || &event.peer_endpoint == endpoint) }); + state.iam_deletion_replays.retain(|record| { + !removed_peers + .iter() + .any(|(deployment_id, endpoint)| &record.peer_deployment_id == deployment_id || &record.peer_endpoint == endpoint) + }); state .resync_status .retain(|deployment_id, _| state.peers.contains_key(deployment_id)); @@ -5407,7 +5415,14 @@ async fn apply_iam_policy_item(iam_sys: &IamSys, name: &str, policy serde_json::from_value(policy).map_err(|e| s3_error!(InvalidRequest, "invalid policy body: {}", e))?; iam_sys.set_policy(name, policy).await.map_err(ApiError::from)?; } else { - iam_sys.delete_policy(name, true).await.map_err(ApiError::from)?; + // Idempotent delete: the retry drain replays recorded deletions, and + // an entity already absent here IS the converged outcome — erroring + // would wedge the replay forever (backlog#2071). + match iam_sys.delete_policy(name, true).await { + Ok(()) => {} + Err(err) if rustfs_iam::error::is_err_no_such_policy(&err) => {} + Err(err) => return Err(ApiError::from(err).into()), + } } Ok(()) } @@ -5430,10 +5445,26 @@ async fn apply_iam_group_info_item(iam_sys: &IamSys, group_info: Op }; let update = group_info.update_req; if !group_info_requires_upsert(&update) { - iam_sys - .remove_users_from_group(&update.group, update.members) - .await - .map_err(ApiError::from)?; + // Idempotent removal: a replayed deletion may find the group or a + // member already gone (deleted here earlier, or the user tombstone + // was replayed first) — that IS the converged outcome, and erroring + // would wedge the retry drain forever (backlog#2071). Members absent + // here are dropped individually so one missing user cannot veto the + // removal of the members that do exist. + let mut members = Vec::with_capacity(update.members.len()); + for member in update.members.iter() { + if iam_sys.get_user(member).await.is_some() { + members.push(member.clone()); + } + } + if members.is_empty() && !update.members.is_empty() { + return Ok(()); + } + match iam_sys.remove_users_from_group(&update.group, members).await { + Ok(_) => {} + Err(err) if rustfs_iam::error::is_err_no_such_group(&err) => {} + Err(err) => return Err(ApiError::from(err).into()), + } return Ok(()); } @@ -5841,6 +5872,7 @@ impl Operation for SiteReplicationAddHandler { pending_remove: _, pending_endpoint_refresh: _, retry_queue: _, + iam_deletion_replays: _, sync_state_initialized, edit_generation: _, applied_edit_generations: _, @@ -9889,6 +9921,13 @@ mod tests { path: "/rustfs/admin/v3/site-replication/peer/iam-item".to_string(), ..Default::default() }], + iam_deletion_replays: vec![SiteReplicationIamDeletionReplay { + id: "record-1".to_string(), + peer_deployment_id: "remote-dep".to_string(), + peer_endpoint: "https://remote.example.com".to_string(), + entity: "iam-user:alice".to_string(), + ..Default::default() + }], ..Default::default() }; @@ -9901,6 +9940,10 @@ mod tests { ); assert!(state.retry_queue.is_empty()); + assert!( + state.iam_deletion_replays.is_empty(), + "a departed peer's recorded deletions can never be replayed" + ); } #[test] @@ -12073,6 +12116,7 @@ mod tests { last_error: "site replication is not enabled".to_string(), updated_at: Some(OffsetDateTime::now_utc()), edit_generation: None, + deletions_recorded: false, }], ..Default::default() }; @@ -12270,6 +12314,7 @@ mod tests { last_error: "peer offline".to_string(), updated_at: Some(OffsetDateTime::now_utc()), edit_generation: None, + deletions_recorded: false, }], ..Default::default() }; diff --git a/rustfs/src/site_replication/hooks.rs b/rustfs/src/site_replication/hooks.rs index ee03083a6..cd62704ee 100644 --- a/rustfs/src/site_replication/hooks.rs +++ b/rustfs/src/site_replication/hooks.rs @@ -434,8 +434,43 @@ pub async fn site_replication_bucket_meta_hook(mut item: SRBucketMeta) -> S3Resu .await } +/// Broadcast one IAM change to every peer. Unlike the generic JSON broadcast +/// this attempts ALL peers instead of failing fast, and books every failure — +/// transport construction included — through the deletion-aware recorder: a +/// deletion body that failed to reach a peer must be persisted for the retry +/// drain, or the peer keeps the deleted entity forever (backlog#2071). pub async fn site_replication_iam_change_hook(item: SRIAMItem) -> S3Result<()> { - broadcast_site_replication_json("/rustfs/admin/v3/site-replication/peer/iam-item", &item).await + let Some(runtime) = runtime_site_replication_targets().await? else { + return Ok(()); + }; + let mut first_error: Option = None; + for peer in runtime.state.peers.values() { + if peer.deployment_id == runtime.local_peer.deployment_id + || same_identity_endpoint(&peer.endpoint, &runtime.local_peer.endpoint) + { + continue; + } + let sent = async { + let transport = PeerTransport::for_runtime_peer(peer).await?; + PeerAdminRequest::put( + &transport.connection, + SITE_REPLICATION_PEER_IAM_ITEM_WIRE_PATH, + &runtime.state.service_account_access_key, + ) + .with_client(&transport.client) + .send(&runtime.service_account_secret_key, &item) + .await + } + .await; + match sent { + Ok(_) => dequeue_site_replication_retry_event(peer, SITE_REPLICATION_PEER_IAM_ITEM_WIRE_PATH).await, + Err(err) => { + record_failed_site_replication_iam_delivery(peer, &item, &err).await; + first_error.get_or_insert(err); + } + } + } + first_error.map_or(Ok(()), Err) } pub(crate) fn raw_config_to_string(raw: &[u8]) -> Option { diff --git a/rustfs/src/site_replication/repair.rs b/rustfs/src/site_replication/repair.rs index f2c220c47..5b3c260e3 100644 --- a/rustfs/src/site_replication/repair.rs +++ b/rustfs/src/site_replication/repair.rs @@ -652,6 +652,13 @@ pub(crate) async fn persist_site_replication_repair_task( Some(error) => upsert_site_replication_retry_event(&mut state.retry_queue, &peer, &path, error, None), None => { dequeue_site_replication_retry_events_including_escalated(&mut state.retry_queue, &peer, &path); + // A repair is the operator's accountability transfer for the + // possibly-unreplayed deletions too; keeping the records + // without their entry would strand them forever (the drain + // only visits queued entries). + if collapsed_retry_queue_path(&path) == Some(SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH) { + clear_iam_deletion_replays_for_peer(state, &peer); + } } } Ok(()) diff --git a/rustfs/src/site_replication/retry.rs b/rustfs/src/site_replication/retry.rs index ef12c9786..818fbee40 100644 --- a/rustfs/src/site_replication/retry.rs +++ b/rustfs/src/site_replication/retry.rs @@ -16,6 +16,10 @@ use super::*; pub(crate) const SITE_REPLICATION_RETRY_QUEUE_LIMIT: usize = 256; +/// 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 +/// `failed` mark is a timely operator signal, not a dead end. pub(crate) const SITE_REPLICATION_RETRY_FAILED_AFTER: u32 = 3; pub(crate) const SITE_REPLICATION_ENDPOINT_REFRESH_RETRY_PATH: &str = "internal:endpoint-target-refresh"; @@ -37,6 +41,14 @@ pub(crate) struct SiteReplicationRetryEvent { /// [`settle_site_replication_retry_events`]. #[serde(default, skip_serializing_if = "Option::is_none")] pub(crate) edit_generation: Option, + /// 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 + /// successful deletion replay plus a stable snapshot resend settle the + /// entry; a legacy entry (or one degraded by record overflow) keeps the + /// escalation semantics because an unrecorded deletion may hide in it. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub(crate) deletions_recorded: bool, } pub(crate) fn retry_event_matches(event: &SiteReplicationRetryEvent, peer: &PeerInfo, path: &str) -> bool { @@ -87,6 +99,10 @@ pub(crate) fn normalize_collapsed_retry_queue_paths(queue: &mut Vec true, _ => false, }; + // Merged rows may span binaries: a row written without deletion + // recording taints the merged entry, so only both-recorded merges + // stay settleable. + let deletions_recorded = existing.deletions_recorded && event.deletions_recorded; if event_is_newer { let retry_count = existing.retry_count.max(event.retry_count); *existing = event; @@ -94,6 +110,7 @@ pub(crate) fn normalize_collapsed_retry_queue_paths(queue: &mut Vec= SITE_REPLICATION_RETRY_FAILED_AFTER; } *queue = normalized; @@ -210,6 +227,7 @@ pub(crate) fn upsert_site_replication_retry_event( last_error: detail, updated_at: Some(now), edit_generation: generation, + deletions_recorded: false, }); if queue.len() > SITE_REPLICATION_RETRY_QUEUE_LIMIT { let overflow = queue.len() - SITE_REPLICATION_RETRY_QUEUE_LIMIT; @@ -272,6 +290,342 @@ pub(crate) async fn enqueue_site_replication_retry_event_for_generation( } } +pub(crate) const SITE_REPLICATION_PEER_IAM_ITEM_WIRE_PATH: &str = "/rustfs/admin/v3/site-replication/peer/iam-item"; + +/// Per-peer cap on recorded deletion bodies. Beyond it the peer's collapsed +/// IAM entry degrades to the escalation semantics (an unrecorded deletion may +/// exist), so the list stays bounded without silently dropping liability. +pub(crate) const SITE_REPLICATION_IAM_DELETION_REPLAY_LIMIT_PER_PEER: usize = 256; + +/// One IAM deletion event whose delivery to `peer` failed, kept verbatim so +/// the retry drain can replay it before the snapshot resend. `entity` +/// collapses repeated deletions of the same entity into the newest body. +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +pub(crate) struct SiteReplicationIamDeletionReplay { + pub(crate) id: String, + pub(crate) peer_deployment_id: String, + pub(crate) peer_endpoint: String, + pub(crate) entity: String, + pub(crate) item: Value, + #[serde(default, with = "time::serde::rfc3339::option", skip_serializing_if = "Option::is_none")] + pub(crate) recorded_at: Option, +} + +pub(crate) fn iam_deletion_replay_matches(record: &SiteReplicationIamDeletionReplay, peer: &PeerInfo) -> bool { + record.peer_deployment_id == peer.deployment_id || record.peer_endpoint == peer.endpoint +} + +/// The entity a deletion-shaped IAM item removes, or `None` for items that +/// create or update state (those are faithfully replayed by the snapshot +/// resend and need no record). Group member removal keys on the removed +/// member set: two removals from the same group are distinct events, not a +/// newer revision of one another. +pub(crate) fn iam_item_deletion_entity(item: &SRIAMItem) -> Option { + match item.r#type.as_str() { + "policy" if item.policy.is_none() => Some(format!("policy:{}", item.name)), + "iam-user" => item + .iam_user + .as_ref() + .filter(|user| user.is_delete_req) + .map(|user| format!("iam-user:{}", user.access_key)), + "group-info" => item + .group_info + .as_ref() + .filter(|group| group.update_req.is_remove) + .map(|group| { + let mut members = group.update_req.members.clone(); + members.sort_unstable(); + format!("group-remove:{}:{}", group.update_req.group, members.join(",")) + }), + "policy-mapping" => item + .policy_mapping + .as_ref() + .filter(|mapping| mapping.policy.is_empty()) + .map(|mapping| format!("policy-mapping:{}:{}:{}", mapping.user_or_group, mapping.user_type, mapping.is_group)), + "service-account" => item + .svc_acc_change + .as_ref() + .and_then(|change| change.delete.as_ref()) + .map(|delete| format!("svc-acc:{}", delete.access_key)), + _ => None, + } +} + +/// Failure bookkeeping for one IAM item delivery: upsert the collapsed retry +/// event and, when the item is a deletion, record its body for replay. Both +/// live in the same state so the caller commits them in one transaction — a +/// retry entry can never exist whose deletion body was lost to a separate +/// failed write. +pub(crate) fn record_failed_iam_delivery(state: &mut SiteReplicationState, peer: &PeerInfo, item: &SRIAMItem, error: &str) { + let existed = state + .retry_queue + .iter() + .any(|event| retry_event_matches(event, peer, SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH)); + upsert_site_replication_retry_event(&mut state.retry_queue, peer, SITE_REPLICATION_PEER_IAM_ITEM_WIRE_PATH, error, None); + if !existed + && let Some(event) = state + .retry_queue + .iter_mut() + .find(|event| retry_event_matches(event, peer, SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH)) + { + // Fresh entry: every failure it will ever collapse goes through this + // recording path, so a deletion replay plus a stable snapshot resend + // can later settle it instead of escalating. + event.deletions_recorded = true; + } + + let Some(entity) = iam_item_deletion_entity(item) else { + return; + }; + let item_value = match serde_json::to_value(item) { + Ok(value) => value, + Err(_) => { + degrade_iam_retry_event_to_escalation(state, peer); + return; + } + }; + let now = OffsetDateTime::now_utc(); + if let Some(existing) = state + .iam_deletion_replays + .iter_mut() + .find(|record| iam_deletion_replay_matches(record, peer) && record.entity == entity) + { + existing.item = item_value; + existing.recorded_at = Some(now); + return; + } + + let per_peer = state + .iam_deletion_replays + .iter() + .filter(|record| iam_deletion_replay_matches(record, peer)) + .count(); + if per_peer >= SITE_REPLICATION_IAM_DELETION_REPLAY_LIMIT_PER_PEER { + // The record set is no longer complete for this peer; the entry must + // escalate rather than settle. Drop the oldest record to stay + // bounded — remaining records are still replayed best-effort. + degrade_iam_retry_event_to_escalation(state, peer); + if let Some(oldest) = state + .iam_deletion_replays + .iter() + .enumerate() + .filter(|(_, record)| iam_deletion_replay_matches(record, peer)) + .min_by_key(|(_, record)| record.recorded_at) + .map(|(index, _)| index) + { + state.iam_deletion_replays.remove(oldest); + } + } + state.iam_deletion_replays.push(SiteReplicationIamDeletionReplay { + id: Uuid::new_v4().to_string(), + peer_deployment_id: peer.deployment_id.clone(), + peer_endpoint: peer.endpoint.clone(), + entity, + item: item_value, + recorded_at: Some(now), + }); +} + +pub(crate) fn degrade_iam_retry_event_to_escalation(state: &mut SiteReplicationState, peer: &PeerInfo) { + if let Some(event) = state + .retry_queue + .iter_mut() + .find(|event| retry_event_matches(event, peer, SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH)) + { + event.deletions_recorded = false; + } +} + +pub(crate) fn iam_deletion_replays_for_peer( + state: &SiteReplicationState, + peer: &PeerInfo, +) -> Vec { + let mut records: Vec = state + .iam_deletion_replays + .iter() + .filter(|record| iam_deletion_replay_matches(record, peer)) + .cloned() + .collect(); + // Oldest first, so a later deletion of a recreated entity lands after + // the earlier one. + records.sort_by_key(|record| record.recorded_at); + records +} + +pub(crate) fn clear_iam_deletion_replays_for_peer(state: &mut SiteReplicationState, peer: &PeerInfo) { + state + .iam_deletion_replays + .retain(|record| !iam_deletion_replay_matches(record, peer)); +} + +pub(crate) async fn record_failed_site_replication_iam_delivery(peer: &PeerInfo, item: &SRIAMItem, error: &S3Error) { + let peer_owned = peer.clone(); + let item_owned = item.clone(); + let error_text = error.to_string(); + let deletion_entity = iam_item_deletion_entity(item); + let result = update_site_replication_state(move |state| { + // A departed peer can never drain its entries again (remove_sites + // already pruned them) — mirror enqueue_site_replication_retry_event. + if state.peers.contains_key(&peer_owned.deployment_id) { + record_failed_iam_delivery(state, &peer_owned, &item_owned, &error_text); + } + Ok(()) + }) + .await; + + match result { + Ok(()) => { + if let Some(entity) = deletion_entity { + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SITE_REPLICATION, + event = EVENT_ADMIN_SITE_REPLICATION_STATE, + peer = %peer.endpoint, + deployment_id = %peer.deployment_id, + entity = %entity, + result = "iam_deletion_recorded_for_replay", + "IAM deletion delivery to peer failed; recorded for retry-drain replay" + ); + } + } + Err(err) => { + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SITE_REPLICATION, + event = EVENT_ADMIN_SITE_REPLICATION_STATE, + peer = %peer.endpoint, + deployment_id = %peer.deployment_id, + error = ?err, + "failed to persist site replication IAM delivery failure" + ); + } + } +} + +/// Post-replay settlement for a collapsed IAM entry: remove the replayed +/// deletion records and, when the entry's whole liability is provably +/// replayed (`deletions_recorded` and no residual records), remove the entry. +/// Anything else falls back to the escalation marker, and a failure stamped +/// after `snapshot_updated_at` keeps the entry drain-eligible untouched. +/// Returns whether the entry was fully settled. +pub(crate) fn settle_replayed_iam_retry_events( + state: &mut SiteReplicationState, + peer: &PeerInfo, + path: &str, + snapshot_updated_at: Option, + replayed_record_ids: &[String], +) -> bool { + state + .iam_deletion_replays + .retain(|record| !(iam_deletion_replay_matches(record, peer) && replayed_record_ids.contains(&record.id))); + + if collapsed_retry_queue_path(path) != Some(SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH) { + return false; + } + let Some(index) = state + .retry_queue + .iter() + .position(|event| retry_event_matches(event, peer, SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH)) + else { + return false; + }; + let event = &state.retry_queue[index]; + let newer_failure_recorded = match (event.updated_at, snapshot_updated_at) { + (Some(current), Some(seen)) => current > seen, + (Some(_), None) => true, + (None, _) => false, + }; + if newer_failure_recorded && event.last_error != SITE_REPLICATION_RETRY_SNAPSHOT_REPLAYED_MARKER { + // The newer failure's own deletion (if any) has its own record; the + // next drain pass replays it. + return false; + } + let residual_records = state + .iam_deletion_replays + .iter() + .any(|record| iam_deletion_replay_matches(record, peer)); + if event.deletions_recorded && !residual_records { + state.retry_queue.remove(index); + return true; + } + escalate_site_replication_retry_events_up_to(&mut state.retry_queue, peer, path, snapshot_updated_at); + false +} + +pub(crate) async fn settle_replayed_site_replication_iam_retry_event( + peer: &PeerInfo, + path: &str, + snapshot_updated_at: Option, + replayed_record_ids: Vec, +) { + let peer_owned = peer.clone(); + let path_owned = path.to_string(); + let result = update_site_replication_state(move |state| { + Ok(settle_replayed_iam_retry_events( + state, + &peer_owned, + &path_owned, + snapshot_updated_at, + &replayed_record_ids, + )) + }) + .await; + + match result { + Ok(true) => { + info!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SITE_REPLICATION, + event = EVENT_ADMIN_SITE_REPLICATION_STATE, + peer = %peer.endpoint, + deployment_id = %peer.deployment_id, + result = "iam_retry_event_settled", + "recorded IAM deletions replayed and snapshot stable; collapsed retry entry settled" + ); + } + Ok(false) => {} + Err(err) => { + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SITE_REPLICATION, + event = EVENT_ADMIN_SITE_REPLICATION_STATE, + peer = %peer.endpoint, + deployment_id = %peer.deployment_id, + path, + error = ?err, + "failed to settle replayed site replication IAM retry event" + ); + } + } +} + +/// Drop a deletion record whose body no longer deserializes (it can never be +/// replayed) and degrade the peer's entry to escalation so the liability +/// stays operator-visible instead of silently settling. +pub(crate) async fn drop_corrupt_iam_deletion_replay(peer: &PeerInfo, record_id: &str) { + let peer_owned = peer.clone(); + let record_id_owned = record_id.to_string(); + let result = update_site_replication_state(move |state| { + state.iam_deletion_replays.retain(|record| record.id != record_id_owned); + degrade_iam_retry_event_to_escalation(state, &peer_owned); + Ok(()) + }) + .await; + + if let Err(err) = result { + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SITE_REPLICATION, + event = EVENT_ADMIN_SITE_REPLICATION_STATE, + peer = %peer.endpoint, + deployment_id = %peer.deployment_id, + record_id, + error = ?err, + "failed to drop corrupt site replication IAM deletion record" + ); + } +} + pub(crate) fn retry_bucket_operation(path: &str) -> Option { let (base_path, query) = path.split_once('?')?; if base_path != SITE_REPLICATION_PEER_BUCKET_OPS_PATH { @@ -292,7 +646,10 @@ pub(crate) fn retry_event_replayed_by_bootstrap(event: &SiteReplicationRetryEven /// reconcile cadence (`site_replication_reconcile::RECONCILE_INTERVAL`). pub(crate) const SITE_REPLICATION_RETRY_DRAIN_BASE_BACKOFF_SECS: i64 = 600; -/// Backoff ceiling: a permanently failed peer is still probed daily. +/// Backoff ceiling for *replay attempts*: a permanently failed peer still +/// gets a full replay daily. Reachability is separately probed every tick +/// ([`promote_reachable_deferred_retry_events`]), so a peer that recovers +/// converges at the next tick instead of waiting out this ceiling. pub(crate) const SITE_REPLICATION_RETRY_DRAIN_MAX_BACKOFF_SECS: i64 = 86_400; /// What the background drain may do for one retry event. Everything not @@ -628,16 +985,150 @@ pub(crate) fn actionable_site_replication_retry_events( .collect() } +/// Replayable events currently held back only by backoff. The exponential +/// 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. +pub(crate) fn deferred_site_replication_retry_events( + state: &SiteReplicationState, + now: OffsetDateTime, +) -> Vec { + state + .retry_queue + .iter() + .filter(|event| classify_site_replication_retry_event(event).is_some()) + .filter(|event| state.peers.contains_key(&event.peer_deployment_id)) + .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 + }) + }) + .cloned() + .collect() +} + +/// Peer link-check upload drain: a bodiless-in-spirit POST every peer accepts +/// with the replication service account, discarding the payload. The cheapest +/// authenticated proof that the peer is reachable again. +pub(crate) const SITE_REPLICATION_PEER_DEVNULL_PATH: &str = "/rustfs/admin/v3/site-replication/devnull"; + +pub(crate) async fn probe_site_replication_peer_reachable(runtime: &SiteReplicationRuntime, peer: &PeerInfo) -> bool { + let Ok(transport) = PeerTransport::for_runtime_peer(peer).await else { + return false; + }; + PeerAdminRequest::post( + &transport.connection, + SITE_REPLICATION_PEER_DEVNULL_PATH, + &runtime.state.service_account_access_key, + ) + .with_client(&transport.client) + .send(&runtime.service_account_secret_key, &serde_json::json!({})) + .await + .is_ok() +} + +/// Probe the peers whose whole backlog is deferred and promote the backlog of +/// every peer that answers. A probe failure advances nothing: retry counts +/// only move on real delivery attempts, so the per-event backoff is intact +/// when the peer is genuinely down. +pub(crate) async fn promote_reachable_deferred_retry_events( + runtime: &SiteReplicationRuntime, + actionable: &mut Vec, + deferred: Vec, +) { + 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 { + if due_peers.contains(&event.peer_deployment_id) { + // The peer is being dialed this tick anyway; its other events keep + // their own backoff. + continue; + } + deferred_by_peer + .entry(event.peer_deployment_id.clone()) + .or_default() + .push(event); + } + for (deployment_id, events) in deferred_by_peer { + let Some(peer) = runtime.state.peers.get(&deployment_id) else { + continue; + }; + if deployment_id == runtime.local_peer.deployment_id + || same_identity_endpoint(&peer.endpoint, &runtime.local_peer.endpoint) + { + continue; + } + if probe_site_replication_peer_reachable(runtime, peer).await { + info!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SITE_REPLICATION, + event = EVENT_ADMIN_SITE_REPLICATION_STATE, + peer = %peer.endpoint, + deployment_id = %deployment_id, + promoted = events.len(), + result = "retry_backoff_probe_promoted", + "peer reachable again; replaying its backed-off retry events this tick" + ); + actionable.extend(events); + } + } +} + +/// Operator-visible per-tick alert for retry entries that no longer converge +/// on their own: `failed` deliveries deep in backoff and snapshot-escalated +/// markers awaiting a repair. Healthy pending entries stay silent. +pub(crate) fn log_site_replication_retry_liabilities(state: &SiteReplicationState) { + let escalated = state + .retry_queue + .iter() + .filter(|event| event.last_error == SITE_REPLICATION_RETRY_SNAPSHOT_REPLAYED_MARKER) + .count(); + let failed = state + .retry_queue + .iter() + .filter(|event| event.failed && event.last_error != SITE_REPLICATION_RETRY_SNAPSHOT_REPLAYED_MARKER) + .count(); + if failed == 0 && escalated == 0 { + return; + } + let pending = state.retry_queue.len().saturating_sub(failed + escalated); + let oldest_updated_at = state + .retry_queue + .iter() + .filter(|event| event.failed || event.last_error == SITE_REPLICATION_RETRY_SNAPSHOT_REPLAYED_MARKER) + .filter_map(|event| event.updated_at) + .min(); + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SITE_REPLICATION, + event = EVENT_ADMIN_SITE_REPLICATION_STATE, + failed, + escalated, + pending, + oldest_updated_at = ?oldest_updated_at, + recorded_deletions = state.iam_deletion_replays.len(), + result = "retry_liabilities_outstanding", + "site replication retry queue holds failed or escalated deliveries; peer convergence is degraded" + ); +} + /// Background consumer for the retry queue, run from the reconcile tick. /// /// Scope: this settles "delivered once and failed" entries whose replay is -/// faithful (bucket ops, peer edits). Collapsed iam-item / bucket-meta -/// entries are snapshot-resent and then *escalated*, not cleared — a failed -/// deletion leaves no task in the snapshot, so remote absence stays unproven -/// until a later delivery or a manual repair. A hook that never fired (crash -/// between the local commit and the send) leaves no entry at all, so the -/// drain is not a full cross-site diff-heal; manual repair remains the -/// authoritative catch-all. +/// faithful (bucket ops, peer edits). Collapsed iam-item entries replay the +/// recorded deletion bodies and then the snapshot, which together cover every +/// failure the hook recorded, so a fully-recorded entry settles; an entry +/// with an unrecorded deletion (legacy rows, record overflow) is *escalated* +/// instead — remote absence stays unproven until a later delivery or a +/// manual repair. Collapsed bucket-meta entries keep the escalate-only +/// semantics. A hook that never fired (crash between the local commit and +/// the send) leaves no entry at all, so the drain is not a full cross-site +/// diff-heal; manual repair remains the authoritative catch-all. pub(crate) async fn drain_site_replication_retry_queue() { if let Err(err) = drain_site_replication_retry_queue_inner().await { warn!( @@ -655,8 +1146,13 @@ pub(crate) async fn drain_site_replication_retry_queue_inner() -> S3Result<()> { let Some(runtime) = runtime_site_replication_targets().await? else { return Ok(()); }; - let actionable = actionable_site_replication_retry_events(&runtime.state, OffsetDateTime::now_utc()); - if actionable.is_empty() { + // The alert must fire even when nothing is drainable this tick — + // escalated markers are exactly the entries the drain skips. + log_site_replication_retry_liabilities(&runtime.state); + 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); + if actionable.is_empty() && deferred.is_empty() { return Ok(()); } let Some(store) = current_object_store_handle() else { @@ -671,6 +1167,12 @@ pub(crate) async fn drain_site_replication_retry_queue_inner() -> S3Result<()> { // guard) may have started since. Re-check on the fresh state. return Ok(()); } + // Probe before taking the repair lock: probes are read-only peer traffic + // and a dead peer's connect timeout must not hold the lock. + promote_reachable_deferred_retry_events(&runtime, &mut actionable, deferred).await; + if actionable.is_empty() { + return Ok(()); + } // Serialize against operator repair execution. This does NOT close the // dry-run -> execute window (dry-run takes no lock): a drain settling a // replayable bucket-op entry in that window changes the preflight token @@ -778,6 +1280,28 @@ pub(crate) async fn drain_one_site_replication_retry_event( let Some(plan) = plan else { return Ok(false); }; + // Replay recorded IAM deletion bodies BEFORE the snapshot: an + // entity deleted and later recreated locally is restored by the + // snapshot that follows, so the replay can never end below the + // current local state (backlog#2071). + let is_iam = matches!(action, RetryDrainAction::IamSnapshot); + let mut replayed_record_ids = Vec::new(); + if is_iam { + for record in iam_deletion_replays_for_peer(&runtime.state, peer) { + let Ok(item) = serde_json::from_value::(record.item.clone()) else { + drop_corrupt_iam_deletion_replay(peer, &record.id).await; + continue; + }; + if let Err(err) = SiteReplicationRepairTask::Iam(&item) + .send(transport, access_key, secret_key) + .await + { + enqueue_site_replication_retry_event(peer, &event.path, &err).await; + return Err(err); + } + replayed_record_ids.push(record.id.clone()); + } + } let mut current_snapshot = RetrySnapshot::from_plan(&action, plan).expect("snapshot action has a snapshot"); let mut replay = current_snapshot.clone(); for _ in 0..SITE_REPLICATION_RETRY_SNAPSHOT_STABILITY_ATTEMPTS { @@ -790,7 +1314,17 @@ pub(crate) async fn drain_one_site_replication_retry_event( let fresh_plan = site_replication_bootstrap_plan(&fresh_info)?; let fresh_snapshot = RetrySnapshot::from_plan(&action, &fresh_plan).expect("snapshot action has a snapshot"); if fresh_snapshot.fingerprint()? == current_fingerprint { - escalate_site_replication_retry_event_up_to(peer, &event.path, event.updated_at).await; + if is_iam { + settle_replayed_site_replication_iam_retry_event( + peer, + &event.path, + event.updated_at, + replayed_record_ids, + ) + .await; + } else { + escalate_site_replication_retry_event_up_to(peer, &event.path, event.updated_at).await; + } return Ok(true); } replay = RetrySnapshot::replay_after_change(¤t_snapshot, &fresh_snapshot, OffsetDateTime::now_utc()); diff --git a/rustfs/src/site_replication/state.rs b/rustfs/src/site_replication/state.rs index 3d2c5c74f..04619a9ac 100644 --- a/rustfs/src/site_replication/state.rs +++ b/rustfs/src/site_replication/state.rs @@ -41,6 +41,14 @@ pub(crate) struct SiteReplicationState { pub(crate) pending_endpoint_refresh: Option, #[serde(default, skip_serializing_if = "Vec::is_empty")] pub(crate) retry_queue: Vec, + /// Explicitly recorded IAM deletion events whose delivery to a peer + /// failed. The bootstrap snapshot cannot express "this entity no longer + /// exists", so these bodies are what makes a failed deletion replayable + /// by the retry drain instead of a permanent escalated marker + /// (backlog#2071). Only deletions recorded here are ever replayed — the + /// drain never derives deletions from a cross-site diff. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub(crate) iam_deletion_replays: Vec, #[serde(default)] pub(crate) sync_state_initialized: bool, /// Fencing token for peer-edit delivery, allocated inside the state @@ -121,6 +129,14 @@ pub(crate) fn parse_site_replication_state(data: &[u8]) -> S3Result SRIAMItem { + SRIAMItem { + r#type: "iam-user".to_string(), + iam_user: Some(rustfs_madmin::SRIAMUser { + access_key: access_key.to_string(), + is_delete_req: true, + user_req: None, + api_version: Some(SITE_REPL_API_VERSION.to_string()), + }), + updated_at: Some(OffsetDateTime::now_utc()), + ..Default::default() + } +} + +fn policy_delete_item(name: &str) -> SRIAMItem { + SRIAMItem { + r#type: "policy".to_string(), + name: name.to_string(), + policy: None, + updated_at: Some(OffsetDateTime::now_utc()), + ..Default::default() + } +} + +fn deletion_replay_state(peer: &PeerInfo) -> SiteReplicationState { + let mut state = SiteReplicationState::default(); + state.peers.insert(peer.deployment_id.clone(), peer.clone()); + state +} + +/// Deletion-shaped IAM items get an entity key (and hence a replay record); +/// creations and updates are covered by the snapshot resend and get none. +#[test] +fn test_iam_item_deletion_entity_shapes() { + assert_eq!(iam_item_deletion_entity(&user_delete_item("alice")).as_deref(), Some("iam-user:alice")); + assert_eq!( + iam_item_deletion_entity(&policy_delete_item("readonly")).as_deref(), + Some("policy:readonly") + ); + + let group_remove = SRIAMItem { + r#type: "group-info".to_string(), + group_info: Some(SRGroupInfo { + update_req: GroupAddRemove { + group: "devs".to_string(), + members: vec!["bob".to_string(), "alice".to_string()], + status: GroupStatus::Enabled, + is_remove: true, + }, + api_version: None, + }), + ..Default::default() + }; + assert_eq!( + iam_item_deletion_entity(&group_remove).as_deref(), + Some("group-remove:devs:alice,bob"), + "member set must be part of the key so distinct removals do not collapse" + ); + + let mapping_clear = SRIAMItem { + r#type: "policy-mapping".to_string(), + policy_mapping: Some(SRPolicyMapping { + user_or_group: "alice".to_string(), + user_type: 0, + is_group: false, + policy: String::new(), + ..Default::default() + }), + ..Default::default() + }; + assert_eq!(iam_item_deletion_entity(&mapping_clear).as_deref(), Some("policy-mapping:alice:0:false")); + + let svc_acc_delete = SRIAMItem { + r#type: "service-account".to_string(), + svc_acc_change: Some(rustfs_madmin::SRSvcAccChange { + delete: Some(rustfs_madmin::SRSvcAccDelete { + access_key: "svc-1".to_string(), + api_version: None, + }), + ..Default::default() + }), + ..Default::default() + }; + assert_eq!(iam_item_deletion_entity(&svc_acc_delete).as_deref(), Some("svc-acc:svc-1")); + + // Creations/updates carry no deletion entity. + let mut user_create = user_delete_item("alice"); + user_create.iam_user.as_mut().expect("iam user").is_delete_req = false; + assert!(iam_item_deletion_entity(&user_create).is_none()); + let mut policy_set = policy_delete_item("readonly"); + policy_set.policy = Some(serde_json::json!({"Version": "2012-10-17"})); + assert!(iam_item_deletion_entity(&policy_set).is_none()); +} + +/// A failed deletion delivery persists a replay record next to the collapsed +/// retry entry; a fresh entry is stamped `deletions_recorded` so a later +/// replay can settle it, and a repeated deletion of the same entity keeps the +/// newest body instead of growing the list. +#[test] +fn test_record_failed_iam_delivery_records_deletions_and_flags_entry() { + let target = PeerInfo { + deployment_id: "remote-dep".to_string(), + ..peer("remote", "https://remote.example.com") + }; + let mut state = deletion_replay_state(&target); + + // Non-deletion failure: entry flagged, no record. + let mut user_update = user_delete_item("alice"); + user_update.iam_user.as_mut().expect("iam user").is_delete_req = false; + record_failed_iam_delivery(&mut state, &target, &user_update, "peer offline"); + assert_eq!(state.retry_queue.len(), 1); + assert_eq!(state.retry_queue[0].path, SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH); + assert!(state.retry_queue[0].deletions_recorded); + assert!(state.iam_deletion_replays.is_empty()); + + // Deletion failure: recorded for replay, entry stays flagged. + record_failed_iam_delivery(&mut state, &target, &user_delete_item("alice"), "peer offline"); + assert_eq!(state.iam_deletion_replays.len(), 1); + assert_eq!(state.iam_deletion_replays[0].entity, "iam-user:alice"); + assert!(state.retry_queue[0].deletions_recorded); + assert_eq!(state.retry_queue.len(), 1, "IAM failures stay collapsed per peer"); + + // Same entity again: newest body replaces the record. + record_failed_iam_delivery(&mut state, &target, &user_delete_item("alice"), "peer offline"); + assert_eq!(state.iam_deletion_replays.len(), 1); + + // Different entity: second record. + record_failed_iam_delivery(&mut state, &target, &policy_delete_item("readonly"), "peer offline"); + assert_eq!(state.iam_deletion_replays.len(), 2); + + // A legacy entry (created without recording) is never stamped. + let legacy = PeerInfo { + deployment_id: "legacy-dep".to_string(), + ..peer("legacy", "https://legacy.example.com") + }; + state.peers.insert(legacy.deployment_id.clone(), legacy.clone()); + upsert_site_replication_retry_event( + &mut state.retry_queue, + &legacy, + SITE_REPLICATION_PEER_IAM_ITEM_WIRE_PATH, + "peer offline", + None, + ); + record_failed_iam_delivery(&mut state, &legacy, &user_delete_item("bob"), "peer offline"); + let legacy_event = state + .retry_queue + .iter() + .find(|event| event.peer_deployment_id == legacy.deployment_id) + .expect("legacy entry"); + assert!( + !legacy_event.deletions_recorded, + "an entry that predates recording may hide an unrecorded deletion" + ); +} + +/// Overflowing the per-peer record cap degrades the entry back to the +/// escalation semantics: the record set is no longer complete, so a replay +/// can no longer prove the peer converged. +#[test] +fn test_record_failed_iam_delivery_overflow_degrades_to_escalation() { + let target = PeerInfo { + deployment_id: "remote-dep".to_string(), + ..peer("remote", "https://remote.example.com") + }; + let mut state = deletion_replay_state(&target); + for index in 0..SITE_REPLICATION_IAM_DELETION_REPLAY_LIMIT_PER_PEER { + record_failed_iam_delivery(&mut state, &target, &policy_delete_item(&format!("p{index}")), "peer offline"); + } + assert!(state.retry_queue[0].deletions_recorded); + assert_eq!(state.iam_deletion_replays.len(), SITE_REPLICATION_IAM_DELETION_REPLAY_LIMIT_PER_PEER); + + record_failed_iam_delivery(&mut state, &target, &policy_delete_item("one-too-many"), "peer offline"); + assert_eq!( + state.iam_deletion_replays.len(), + SITE_REPLICATION_IAM_DELETION_REPLAY_LIMIT_PER_PEER, + "the list stays bounded" + ); + assert!( + !state.retry_queue[0].deletions_recorded, + "an overflowed record set can no longer settle the entry" + ); +} + +/// After a successful deletion replay plus a stable snapshot resend, a +/// fully-recorded entry settles (entry and replayed records removed); an +/// unrecorded entry escalates as before, and a failure stamped after the +/// snapshot keeps the entry drain-eligible. +#[test] +fn test_settle_replayed_iam_retry_events_settles_or_escalates() { + let target = PeerInfo { + deployment_id: "remote-dep".to_string(), + ..peer("remote", "https://remote.example.com") + }; + let snapshot_at = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp"); + + // Fully recorded: settles. + let mut state = deletion_replay_state(&target); + record_failed_iam_delivery(&mut state, &target, &user_delete_item("alice"), "peer offline"); + state.retry_queue[0].updated_at = Some(snapshot_at); + let replayed: Vec = state.iam_deletion_replays.iter().map(|record| record.id.clone()).collect(); + assert!(settle_replayed_iam_retry_events( + &mut state, + &target, + SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH, + Some(snapshot_at), + &replayed, + )); + assert!(state.retry_queue.is_empty()); + assert!(state.iam_deletion_replays.is_empty()); + + // Not fully recorded: replayed records are still removed, but the entry + // escalates instead of settling. + let mut state = deletion_replay_state(&target); + record_failed_iam_delivery(&mut state, &target, &user_delete_item("alice"), "peer offline"); + state.retry_queue[0].updated_at = Some(snapshot_at); + state.retry_queue[0].deletions_recorded = false; + let replayed: Vec = state.iam_deletion_replays.iter().map(|record| record.id.clone()).collect(); + assert!(!settle_replayed_iam_retry_events( + &mut state, + &target, + SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH, + Some(snapshot_at), + &replayed, + )); + assert!(state.iam_deletion_replays.is_empty()); + assert_eq!(state.retry_queue.len(), 1); + assert_eq!(state.retry_queue[0].last_error, SITE_REPLICATION_RETRY_SNAPSHOT_REPLAYED_MARKER); + + // Newer failure since the snapshot: entry untouched and drain-eligible, + // residual (unreplayed) record kept for the next pass. + let mut state = deletion_replay_state(&target); + record_failed_iam_delivery(&mut state, &target, &user_delete_item("alice"), "peer offline"); + let replayed: Vec = state.iam_deletion_replays.iter().map(|record| record.id.clone()).collect(); + record_failed_iam_delivery(&mut state, &target, &user_delete_item("bob"), "peer offline"); + state.retry_queue[0].updated_at = Some(snapshot_at + time::Duration::seconds(5)); + assert!(!settle_replayed_iam_retry_events( + &mut state, + &target, + SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH, + Some(snapshot_at), + &replayed, + )); + assert_eq!(state.retry_queue.len(), 1); + assert_ne!(state.retry_queue[0].last_error, SITE_REPLICATION_RETRY_SNAPSHOT_REPLAYED_MARKER); + assert!( + classify_site_replication_retry_event(&state.retry_queue[0]).is_some(), + "the newer failure must stay drain-eligible" + ); + assert_eq!(state.iam_deletion_replays.len(), 1); + assert_eq!(state.iam_deletion_replays[0].entity, "iam-user:bob"); +} + +/// Merging legacy wire-path rows into the collapsed entry must not launder an +/// unrecorded deletion into a settleable entry. +#[test] +fn test_normalize_collapsed_paths_taints_merged_deletion_recording() { + let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp"); + let mut recorded = drain_event("remote-dep", SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH, 1, Some(now)); + recorded.deletions_recorded = true; + let legacy = drain_event( + "remote-dep", + "/rustfs/admin/v3/site-replication/peer/iam-item", + 2, + Some(now + time::Duration::seconds(5)), + ); + let mut queue = vec![recorded, legacy]; + assert!(normalize_collapsed_retry_queue_paths(&mut queue)); + assert_eq!(queue.len(), 1); + assert!(!queue[0].deletions_recorded, "a merged legacy row may hide an unrecorded deletion"); +} + /// P1-3 red-light: the drain must only ever act on deliveries it can /// replay faithfully. IAM / bucket-meta entries collapse per (peer, path) /// with no body persisted — only a snapshot resend is truthful; bucket @@ -590,6 +862,59 @@ fn test_actionable_site_replication_retry_events_filters() { assert_eq!(actionable[0].path, SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH); } +/// The deferred subset is the probe's territory: replayable events held back +/// only by backoff, at least one base interval after their last failure. A +/// recovered peer's bucket-op stuck behind a 2400s+ backoff (the round-four +/// R1.6 shape: three outage-window failures, then a 900s test window) must +/// appear here so the probe can promote it at the first tick. +#[test] +fn test_deferred_site_replication_retry_events_partition() { + 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 = vec![ + // retry_count 3 => 2400s backoff; failed 700s ago: deferred. + drain_event("remote", bucket_make, 3, Some(now - time::Duration::seconds(700))), + // Failed less than one base interval ago: neither due nor probed. + drain_event( + "remote", + SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH, + 3, + Some(now - time::Duration::seconds(300)), + ), + // Past its own backoff: actionable, not deferred. + drain_event( + "remote", + "/rustfs/admin/v3/site-replication/peer/bucket-meta", + 1, + Some(now - time::Duration::seconds(700)), + ), + // Unknown peer: never probed. + drain_event("gone", bucket_make, 3, Some(now - time::Duration::seconds(700))), + ]; + // Escalated marker: not replayable, never probed. + let mut escalated = drain_event( + "remote", + SITE_REPLICATION_RETRY_BUCKET_METADATA_SNAPSHOT_PATH, + 3, + Some(now - time::Duration::seconds(700)), + ); + escalated.last_error = SITE_REPLICATION_RETRY_SNAPSHOT_REPLAYED_MARKER.to_string(); + state.retry_queue.push(escalated); + + let deferred = deferred_site_replication_retry_events(&state, now); + assert_eq!(deferred.len(), 1, "only the backed-off, replayable, known-peer event defers"); + assert_eq!(deferred[0].path, bucket_make); + + let actionable = actionable_site_replication_retry_events(&state, now); + assert_eq!(actionable.len(), 1); + assert_eq!(actionable[0].path, "/rustfs/admin/v3/site-replication/peer/bucket-meta"); +} + /// 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)`