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

@@ -1 +1 @@
-sha256=0e338d305260229e17ccfb2adc48a6212dbdfea36a9ebfb5a4e0d38658e6cc45
+sha256=fa9281eb620cec40bfc68e62eab2773cd313f8ce1497d808ff813cfe5aa243ec

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<dyn Error + Send + Sync>> {
+    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<serde_json::Value> = 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<dyn Error + Send + Sync>> {
     init_logging();

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;

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");