diff --git a/.config/e2e-distributed-selection.txt b/.config/e2e-distributed-selection.txt index d99f12976..6cc442542 100644 --- a/.config/e2e-distributed-selection.txt +++ b/.config/e2e-distributed-selection.txt @@ -1,2 +1,2 @@ -sha256-linux=4696a43b167ac608b3b8677027c9fe9fdac3396d37c8cca11dce531c720ac6d2 -sha256-darwin=9785867929047dfd8c6f768e0d2b1e0a8fdba85216f4a4139093b1619d03ff07 +sha256-linux=4a399af73f3a954702967b80649958ae127f4e14560194e193de37e3eac5a17a +sha256-darwin=4a399af73f3a954702967b80649958ae127f4e14560194e193de37e3eac5a17a diff --git a/crates/e2e_test/src/distributed/mod.rs b/crates/e2e_test/src/distributed/mod.rs index 108721f4c..1759440f8 100644 --- a/crates/e2e_test/src/distributed/mod.rs +++ b/crates/e2e_test/src/distributed/mod.rs @@ -27,6 +27,7 @@ mod extra_test; mod harness; mod object_lock_test; mod observability_test; +mod replication_delete_marker_test; mod replication_quota_test; mod s3_basic_test; mod s3_during_data_movement_test; diff --git a/crates/e2e_test/src/distributed/replication_delete_marker_test.rs b/crates/e2e_test/src/distributed/replication_delete_marker_test.rs new file mode 100644 index 000000000..89e7965a6 --- /dev/null +++ b/crates/e2e_test/src/distributed/replication_delete_marker_test.rs @@ -0,0 +1,137 @@ +// Copyright 2026 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Functional REP-105 (rustfs/backlog#2195 item 4): a delete marker created +//! on a multi-node source cluster must replicate to the bucket-replication +//! target. Objects converged in seconds while delete markers did not arrive +//! within 180 s on the shared 3-node functional environment; the single-node +//! e2e never saw it. + +use super::harness::{ + DistCluster, DistLayout, TestResult, enable_versioning, put_bucket_replication, put_object, set_remote_target, unique_bucket, + wait_for_replicated_bytes, wait_until, +}; +use crate::common::{FAST_DATA_USAGE_SCANNER_ENV, RustFSTestEnvironment, init_logging, replication_fast_env, signed_request}; +use crate::replication_extension_test::LOOPBACK_REPLICATION_TARGET_ENV; +use aws_sdk_s3::Client; +use http::{Method, StatusCode}; +use std::time::Duration; + +async fn target_has_delete_marker(client: &Client, bucket: &str, key: &str) -> TestResult { + let versions = client.list_object_versions().bucket(bucket).prefix(key).send().await?; + Ok(versions.delete_markers().iter().any(|marker| marker.key() == Some(key))) +} + +async fn delete_marker_replicates( + source: &DistCluster, + source_bucket: &str, + target_client: &Client, + target_bucket: &str, +) -> TestResult { + let key = "delete-marker/object.bin"; + let body = b"delete marker replication payload".to_vec(); + // Write through one node, delete through another: behind a load + // balancer consecutive requests land on different nodes. + put_object(&source.client(1)?, source_bucket, key, body.clone()).await?; + wait_for_replicated_bytes(target_client, target_bucket, key, &body, Duration::from_secs(60)).await?; + + let delete = source + .client(2)? + .delete_object() + .bucket(source_bucket) + .key(key) + .send() + .await?; + assert_eq!( + delete.delete_marker(), + Some(true), + "a versioned DELETE without versionId must create a marker" + ); + + wait_until( + Duration::from_secs(90), + || async { target_has_delete_marker(target_client, target_bucket, key).await }, + "delete marker replicated to the target bucket", + ) + .await +} + +#[tokio::test] +async fn four_node_bucket_replication_replicates_delete_marker_to_peer_cluster() -> TestResult { + init_logging(); + let (source, target) = DistCluster::start_replication_pair().await?; + let source_bucket = unique_bucket("dm-src"); + let target_bucket = unique_bucket("dm-dst"); + source.create_bucket(&source_bucket).await?; + target.create_bucket(&target_bucket).await?; + enable_versioning(&source.client(0)?, &source_bucket).await?; + enable_versioning(&target.client(0)?, &target_bucket).await?; + + let arn = set_remote_target(&source.cluster, &source_bucket, &target.cluster, &target_bucket).await?; + put_bucket_replication(&source.cluster, &source_bucket, &arn).await?; + + delete_marker_replicates(&source, &source_bucket, &target.client(0)?, &target_bucket).await +} + +/// The functional environment replicates from a 3-node site to a single-node +/// target; keep that shape as its own case. +#[tokio::test] +async fn four_node_bucket_replication_replicates_delete_marker_to_single_node_target() -> TestResult { + init_logging(); + let mut extra: Vec<(&str, &str)> = replication_fast_env(); + extra.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV); + extra.extend_from_slice(FAST_DATA_USAGE_SCANNER_ENV); + let source = DistCluster::start_with_env(DistLayout::FourNodeFourDisk, &extra).await?; + let mut target = RustFSTestEnvironment::new().await?; + target.start_rustfs_server_without_cleanup(vec![]).await?; + + let source_bucket = unique_bucket("dm-src"); + let target_bucket = unique_bucket("dm-dst"); + source.create_bucket(&source_bucket).await?; + let target_client = target.create_s3_client(); + target_client.create_bucket().bucket(&target_bucket).send().await?; + enable_versioning(&source.client(0)?, &source_bucket).await?; + enable_versioning(&target_client, &target_bucket).await?; + + let body = serde_json::json!({ + "endpoint": target.address, + "credentials": { "accessKey": target.access_key, "secretKey": target.secret_key }, + "targetbucket": target_bucket, + "secure": false, + "type": "replication" + }); + let url = format!( + "{}/rustfs/admin/v3/set-remote-target?bucket={}", + source.cluster.nodes[0].url, + urlencoding::encode(&source_bucket) + ); + let response = signed_request( + Method::PUT, + &url, + &source.cluster.access_key, + &source.cluster.secret_key, + Some(body.to_string().into_bytes()), + Some("application/json"), + ) + .await?; + if response.status() != StatusCode::OK { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + return Err(format!("set remote target failed: {status} {body}").into()); + } + let arn: String = serde_json::from_slice(&response.bytes().await?)?; + put_bucket_replication(&source.cluster, &source_bucket, &arn).await?; + + delete_marker_replicates(&source, &source_bucket, &target_client, &target_bucket).await +} diff --git a/crates/e2e_test/src/distributed/site_replication_test.rs b/crates/e2e_test/src/distributed/site_replication_test.rs index 1a5b33e42..a7837d6b8 100644 --- a/crates/e2e_test/src/distributed/site_replication_test.rs +++ b/crates/e2e_test/src/distributed/site_replication_test.rs @@ -126,3 +126,165 @@ async fn four_node_site_replication_replicates_object_to_peer_site() -> TestResu wait_for_replicated_bytes(&site_a.client(3)?, &bucket, reverse_key, &reverse_body, Duration::from_secs(60)).await?; Ok(()) } + +async fn node_admin( + cluster: &crate::common::RustFSTestClusterEnvironment, + node_idx: usize, + method: Method, + path_and_query: &str, + body: Option, +) -> TestResult<(StatusCode, String)> { + crate::common::admin_request( + &cluster.nodes[node_idx].url, + method, + path_and_query, + body, + &cluster.access_key, + &cluster.secret_key, + ) + .await +} + +/// Pair two clusters through site A's first node and wait until both report +/// the two-site topology as enabled. +async fn pair_sites(site_a: &DistCluster, site_b: &DistCluster) -> TestResult { + let sites = vec![ + PeerSite { + name: "site-a".to_string(), + endpoint: site_a.cluster.nodes[0].url.clone(), + access_key: site_a.cluster.access_key.clone(), + secret_key: site_a.cluster.secret_key.clone(), + ..Default::default() + }, + PeerSite { + name: "site-b".to_string(), + endpoint: site_b.cluster.nodes[0].url.clone(), + access_key: site_b.cluster.access_key.clone(), + secret_key: site_b.cluster.secret_key.clone(), + ..Default::default() + }, + ]; + let add_status = site_replication_add(&site_a.cluster, &sites).await?; + assert!( + add_status.success && add_status.err_detail.is_empty() && add_status.initial_sync_error_message.is_empty(), + "site replication add reported failure: {add_status:?}" + ); + wait_for_site_replication_enabled(&site_a.cluster).await?; + wait_for_site_replication_enabled(&site_b.cluster).await?; + Ok(()) +} + +async fn list_users_contains( + cluster: &crate::common::RustFSTestClusterEnvironment, + node_idx: usize, + access_key: &str, +) -> TestResult { + let (status, body) = node_admin(cluster, node_idx, Method::GET, "/rustfs/admin/v3/list-users", None).await?; + if !status.is_success() { + return Err(format!("list-users on node {node_idx} failed: {status} {body}").into()); + } + let users: serde_json::Value = serde_json::from_str(&body)?; + Ok(users.get(access_key).is_some()) +} + +/// backlog#2367 A-7 / functional SITE-102: an IAM change handled by a node +/// other than the one that ran `site-replication/add` must still reach the +/// peer site. Behind a load balancer every admin call may land on a +/// different node, so the coordinator node is not special. +#[tokio::test] +async fn four_node_site_replication_converges_iam_user_created_on_a_non_coordinator_node() -> TestResult { + init_logging(); + let (site_a, site_b) = DistCluster::start_replication_pair().await?; + pair_sites(&site_a, &site_b).await?; + + let user = format!("siteuser-{}", &uuid::Uuid::new_v4().simple().to_string()[..8]); + let body = serde_json::json!({ "secretKey": "siteuser-secret-key-1234", "status": "enabled" }).to_string(); + let (status, response) = node_admin( + &site_a.cluster, + 1, + Method::PUT, + &format!("/rustfs/admin/v3/add-user?accessKey={user}"), + Some(body), + ) + .await?; + assert!(status.is_success(), "add-user on site A node 1 failed: {status} {response}"); + + let site_b_cluster = &site_b.cluster; + let user_ref = user.as_str(); + wait_until( + Duration::from_secs(90), + || async move { list_users_contains(site_b_cluster, 0, user_ref).await }, + "user created on site A node 1 visible on site B", + ) + .await?; + assert!( + list_users_contains(&site_a.cluster, 2, &user).await?, + "the user must be visible on every site A node" + ); + Ok(()) +} + +/// backlog#2367 A-5 / functional SITE-105: a resync started right after +/// pairing must not report buckets as failed. The bucket carrying an +/// operator-configured bucket-replication target to the peer (the shape the +/// functional suite leaves behind) and a plain versioned bucket are both +/// wired by the pairing itself. +#[tokio::test] +async fn four_node_site_replication_resync_start_right_after_pairing_reports_no_failed_bucket() -> TestResult { + init_logging(); + let (site_a, site_b) = DistCluster::start_replication_pair().await?; + + let pre_src = unique_bucket("pre-src"); + let pre_dst = unique_bucket("pre-dst"); + let plain = unique_bucket("plain"); + site_a.create_bucket(&pre_src).await?; + site_b.create_bucket(&pre_dst).await?; + site_a.create_bucket(&plain).await?; + enable_versioning(&site_a.client(0)?, &pre_src).await?; + enable_versioning(&site_b.client(0)?, &pre_dst).await?; + enable_versioning(&site_a.client(0)?, &plain).await?; + let arn = super::harness::set_remote_target(&site_a.cluster, &pre_src, &site_b.cluster, &pre_dst).await?; + super::harness::put_bucket_replication(&site_a.cluster, &pre_src, &arn).await?; + + pair_sites(&site_a, &site_b).await?; + + let (status, info) = node_admin(&site_a.cluster, 1, Method::GET, "/rustfs/admin/v3/site-replication/info", None).await?; + assert!(status.is_success(), "site-replication/info failed: {status} {info}"); + let info: serde_json::Value = serde_json::from_str(&info)?; + let peer = info["sites"] + .as_array() + .and_then(|sites| sites.iter().find(|site| site["name"] == "site-b")) + .cloned() + .ok_or_else(|| format!("site-b peer missing from info: {info}"))?; + + // Through a non-coordinator node, like a load-balanced admin call. + let (status, response) = node_admin( + &site_a.cluster, + 1, + Method::PUT, + "/rustfs/admin/v3/site-replication/resync/op?operation=start", + Some(peer.to_string()), + ) + .await?; + assert!(status.is_success(), "resync start failed: {status} {response}"); + let resync: rustfs_madmin::SRResyncOpStatus = serde_json::from_str(&response)?; + let failed: Vec = resync + .buckets + .iter() + .filter(|bucket| bucket.status == "failed") + .map(|bucket| format!("{}: {}", bucket.bucket, bucket.err_detail)) + .collect(); + assert!( + failed.is_empty(), + "resync right after pairing reported failed buckets: {failed:?} (status={}, detail={})", + resync.status, + resync.err_detail + ); + assert!( + resync.buckets.iter().any(|bucket| bucket.bucket == pre_src) + && resync.buckets.iter().any(|bucket| bucket.bucket == plain), + "both buckets must be part of the resync: {:?}", + resync.buckets + ); + Ok(()) +} diff --git a/crates/replication/src/config.rs b/crates/replication/src/config.rs index 16a258ff7..fecdb8d80 100644 --- a/crates/replication/src/config.rs +++ b/crates/replication/src/config.rs @@ -730,12 +730,16 @@ impl ReplicationConfigurationExt for ReplicationConfiguration { } } + // Highest priority first, like MinIO's `FilterActionableRules`. The + // tie-breakers make this a total order: a comparator that only + // orders same-destination pairs is not transitive, and the standard + // library sort panics on such inputs past its insertion-sort + // threshold (backlog#2367 C-1). rules.sort_by(|a, b| { - if a.destination == b.destination { - b.priority.cmp(&a.priority) - } else { - std::cmp::Ordering::Equal - } + b.priority + .cmp(&a.priority) + .then_with(|| a.destination.bucket.cmp(&b.destination.bucket)) + .then_with(|| a.id.cmp(&b.id)) }); rules @@ -813,24 +817,19 @@ impl ReplicationConfigurationExt for ReplicationConfiguration { return vec![role.to_string()]; } - let mut arns = Vec::new(); - let mut targets_map: HashSet = HashSet::new(); - let rules = self.filter_actionable_rules(obj); - - for rule in rules { + // Rule order (priority descending) is the ARN order: callers that + // iterate targets see the highest-priority destination first. + let mut arns: Vec = Vec::new(); + for rule in self.filter_actionable_rules(obj) { if rule.status == ReplicationRuleStatus::from_static(ReplicationRuleStatus::DISABLED) { continue; } let arn = rule.destination.bucket.trim(); - if !arn.is_empty() && !targets_map.contains(arn) { - targets_map.insert(arn.to_string()); + if !arn.is_empty() && !arns.iter().any(|seen| seen == arn) { + arns.push(arn.to_string()); } } - - for arn in targets_map { - arns.push(arn); - } arns } @@ -1908,6 +1907,84 @@ mod tests { assert_eq!(decisions, vec![(target_a.to_string(), false), (target_b.to_string(), true)]); } + // backlog#2367 C-1: the actionable-rule sort must be a total order. A + // comparator that answers `Equal` for different destinations but orders + // same-destination rules by priority is not transitive, and the standard + // library sort panics on such inputs once the slice is past the + // insertion-sort threshold (> 20 rules). + #[test] + fn actionable_rule_sort_is_a_total_order_across_destinations() { + let targets = ["arn:target:a", "arn:target:b", "arn:target:c"]; + let mut seed: u64 = 0x2367; + for _ in 0..200 { + let rule_count = 21 + (seed % 200) as usize; + let rules = (0..rule_count) + .map(|index| { + seed = seed.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407); + let target = targets[(seed >> 33) as usize % targets.len()]; + delete_marker_rule(&format!("r{index}"), target, "", index as i32, true) + }) + .collect(); + let config = ReplicationConfiguration { + role: String::new(), + rules, + }; + let ordered = config.filter_actionable_rules(&ObjectOpts { + name: "logs/app.log".to_string(), + op_type: ReplicationType::Object, + ..Default::default() + }); + assert_eq!(ordered.len(), rule_count); + assert!( + ordered.windows(2).all(|pair| pair[0].priority >= pair[1].priority), + "actionable rules must be ordered by descending priority" + ); + } + } + + // backlog#2367 C-2: a V1 rule carries its prefix at the top level (no + // ). Ignoring it made `logs/` match every object. + #[test] + fn top_level_rule_prefix_scopes_matching_without_a_filter() { + let arn = "arn:target:a"; + let config = ReplicationConfiguration { + role: String::new(), + rules: vec![delete_marker_rule("v1-prefix", arn, "logs/", 1, true)], + }; + assert_eq!(config.rules[0].prefix(), "logs/"); + + let matching = config.filter_actionable_rules(&ObjectOpts { + name: "logs/app.log".to_string(), + op_type: ReplicationType::Object, + ..Default::default() + }); + assert_eq!(matching.len(), 1); + + let outside = config.filter_actionable_rules(&ObjectOpts { + name: "data/app.log".to_string(), + op_type: ReplicationType::Object, + ..Default::default() + }); + assert!(outside.is_empty(), "an object outside the V1 prefix must not match: {outside:?}"); + assert!( + config + .filter_target_arns(&ObjectOpts { + name: "data/app.log".to_string(), + op_type: ReplicationType::Object, + ..Default::default() + }) + .is_empty() + ); + + // A still wins over the deprecated top-level element. + let mut filtered = delete_marker_rule("filtered", arn, "logs/", 1, true); + filtered.filter = Some(s3s::dto::ReplicationRuleFilter { + prefix: Some("photos/".to_string()), + ..Default::default() + }); + assert_eq!(filtered.prefix(), "photos/"); + } + #[test] fn force_delete_targets_use_overlapping_rules_and_highest_priority_switch() { let target_a = "arn:target:a"; diff --git a/crates/replication/src/rule.rs b/crates/replication/src/rule.rs index 4ca1871a0..60f97e8cc 100644 --- a/crates/replication/src/rule.rs +++ b/crates/replication/src/rule.rs @@ -22,6 +22,10 @@ pub trait ReplicationRuleExt { } impl ReplicationRuleExt for ReplicationRule { + /// The rule's key prefix: `Filter.Prefix`, else `Filter.And.Prefix`, else + /// the deprecated top-level `Prefix` of a V1 rule written without a + /// `` (backlog#2367 C-2). A rule that carries both keeps AWS's + /// precedence: the `` is authoritative. fn prefix(&self) -> &str { if let Some(filter) = &self.filter { if let Some(prefix) = &filter.prefix { @@ -32,7 +36,7 @@ impl ReplicationRuleExt for ReplicationRule { "" } } else { - "" + self.prefix.as_deref().unwrap_or("") } } diff --git a/rustfs/src/admin/handlers/site_replication.rs b/rustfs/src/admin/handlers/site_replication.rs index ad810384f..c288dc047 100644 --- a/rustfs/src/admin/handlers/site_replication.rs +++ b/rustfs/src/admin/handlers/site_replication.rs @@ -3853,6 +3853,30 @@ fn pending_remote_peer_ids(peers: &BTreeMap, local_peer: &Peer .collect() } +/// The peers a pending remove / rotation still has to notify: every remote +/// peer that has not acked, with the local site excluded by the same +/// deployment-id-or-endpoint identity [`pending_remote_peer_ids`] finalizes +/// on. The tick-driven `local_peer` carries the node's own listen address +/// rather than the registered site endpoint (and a handler's carries the +/// request `Host`, which behind a load balancer differs too), so an +/// endpoint-only check dialed the site itself, timed out against the +/// lifecycle lock this very request holds, and reported the operation as +/// `Partial` (backlog#2367 A-4). +fn pending_peers_awaiting_notification<'a>( + peers: &'a BTreeMap, + local_peer: &PeerInfo, + acked_deployment_ids: &BTreeSet, +) -> Vec<&'a PeerInfo> { + peers + .values() + .filter(|peer| { + peer.deployment_id != local_peer.deployment_id + && !same_identity_endpoint(&peer.endpoint, &local_peer.endpoint) + && !acked_deployment_ids.contains(&peer.deployment_id) + }) + .collect() +} + fn pending_all_remote_peers_acked( peers: &BTreeMap, local_peer: &PeerInfo, @@ -4068,12 +4092,7 @@ async fn drive_pending_rotation(pending: &PendingRotation, local_peer: &PeerInfo }; let mut peer_errors = Vec::new(); - for peer in pending.peers.values() { - if same_identity_endpoint(&peer.endpoint, &local_peer.endpoint) - || pending.acked_deployment_ids.contains(&peer.deployment_id) - { - continue; - } + for peer in pending_peers_awaiting_notification(&pending.peers, local_peer, &pending.acked_deployment_ids) { // A superseded join returns BEFORE `apply_iam`, so a no-op answer // means the peer never installed the new secret. Acking it would // finalize a rotation half the mesh cannot authenticate against @@ -4376,12 +4395,9 @@ async fn drive_pending_remove(pending_remove: &PendingRemove, local_peer: &PeerI if secret_candidates.is_empty() { peer_errors.push("site replication service account secret unavailable".to_string()); } else { - for peer in pending_remove.original_peers.values() { - if same_identity_endpoint(&peer.endpoint, &local_peer.endpoint) - || pending_remove.acked_deployment_ids.contains(&peer.deployment_id) - { - continue; - } + for peer in + pending_peers_awaiting_notification(&pending_remove.original_peers, local_peer, &pending_remove.acked_deployment_ids) + { if let Err(err) = PeerAdminRequest::put( &runtime_peer_connection(peer)?, SITE_REPLICATION_PEER_REMOVE_PATH, @@ -4931,7 +4947,7 @@ async fn ensure_site_replication_bucket_targets(bucket: &str) -> S3Result<()> { return Ok(()); }; let config = bucket_replication_config_for_target_refresh(bucket).await?; - ensure_site_replication_bucket_targets_with_runtime( + let written = ensure_site_replication_bucket_targets_with_runtime( bucket, &runtime.state, &runtime.local_peer, @@ -4939,7 +4955,11 @@ async fn ensure_site_replication_bucket_targets(bucket: &str) -> S3Result<()> { &runtime.service_account_secret_key, expected_incarnation_id, ) - .await + .await?; + if written { + reload_bucket_metadata_on_peers(bucket, "site_replication_bucket_targets", false).await; + } + Ok(()) } async fn ensure_site_replication_bucket_setup(bucket: &str) -> S3Result { @@ -5009,6 +5029,9 @@ async fn cleanup_removed_site_replication_bucket(bucket: &str, removed_deploymen Err(err) => return Err(ApiError::from(err).into()), } + if removed > 0 { + reload_bucket_metadata_on_peers(bucket, "site_replication_bucket_cleanup", true).await; + } Ok(removed) } @@ -5325,7 +5348,7 @@ async fn refresh_bucket_targets_after_endpoint_edit(pending_id: &str, service_ac let local_peer = current_local_runtime_peer(&target_state); let _targets_guard = lock_bucket_targets_metadata(&bucket.name).await; let replication_config = bucket_replication_config_for_target_refresh(&bucket.name).await?; - ensure_site_replication_bucket_targets_with_runtime( + let written = ensure_site_replication_bucket_targets_with_runtime( &bucket.name, &target_state, &local_peer, @@ -5334,6 +5357,9 @@ async fn refresh_bucket_targets_after_endpoint_edit(pending_id: &str, service_ac expected_incarnation_id, ) .await?; + if written { + reload_bucket_metadata_on_peers(&bucket.name, "site_replication_endpoint_refresh", false).await; + } rewritten.push(bucket.name.clone()); @@ -5377,16 +5403,12 @@ async fn site_bucket_resync_manifest_entry(bucket: &str, peer: &PeerInfo, now: O ..Default::default() }; let _targets_guard = lock_bucket_targets_metadata(bucket).await; - let (config, _) = match metadata_sys::get_replication_config(bucket).await { - Ok(config) => config, - Err(err) => { - entry.status = "failed".to_string(); - entry.err_detail = summarize_peer_error_detail(&err.to_string()); - return entry; - } - }; - let targets = match metadata_sys::list_bucket_targets(bucket).await { - Ok(targets) => targets, + // Read what is persisted, not this node's cache: the wiring may have + // been written by another node moments ago (`start_site_bucket_resync` + // already reads its targets from disk), and an operator resync must see + // the same records the drive will use. + let (config, targets) = match site_bucket_resync_persisted_wiring(bucket).await { + Ok(wiring) => wiring, Err(err) => { entry.status = "failed".to_string(); entry.err_detail = summarize_peer_error_detail(&err.to_string()); @@ -5417,6 +5439,15 @@ async fn site_bucket_resync_manifest_entry(bucket: &str, peer: &PeerInfo, now: O entry } +/// The persisted replication configuration and bucket targets, bypassing the +/// node-local metadata cache. `ConfigNotFound` surfaces for a bucket without +/// a replication configuration, matching the cached read's error. +async fn site_bucket_resync_persisted_wiring(bucket: &str) -> Result<(ReplicationConfiguration, BucketTargets), StorageError> { + let metadata = metadata_sys::get_config_from_disk(bucket).await?; + let config = metadata.replication_config.ok_or(StorageError::ConfigNotFound)?; + Ok((config, metadata.bucket_target_config.unwrap_or_default())) +} + async fn start_site_bucket_resync(bucket: &str, target_arn: &str, resync_id: &str) -> ResyncBucketStatus { let mut bucket_status = ResyncBucketStatus { bucket: bucket.to_string(), @@ -5439,17 +5470,8 @@ async fn start_site_bucket_resync(bucket: &str, target_arn: &str, resync_id: &st } }; - let (config, _) = match metadata_sys::get_replication_config(bucket).await { - Ok(config) => config, - Err(err) => { - bucket_status.status = "failed".to_string(); - bucket_status.err_detail = err.to_string(); - return bucket_status; - } - }; - - let targets = match metadata_sys::list_bucket_targets_from_disk(bucket).await { - Ok(targets) => targets, + let (config, targets) = match site_bucket_resync_persisted_wiring(bucket).await { + Ok(wiring) => wiring, Err(err) => { bucket_status.status = "failed".to_string(); bucket_status.err_detail = err.to_string(); @@ -6046,6 +6068,10 @@ async fn apply_bucket_meta_item(item: SRBucketMeta) -> S3Result<()> { drop(lifecycle_guard); drop(targets_guard); + if !skip_config_write { + reload_bucket_metadata_on_peers(&item.bucket, "site_replication_bucket_meta", item.r#type == "lc-config").await; + } + if item.r#type == "replication-config" { // Rebuild the local outbound rules too: a site that joined an already-replicated // bucket receives this item before it has any `site-repl-*` rule of its own. @@ -7432,6 +7458,7 @@ impl Operation for SRPeerBucketOpsHandler { ) .await .map_err(ApiError::from)?; + reload_bucket_metadata_on_peers(&bucket, "site_replication_make_bucket", false).await; } "configure-replication" => { store @@ -15434,4 +15461,54 @@ mod tests { "no replicated config write may bypass the source stamp" ); } + + /// backlog#2367 A-4: `remove --all` notified "the peer" at the site's own + /// registered endpoint. The tick-driven local peer carries the node's + /// listen address, so an endpoint-only self check let the loop dial the + /// site itself and report `Partial: failed to notify 1 peer(s)`. + #[test] + fn pending_notifications_skip_the_local_site_by_deployment_id() { + let local_registered = PeerInfo { + deployment_id: "site-b".to_string(), + ..peer("site-b", "http://site-b.example.com:9000") + }; + let remote = PeerInfo { + deployment_id: "site-a".to_string(), + ..peer("site-a", "http://site-a.example.com:9000") + }; + let acked = PeerInfo { + deployment_id: "site-c".to_string(), + ..peer("site-c", "http://site-c.example.com:9000") + }; + let peers = BTreeMap::from([ + (local_registered.deployment_id.clone(), local_registered.clone()), + (remote.deployment_id.clone(), remote), + (acked.deployment_id.clone(), acked.clone()), + ]); + let acked_ids = BTreeSet::from([acked.deployment_id]); + + // The tick resolves the local peer from its own listen address. + let local_from_tick = PeerInfo { + deployment_id: "site-b".to_string(), + ..peer("site-b", "http://127.0.0.1:9000") + }; + let to_notify: Vec<&str> = pending_peers_awaiting_notification(&peers, &local_from_tick, &acked_ids) + .iter() + .map(|peer| peer.deployment_id.as_str()) + .collect(); + assert_eq!(to_notify, vec!["site-a"], "the local site and the acked peer are never dialed"); + + // Identity stays consistent with what finalization waits for. + assert_eq!( + pending_remote_peer_ids(&peers, &local_from_tick), + BTreeSet::from(["site-a".to_string(), "site-c".to_string()]) + ); + + // A handler-resolved local peer (registered endpoint) agrees. + let to_notify: Vec<&str> = pending_peers_awaiting_notification(&peers, &local_registered, &acked_ids) + .iter() + .map(|peer| peer.deployment_id.as_str()) + .collect(); + assert_eq!(to_notify, vec!["site-a"]); + } } diff --git a/rustfs/src/admin/handlers/system.rs b/rustfs/src/admin/handlers/system.rs index 062159287..165baf70c 100644 --- a/rustfs/src/admin/handlers/system.rs +++ b/rustfs/src/admin/handlers/system.rs @@ -73,6 +73,7 @@ const SITE_REPLICATION_RESYNC_ROUTE: &str = "/rustfs/admin/v3/site-replication/r const SITE_REPLICATION_REPAIR_ROUTE: &str = "/rustfs/admin/v3/site-replication/repair"; const SITE_REPLICATION_REPAIR_STATUS_ROUTE: &str = "/rustfs/admin/v3/site-replication/repair/status"; const IAM_POLICY_ATTACH_ROUTE: &str = "/rustfs/admin/v3/idp/builtin/policy/attach"; +const DATA_USAGE_INFO_ROUTE: &str = "/rustfs/admin/v3/datausageinfo"; const IAM_POLICY_DETACH_ROUTE: &str = "/rustfs/admin/v3/idp/builtin/policy/detach"; const IAM_POLICY_ENTITIES_ROUTE: &str = "/rustfs/admin/v3/idp/builtin/policy-entities"; const IAM_ACCESS_KEYS_BULK_ROUTE: &str = "/rustfs/admin/v3/list-access-keys-bulk"; @@ -1077,12 +1078,25 @@ fn advertised_admin_capabilities() -> Vec { ("admin.account.mfa", HttpMethod::Get, ACCOUNT_MFA_ROUTE), ("admin.mfa.challenge", HttpMethod::Get, MFA_CHALLENGE_ROUTE), ("admin.user.mfa", HttpMethod::Get, USER_MFA_ROUTE), + // `rc du` is gated on this name. Before it was advertised the client + // inferred it from a `1.0.0-rc.` version prefix, which no longer + // matches once the server reports `1.0.0` (backlog#2367 E-2). + ("admin.data-usage", HttpMethod::Get, DATA_USAGE_INFO_ROUTE), ] .into_iter() .map(|(name, method, route)| AdvertisedAdminCapability { name, status: admin_route_capability(method, route), }) + .chain(std::iter::once(AdvertisedAdminCapability { + // `rc watch` streams `GET /{bucket}?events=`, a misc extension route + // dispatched by `admin::router` rather than an admin policy route, + // so its status is not an inventory lookup (same version-prefix + // inference on the client as `admin.data-usage`). + name: "listen_notification", + status: CapabilityStatus::supported() + .with_reason("bucket listen notification (?events=) is dispatched by the admin router"), + })) .collect() } @@ -1258,6 +1272,9 @@ mod tests { "admin.iam.access-keys-bulk", "admin.iam.access-keys-bulk.ldap", "admin.iam.access-keys-bulk.openid", + // rc pinned these two by version prefix until 1.0.0 (backlog#2367 E-2). + "admin.data-usage", + "listen_notification", ]; for name in expected_supported { let entry = response diff --git a/rustfs/src/site_replication/hooks.rs b/rustfs/src/site_replication/hooks.rs index 2af6aa000..b907273a3 100644 --- a/rustfs/src/site_replication/hooks.rs +++ b/rustfs/src/site_replication/hooks.rs @@ -1630,6 +1630,44 @@ pub(crate) fn build_site_replication_config( } } +/// Reload `bucket`'s metadata on every other node of this site after a +/// site-replication write. Every S3 bucket-config write does this +/// (`app::bucket_usecase::notify_bucket_metadata_reload`); the +/// site-replication writers did not, so on a multi-node site a node other +/// than the one that applied the write served the previous targets and +/// rules for up to the 15-minute refresh — a `resync start` routed to such a +/// node reported every freshly wired bucket as `Config not found` or +/// `recorded remote target no longer exists` (backlog#2367 A-5, backlog#2195 +/// item 2). Best effort like the S3 path: the write is durable and the +/// refresh loop is the fallback, so an unreachable node must not fail the +/// operation that already committed. +pub(crate) async fn reload_bucket_metadata_on_peers(bucket: &str, operation: &'static str, scanner_maintenance_change: bool) { + if scanner_maintenance_change { + rustfs_scanner::record_scanner_maintenance_change(bucket); + } + let Some(notification_sys) = crate::admin::runtime_sources::current_notification_system() else { + return; + }; + let result = if scanner_maintenance_change { + notification_sys.load_bucket_metadata_for_scanner_maintenance(bucket).await + } else { + notification_sys.load_bucket_metadata(bucket).await + }; + if let Err(err) = result { + warn!( + event = EVENT_ADMIN_SITE_REPLICATION_STATE, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SITE_REPLICATION, + bucket = %bucket, + operation, + result = "peer_metadata_reload_failed", + error = %err, + "admin site replication state" + ); + } +} + +/// Returns whether the bucket targets were rewritten. pub(crate) async fn ensure_site_replication_bucket_targets_with_runtime( bucket: &str, state: &SiteReplicationState, @@ -1637,7 +1675,7 @@ pub(crate) async fn ensure_site_replication_bucket_targets_with_runtime( config: Option<&ReplicationConfiguration>, service_account_secret_key: &str, expected_incarnation_id: Uuid, -) -> S3Result<()> { +) -> S3Result { let existing = match metadata_sys::list_bucket_targets(bucket).await { Ok(targets) => targets, Err(StorageError::ConfigNotFound) => BucketTargets::default(), @@ -1649,7 +1687,7 @@ pub(crate) async fn ensure_site_replication_bucket_targets_with_runtime( let updated = reconcile_site_replication_bucket_targets(existing, bucket, state, local_peer, config, service_account_secret_key)?; if updated.targets.is_empty() { - return Ok(()); + return Ok(false); } let json_targets = serde_json::to_vec(&updated) @@ -1658,12 +1696,12 @@ pub(crate) async fn ensure_site_replication_bucket_targets_with_runtime( // client — noticeable now that startup reconciles all buckets, not just the one bucket // an operation touched. if json_targets == existing_json { - return Ok(()); + return Ok(false); } metadata_sys::update_if_incarnation(bucket, BUCKET_TARGETS_FILE, json_targets, expected_incarnation_id) .await .map_err(ApiError::from)?; - Ok(()) + Ok(true) } pub(crate) async fn bucket_replication_config_for_target_refresh(bucket: &str) -> S3Result> { @@ -1674,13 +1712,14 @@ pub(crate) async fn bucket_replication_config_for_target_refresh(bucket: &str) - } } +/// Returns whether the replication configuration was rewritten. pub(crate) async fn ensure_site_replication_bucket_replication_config_with_runtime( bucket: &str, state: &SiteReplicationState, local_peer: &PeerInfo, service_account_secret_key: &str, expected_incarnation_id: Uuid, -) -> S3Result<()> { +) -> S3Result { let existing = match metadata_sys::get_replication_config(bucket).await { Ok((existing, _)) => Some(existing), Err(StorageError::ConfigNotFound) => None, @@ -1689,7 +1728,7 @@ pub(crate) async fn ensure_site_replication_bucket_replication_config_with_runti let Some(desired) = build_site_replication_config(bucket, state, local_peer, service_account_secret_key, existing.as_ref())? else { - return Ok(()); + return Ok(false); }; // Derived rules are state owned by this site: rebuild them from the current peer @@ -1721,7 +1760,7 @@ pub(crate) async fn ensure_site_replication_bucket_replication_config_with_runti }; if rules == existing_rules && role == existing_role { - return Ok(()); + return Ok(false); } let data = serialize(&ReplicationConfiguration { role, rules }) @@ -1730,7 +1769,7 @@ pub(crate) async fn ensure_site_replication_bucket_replication_config_with_runti .await .map_err(ApiError::from)?; - Ok(()) + Ok(true) } pub(crate) async fn ensure_site_replication_bucket_setup_with_runtime( @@ -1748,9 +1787,9 @@ pub(crate) async fn ensure_site_replication_bucket_setup_with_runtime_for_incarn runtime: &SiteReplicationRuntime, expected_incarnation_id: Uuid, ) -> S3Result<()> { - let _targets_guard = lock_bucket_targets_metadata(bucket).await; + let targets_guard = lock_bucket_targets_metadata(bucket).await; let config = bucket_replication_config_for_target_refresh(bucket).await?; - ensure_site_replication_bucket_targets_with_runtime( + let targets_written = ensure_site_replication_bucket_targets_with_runtime( bucket, &runtime.state, &runtime.local_peer, @@ -1759,7 +1798,7 @@ pub(crate) async fn ensure_site_replication_bucket_setup_with_runtime_for_incarn expected_incarnation_id, ) .await?; - ensure_site_replication_bucket_replication_config_with_runtime( + let config_written = ensure_site_replication_bucket_replication_config_with_runtime( bucket, &runtime.state, &runtime.local_peer, @@ -1767,6 +1806,10 @@ pub(crate) async fn ensure_site_replication_bucket_setup_with_runtime_for_incarn expected_incarnation_id, ) .await?; + drop(targets_guard); + if targets_written || config_written { + reload_bucket_metadata_on_peers(bucket, "site_replication_bucket_setup", config_written).await; + } Ok(()) } @@ -1791,6 +1834,7 @@ pub(crate) async fn ensure_site_replication_bucket_versioning(bucket: &str) -> S metadata_sys::update_if_incarnation(bucket, BUCKET_VERSIONING_CONFIG, bucket_versioning_xml()?, expected_incarnation_id) .await .map_err(ApiError::from)?; + reload_bucket_metadata_on_peers(bucket, "site_replication_bucket_versioning", false).await; Ok(()) } diff --git a/rustfs/src/site_replication/retry.rs b/rustfs/src/site_replication/retry.rs index 738a645d6..383cb15f7 100644 --- a/rustfs/src/site_replication/retry.rs +++ b/rustfs/src/site_replication/retry.rs @@ -52,7 +52,11 @@ pub(crate) struct SiteReplicationRetryEvent { /// 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 + /// entry. Every entry this binary creates starts recorded: the IAM + /// change hook records deletion bodies, and the other creators (the add + /// bootstrap's snapshot send, the drain's own replay) never carry a + /// deletion. A legacy entry persisted by a binary that predates recording + /// (serde default `false`), 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, @@ -308,7 +312,12 @@ fn push_site_replication_retry_event( updated_at: Some(OffsetDateTime::now_utc()), edit_generation: generation, peer_unreachable, - deletions_recorded: false, + // See the field doc: only a row persisted by an older binary is + // unrecorded. Stamping at creation is what lets an entry first + // created by the bootstrap snapshot send settle after a later + // deletion is replayed, instead of escalating forever + // (backlog#2367 A-3). + deletions_recorded: true, }); Ok(evicted) } @@ -598,22 +607,7 @@ pub(crate) fn record_failed_iam_delivery( item: &SRIAMItem, error: &str, ) -> S3Result<()> { - 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 Ok(()); @@ -1418,6 +1412,33 @@ pub(crate) fn site_replication_retry_backoff_elapsed(event: &SiteReplicationRetr now.unix_timestamp().saturating_sub(updated_at.unix_timestamp()) >= delay } +/// Backoff evaluation time for the heavyweight tick: halfway to the next +/// tick. Backoffs are multiples of the tick interval, so an entry stamped δ +/// seconds after a tick is `600 − δ` old at the next one and slipped a whole +/// extra interval for every δ > 0 — a first replay landed at T+1200 rather +/// than T+600 (backlog#2367 A-1). Evaluating at the midpoint bounds the slip +/// to half an interval either way; timestamps written back stay real time. +pub(crate) fn heavyweight_retry_drain_horizon(now: OffsetDateTime) -> OffsetDateTime { + let half_interval = crate::site_replication_reconcile::RECONCILE_INTERVAL / 2; + now + time::Duration::seconds(i64::try_from(half_interval.as_secs()).unwrap_or(i64::MAX)) +} + +/// What the lightweight 30-second pass may act on. It replays bounded bucket +/// ops only, but probes every backed-off class: promotion is a state flip +/// the heavyweight tick then replays, so an IAM or bucket-metadata snapshot +/// owed to a peer that came back is resent at the next tick instead of +/// after its own backoff has fully elapsed (backlog#2367 A-1). +pub(crate) fn lightweight_retry_drain_partition( + state: &SiteReplicationState, + now: OffsetDateTime, +) -> (Vec, Vec) { + let mut actionable = actionable_site_replication_retry_events(state, now); + actionable.retain(|event| { + classify_site_replication_retry_event(event).is_some_and(|action| is_lightweight_retry_drain_action(&action)) + }); + (actionable, deferred_site_replication_retry_events(state, now)) +} + /// The subset of the retry queue the background drain is allowed to touch. pub(crate) fn actionable_site_replication_retry_events( state: &SiteReplicationState, @@ -1666,14 +1687,7 @@ async fn drain_site_replication_retry_queue_lightweight_inner() -> S3Result<()> return Ok(()); } let now = OffsetDateTime::now_utc(); - let mut actionable = actionable_site_replication_retry_events(&runtime.state, now); - let mut deferred = deferred_site_replication_retry_events(&runtime.state, now); - actionable.retain(|event| { - classify_site_replication_retry_event(event).is_some_and(|action| is_lightweight_retry_drain_action(&action)) - }); - deferred.retain(|event| { - classify_site_replication_retry_event(event).is_some_and(|action| is_lightweight_retry_drain_action(&action)) - }); + let (actionable, deferred) = lightweight_retry_drain_partition(&runtime.state, now); if actionable.is_empty() && deferred.is_empty() { return Ok(()); } @@ -1697,10 +1711,7 @@ async fn drain_site_replication_retry_queue_lightweight_inner() -> S3Result<()> return Ok(()); } let now = OffsetDateTime::now_utc(); - let mut actionable = actionable_site_replication_retry_events(&runtime.state, now); - actionable.retain(|event| { - classify_site_replication_retry_event(event).is_some_and(|action| is_lightweight_retry_drain_action(&action)) - }); + let (actionable, _) = lightweight_retry_drain_partition(&runtime.state, now); if actionable.is_empty() { return Ok(()); } @@ -1717,9 +1728,9 @@ pub(crate) async fn drain_site_replication_retry_queue_inner() -> S3Result<()> { // 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 actionable = actionable_site_replication_retry_events(&runtime.state, now); - let deferred = deferred_site_replication_retry_events(&runtime.state, now); + let horizon = heavyweight_retry_drain_horizon(OffsetDateTime::now_utc()); + let actionable = actionable_site_replication_retry_events(&runtime.state, horizon); + let deferred = deferred_site_replication_retry_events(&runtime.state, horizon); if actionable.is_empty() && deferred.is_empty() { return Ok(()); } @@ -1764,8 +1775,8 @@ pub(crate) async fn drain_site_replication_retry_queue_inner() -> S3Result<()> { { return Ok(()); } - let now = OffsetDateTime::now_utc(); - let actionable = actionable_site_replication_retry_events(&runtime.state, now); + let horizon = heavyweight_retry_drain_horizon(OffsetDateTime::now_utc()); + let actionable = actionable_site_replication_retry_events(&runtime.state, horizon); if actionable.is_empty() { return Ok(()); } diff --git a/rustfs/src/site_replication/tests.rs b/rustfs/src/site_replication/tests.rs index b69b1d3d7..42568392c 100644 --- a/rustfs/src/site_replication/tests.rs +++ b/rustfs/src/site_replication/tests.rs @@ -845,7 +845,8 @@ fn test_record_failed_iam_delivery_records_deletions_and_flags_entry() { record_failed_iam_delivery(&mut state, &target, &policy_delete_item("readonly"), "peer offline").expect("record failure"); assert_eq!(state.iam_deletion_replays.len(), 2); - // A legacy entry (created without recording) is never stamped. + // A legacy entry (persisted by a binary that predates recording, so it + // deserialized with the `false` default) is never stamped. let legacy = PeerInfo { deployment_id: "legacy-dep".to_string(), ..peer("legacy", "https://legacy.example.com") @@ -859,6 +860,12 @@ fn test_record_failed_iam_delivery_records_deletions_and_flags_entry() { None, ) .expect("upsert retry event"); + state + .retry_queue + .iter_mut() + .find(|event| event.peer_deployment_id == legacy.deployment_id) + .expect("legacy entry") + .deletions_recorded = false; record_failed_iam_delivery(&mut state, &legacy, &user_delete_item("bob"), "peer offline").expect("record failure"); let legacy_event = state .retry_queue @@ -871,6 +878,43 @@ fn test_record_failed_iam_delivery_records_deletions_and_flags_entry() { ); } +/// backlog#2367 A-3: an entry first created by a non-deletion failure — the +/// add bootstrap's snapshot send, or the drain's own replay — hides no +/// unrecorded deletion, so a deletion recorded later plus a stable snapshot +/// resend must settle it instead of escalating it to the permanent marker +/// that only `replicate repair` clears. +#[test] +fn test_bootstrap_created_iam_entry_settles_after_deletion_replay() { + let target = PeerInfo { + deployment_id: "remote-dep".to_string(), + ..peer("remote", "https://remote.example.com") + }; + let mut state = deletion_replay_state(&target); + upsert_site_replication_retry_event( + &mut state.retry_queue, + &target, + SITE_REPLICATION_PEER_IAM_ITEM_WIRE_PATH, + "peer request to https://remote.example.com failed (connect): connection refused", + None, + ) + .expect("bootstrap send failure"); + assert!(state.retry_queue[0].deletions_recorded, "a fresh entry carries no unrecorded deletion"); + + record_failed_iam_delivery(&mut state, &target, &user_delete_item("alice"), "peer offline").expect("record failure"); + assert_eq!(state.retry_queue.len(), 1, "the hook failure collapses into the bootstrap entry"); + assert!(state.retry_queue[0].deletions_recorded); + assert_eq!(state.iam_deletion_replays.len(), 1); + + let observed = state.retry_queue[0].clone(); + let replayed: Vec = state.iam_deletion_replays.iter().map(|record| record.id.clone()).collect(); + assert!( + settle_replayed_iam_retry_events(&mut state, &target, &observed, &replayed), + "the replayed deletion plus the snapshot resend settle the entry" + ); + assert!(state.retry_queue.is_empty(), "no escalation marker may remain: {:?}", state.retry_queue); + assert!(state.iam_deletion_replays.is_empty()); +} + /// 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. @@ -1626,6 +1670,74 @@ fn test_deferred_retry_events_do_not_probe_fresh_application_failures() { assert!(actionable_site_replication_retry_events(&state, now).is_empty()); } +/// backlog#2367 A-1: the lightweight pass replays bucket ops only, but +/// probes every backed-off class so a recovered peer's IAM snapshot is +/// promoted within 30 seconds instead of waiting for the heavyweight tick +/// to notice it. +#[test] +fn test_lightweight_partition_probes_snapshot_entries_but_replays_bucket_ops_only() { + 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"; + let mut iam_unreachable = drain_event( + "remote", + SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH, + 3, + Some(now - time::Duration::seconds(30)), + ); + iam_unreachable.peer_unreachable = true; + let mut bucket_unreachable = drain_event("remote", bucket_make, 3, Some(now - time::Duration::seconds(30))); + bucket_unreachable.peer_unreachable = true; + state.retry_queue = vec![ + iam_unreachable, + bucket_unreachable, + // Already promoted (or never stamped): due now. + drain_event("remote", SITE_REPLICATION_RETRY_BUCKET_METADATA_SNAPSHOT_PATH, 1, None), + drain_event("remote", bucket_make, 1, None), + ]; + + let (actionable, deferred) = lightweight_retry_drain_partition(&state, now); + let deferred_paths: Vec<&str> = deferred.iter().map(|event| event.path.as_str()).collect(); + assert!( + deferred_paths.contains(&SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH), + "the backed-off IAM snapshot must be probed by the lightweight pass: {deferred_paths:?}" + ); + assert!(deferred_paths.contains(&bucket_make)); + assert_eq!( + actionable.iter().map(|event| event.path.as_str()).collect::>(), + vec![bucket_make], + "only the bounded bucket op is replayed by the lightweight pass" + ); +} + +/// backlog#2367 A-1: the heavyweight tick evaluates backoff halfway to its +/// next tick. A first failure stamped one second after a tick is 599 s old +/// at the next tick; without the horizon it slipped to the tick after. +#[test] +fn test_heavyweight_horizon_absorbs_tick_phase() { + let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp"); + let horizon = heavyweight_retry_drain_horizon(now); + assert_eq!(horizon - now, time::Duration::seconds(300)); + + let elapsed_at_horizon = |secs_ago: i64| { + site_replication_retry_backoff_elapsed( + &drain_event("remote", "/p", 1, Some(now - time::Duration::seconds(secs_ago))), + horizon, + ) + }; + // Stamped just after the previous tick: due at this tick, not the next. + assert!(elapsed_at_horizon(599)); + // Due before the next tick's midpoint: drained now rather than a whole + // interval late. + assert!(elapsed_at_horizon(301)); + // Due after the midpoint: waits for the next tick. + assert!(!elapsed_at_horizon(299)); +} + /// 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)` diff --git a/rustfs/src/site_replication_reconcile.rs b/rustfs/src/site_replication_reconcile.rs index b4ea52c93..89fe2465a 100644 --- a/rustfs/src/site_replication_reconcile.rs +++ b/rustfs/src/site_replication_reconcile.rs @@ -32,7 +32,7 @@ use tokio::time::Instant; use tokio_util::sync::CancellationToken; use tracing::warn; -const RECONCILE_INTERVAL: Duration = Duration::from_secs(600); +pub(crate) const RECONCILE_INTERVAL: Duration = Duration::from_secs(600); pub(crate) const RETRY_DRAIN_INTERVAL: Duration = Duration::from_secs(30); /// A reconciler reports its own failures; the outcome carries no value because neither