From 8fa08b139bbca141135469e9e6969fa3447f77b1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=94=90=E5=B0=8F=E9=B8=AD?= Date: Sat, 12 Sep 2026 15:54:22 +0800 Subject: [PATCH] fix(site-replication): do not inherit a bucket-level reset id when taking over an operator target When site replication wires a bucket for a peer, `reconcile_site_replication_bucket_targets` takes over an existing operator target that already points at that peer and rewrites it under the site ARN. It also copied the target's `reset_id` and `reset_before_date`. Those name a bucket-level `replication-reset` keyed by the operator ARN, which the site resync status map never sees, so `start_site_bucket_resync` found a foreign reset id with no matching status and reported the bucket as `target belongs to a different active resync operation` on every site resync, with no cancel path to clear it (rustfs/backlog#2479, rep107-src in the SITE-105 run). Carry the resync identity over only when the ARN is unchanged, i.e. when an SR target is re-reconciled under its own ARN. Operator tuning (bandwidth, proxy, storage class, health-check interval) still carries over. - Unit test covers the takeover (reset id dropped) and the same-ARN re-reconcile (reset id kept). - e2e `test_site_replication_resync_after_bucket_level_reset_to_same_peer` runs a bucket-level reset against the future peer, joins the sites, and requires the site resync to start without a conflict and complete; the repl-nightly selection digest is refreshed for the new case. --- .config/e2e-repl-nightly-selection.txt | 2 +- .../src/replication_extension_test.rs | 138 ++++++++++++++++++ rustfs/src/site_replication/hooks.rs | 13 +- rustfs/src/site_replication/tests.rs | 74 ++++++++++ 4 files changed, 224 insertions(+), 3 deletions(-) diff --git a/.config/e2e-repl-nightly-selection.txt b/.config/e2e-repl-nightly-selection.txt index 6626a2217..8e0b0b598 100644 --- a/.config/e2e-repl-nightly-selection.txt +++ b/.config/e2e-repl-nightly-selection.txt @@ -1 +1 @@ -sha256=0e338d305260229e17ccfb2adc48a6212dbdfea36a9ebfb5a4e0d38658e6cc45 +sha256=fa9281eb620cec40bfc68e62eab2773cd313f8ce1497d808ff813cfe5aa243ec diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index bebecff9c..dda494ec6 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/crates/e2e_test/src/replication_extension_test.rs @@ -6134,6 +6134,144 @@ async fn test_site_replication_allows_private_ca_https_with_ca_cert_pem_real_dua Ok(()) } +/// rustfs/backlog#2479: a bucket that had a bucket-level `replication-reset` +/// against a target on the future peer must still resync once the sites are +/// joined. Site replication takes that operator target over under its own +/// ARN; the old `reset_id` used to travel with it and every site resync then +/// refused the bucket as "owned by a different active resync operation". +#[tokio::test] +async fn test_site_replication_resync_after_bucket_level_reset_to_same_peer() -> Result<(), Box> { + init_logging(); + let process_env = [ + ("RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET", "true"), + ("RUSTFS_REPL_RESYNC_POLL_MAX_MS", "100"), + ("RUST_LOG", "error"), + ]; + + let mut source_env = RustFSTestEnvironment::new().await?; + source_env.start_rustfs_server_with_env(vec![], &process_env).await?; + let mut target_env = RustFSTestEnvironment::new().await?; + target_env.start_rustfs_server_with_env(vec![], &process_env).await?; + + let source_bucket = "site-repl-after-reset-src"; + let operator_target_bucket = "site-repl-after-reset-dst"; + let key = "preexisting.bin"; + let source_client = source_env.create_s3_client(); + let target_client = target_env.create_s3_client(); + + source_client.create_bucket().bucket(source_bucket).send().await?; + enable_bucket_versioning(&source_env, source_bucket).await?; + target_client.create_bucket().bucket(operator_target_bucket).send().await?; + enable_bucket_versioning(&target_env, operator_target_bucket).await?; + + // Operator-configured bucket replication to the future peer, then an + // existing-object resync that stamps the target with a reset id. + let operator_arn = set_replication_target(&source_env, source_bucket, &target_env, operator_target_bucket).await?; + put_bucket_replication(&source_env, source_bucket, &operator_arn).await?; + source_client + .put_object() + .bucket(source_bucket) + .key(key) + .body(ByteStream::from(vec![b'p'; 4096])) + .send() + .await?; + wait_for_object_on_target(&target_client, operator_target_bucket, key).await?; + let (reset_arn, reset_id) = start_bucket_replication_reset(&source_env, source_bucket).await?; + assert_eq!(reset_arn, operator_arn, "the reset must target the operator ARN"); + wait_for_replication_reset_target(&source_env, source_bucket, &operator_arn, |target| { + target.reset_id == reset_id && matches!(target.status.as_str(), "Completed" | "Failed") + }) + .await?; + + let add_status = site_replication_add( + &source_env, + &[ + PeerSite { + name: "source-site".to_string(), + endpoint: source_env.url.clone(), + access_key: source_env.access_key.clone(), + secret_key: source_env.secret_key.clone(), + ..Default::default() + }, + PeerSite { + name: "target-site".to_string(), + endpoint: target_env.url.clone(), + access_key: target_env.access_key.clone(), + secret_key: target_env.secret_key.clone(), + ..Default::default() + }, + ], + ) + .await?; + assert!(add_status.success, "unexpected site add result: {:?}", add_status); + let source_info = wait_for_site_replication_enabled(&source_env, 2).await?; + wait_for_site_replication_enabled(&target_env, 2).await?; + let remote_peer = source_info + .sites + .into_iter() + .find(|peer| peer.endpoint == target_env.url) + .ok_or("target peer missing from source site replication info")?; + wait_for_bucket_on_target(&target_client, source_bucket).await?; + + // Site replication rewires the operator target under the peer's ARN. + let site_arn = { + let deadline = tokio::time::Instant::now() + Duration::from_secs(30); + loop { + let response = list_replication_targets_request(&source_env, Some(source_bucket)).await?; + if response.status() == StatusCode::OK { + let targets: Vec = response.json().await?; + if let Some(arn) = targets + .iter() + .filter_map(|target| target.get("arn").and_then(|arn| arn.as_str())) + .find(|arn| arn.contains(&remote_peer.deployment_id)) + { + break arn.to_string(); + } + } + if tokio::time::Instant::now() >= deadline { + return Err(format!("site replication never rewired the target of {source_bucket}").into()); + } + sleep(Duration::from_millis(250)).await; + } + }; + assert_ne!(site_arn, operator_arn, "the site target must carry the peer ARN"); + + let started = site_replication_resync_op(&source_env, "start", &remote_peer).await?; + let entry = started + .buckets + .iter() + .find(|entry| entry.bucket == source_bucket) + .ok_or_else(|| format!("start response lost the bucket: {started:?}"))?; + assert_ne!( + entry.status, "conflict", + "a finished bucket-level resync must not block the site resync: {entry:?}" + ); + assert_eq!(started.status, "success", "unexpected start result: {:?}", started); + + let deadline = tokio::time::Instant::now() + Duration::from_secs(60); + let finished = loop { + let status = site_replication_resync_op(&source_env, "status", &remote_peer).await?; + match status.state.as_str() { + "completed" | "failed" => break status, + _ if tokio::time::Instant::now() < deadline => sleep(Duration::from_millis(250)).await, + _ => return Err(format!("site resync did not reach a terminal state in time: {status:?}").into()), + } + }; + let entry = finished + .buckets + .iter() + .find(|entry| entry.bucket == source_bucket) + .ok_or_else(|| format!("resync status lost the bucket: {finished:?}"))?; + assert_eq!( + (finished.state.as_str(), entry.status.as_str(), entry.failed_objects), + ("completed", "completed", 0), + "{finished:?}" + ); + wait_for_object_on_target(&target_client, source_bucket, key).await?; + + Ok(()) +} + #[tokio::test] async fn test_site_replication_resync_lifecycle_survives_real_server_restart() -> Result<(), Box> { init_logging(); diff --git a/rustfs/src/site_replication/hooks.rs b/rustfs/src/site_replication/hooks.rs index b907273a3..0acd3e7fa 100644 --- a/rustfs/src/site_replication/hooks.rs +++ b/rustfs/src/site_replication/hooks.rs @@ -1396,8 +1396,17 @@ pub(crate) fn reconcile_site_replication_bucket_targets( target.storage_class = existing.storage_class; target.health_check_duration = existing.health_check_duration; target.disable_proxy = existing.disable_proxy; - target.reset_before_date = existing.reset_before_date; - target.reset_id = existing.reset_id; + // A resync is keyed by the ARN it started under. When this pass + // takes over an operator target to the same peer (different ARN), + // its `reset_id` names a bucket-level resync the site resync + // status map cannot see, and `start_site_bucket_resync` would + // report the bucket as owned by "a different active resync" + // forever (rustfs/backlog#2479). Only an SR target being + // re-reconciled under its own ARN keeps its resync identity. + if existing.arn == target.arn { + target.reset_before_date = existing.reset_before_date; + target.reset_id = existing.reset_id; + } target.total_downtime = existing.total_downtime; target.last_online = existing.last_online; target.online = existing.online; diff --git a/rustfs/src/site_replication/tests.rs b/rustfs/src/site_replication/tests.rs index 7701c233f..6ce8a4b27 100644 --- a/rustfs/src/site_replication/tests.rs +++ b/rustfs/src/site_replication/tests.rs @@ -1978,6 +1978,80 @@ fn test_site_replication_bucket_target_replaces_tls_and_preserves_operational_fi assert!(target.disable_proxy); } +/// rustfs/backlog#2479: a bucket-level target to the same peer that had a +/// `replication-reset` run against it carries a `reset_id` keyed by its own +/// ARN. Taking that target over for site replication must not inherit the +/// id, otherwise every site resync reports the bucket as owned by another +/// active resync. An SR target re-reconciled under its own ARN keeps it. +#[test] +fn test_reconcile_site_replication_bucket_targets_drops_reset_id_from_taken_over_operator_target() { + let local = PeerInfo { + deployment_id: "local".to_string(), + ..peer("local", "https://local.example.com") + }; + let remote = PeerInfo { + deployment_id: "remote".to_string(), + ..peer("remote", "http://remote.example.com:9000") + }; + let state = SiteReplicationState { + service_account_access_key: "svc".to_string(), + peers: BTreeMap::from([("local".to_string(), local.clone()), ("remote".to_string(), remote.clone())]), + ..Default::default() + }; + + let operator_target = BucketTarget { + arn: "arn:minio:replication::7c0c5a1e-operator:photos-dst".to_string(), + source_bucket: "photos".to_string(), + target_bucket: "photos-dst".to_string(), + endpoint: "remote.example.com:9000".to_string(), + target_type: BucketTargetType::ReplicationService, + deployment_id: "remote".to_string(), + reset_id: "bucket-level-reset".to_string(), + reset_before_date: Some(OffsetDateTime::UNIX_EPOCH), + bandwidth_limit: 7, + ..Default::default() + }; + let reconciled = reconcile_site_replication_bucket_targets( + BucketTargets { + targets: vec![operator_target], + }, + "photos", + &state, + &local, + None, + "secret", + ) + .expect("reconcile targets"); + assert_eq!(reconciled.targets.len(), 1, "the operator target is taken over, not duplicated"); + let taken_over = &reconciled.targets[0]; + assert_eq!(taken_over.arn, "arn:minio:replication::remote:photos"); + assert_eq!(taken_over.target_bucket, "photos"); + assert!( + taken_over.reset_id.is_empty(), + "the bucket-level resync id must not follow the ARN change" + ); + assert!(taken_over.reset_before_date.is_none()); + assert_eq!(taken_over.bandwidth_limit, 7, "operator tuning still carries over"); + + let mut resynced = taken_over.clone(); + resynced.reset_id = "site-resync-1".to_string(); + resynced.reset_before_date = Some(OffsetDateTime::UNIX_EPOCH); + let reconciled_again = reconcile_site_replication_bucket_targets( + BucketTargets { targets: vec![resynced] }, + "photos", + &state, + &local, + None, + "secret", + ) + .expect("reconcile targets again"); + assert_eq!( + reconciled_again.targets[0].reset_id, "site-resync-1", + "an SR target keeps its own resync id" + ); + assert!(reconciled_again.targets[0].reset_before_date.is_some()); +} + #[test] fn test_bucket_versioning_xml_enables_versioning() { let data = bucket_versioning_xml().expect("versioning XML should serialize");