fix(replication): stop duplicate re-drives on own-version-id targets (#7323)

crates/ecstore/src/api/mod.rs

@@ -32,7 +32,7 @@ pub mod bucket {
     pub mod bucket_target_sys {
         pub use crate::bucket::bucket_target_sys::{
             AdvancedPutOptions, BucketTargetError, BucketTargetSys, PutObjectOptions, RemoveObjectOptions, S3ClientError,
-            SsecPassthroughCapability, TargetClient, append_version_id_query,
+            SsecPassthroughCapability, TargetClient, VersionIdentityCapability, append_version_id_query,
         };
     }
 

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -18,7 +18,7 @@ use crate::bucket::metadata_sys::get_replication_config;
 use crate::bucket::remote_s3_client::{
     PathStyle, REPLICATION_TARGET_RETRY_POLICY, RemoteCredentials, RemoteS3EndpointSpec, build_remote_s3_client,
 };
-use crate::bucket::replication::{ObjectLockIntegrity, object_lock_put_integrity};
+use crate::bucket::replication::{ObjectLockIntegrity, object_lock_put_integrity, replication_etags_match};
 use crate::bucket::replication::{ReplicationStatusType, ReplicationTargetConfigBridge};
 use crate::bucket::target::ARN;
 use crate::bucket::target::BucketTargetType;
@@ -126,6 +126,22 @@ impl From<&BucketTarget> for RemoteS3EndpointSpec {
 }
 
 pub type HeadObjectSdkError = Box<SdkError<HeadObjectError>>;
+
+/// Whether an edited bucket target still addresses the same remote service
+/// (endpoint, bucket, path style, TLS and identity), so a verdict learned
+/// about that service stays valid across the edit.
+fn same_replication_service(edited: &BucketTarget, previous: &BucketTarget) -> bool {
+    let access_key = |target: &BucketTarget| target.credentials.as_ref().map(|credentials| credentials.access_key.clone());
+    edited.endpoint == previous.endpoint
+        && edited.target_bucket == previous.target_bucket
+        && edited.secure == previous.secure
+        && edited.path == previous.path
+        && access_key(edited) == access_key(previous)
+}
+
+/// Page size and page budget for [`TargetClient::find_version_by_etag`].
+const FIND_VERSION_BY_ETAG_PAGE_SIZE: i32 = 1000;
+const FIND_VERSION_BY_ETAG_MAX_PAGES: usize = 8;
 pub type GetObjectSdkError = Box<SdkError<GetObjectError>>;
 pub type GetObjectTaggingSdkError = Box<SdkError<GetObjectTaggingError>>;
 pub type PutObjectTaggingSdkError = Box<SdkError<PutObjectTaggingError>>;
@@ -349,6 +365,13 @@ struct TargetClientBuildProbe {
 /// their import path while the verdict vocabulary lives with the
 /// replication decision logic.
 pub use crate::bucket::replication::SsecPassthroughCapability;
+/// Version-identity verdicts (see the enum's own docs in
+/// `rustfs-replication`) are cached here per target ARN and follow the same
+/// `arn_remotes_map` lifecycle. They carry no TTL: the verdict is refreshed
+/// by every replication write's response, so it can only go stale on a
+/// target that receives no writes — and a stale `MintsOwn` costs one extra
+/// content-identity lookup before a PUT, never a lost replica.
+pub use crate::bucket::replication::VersionIdentityCapability;
 
 /// How long an audited SSE-C passthrough verdict stays authoritative.
 ///
@@ -375,6 +398,11 @@ pub struct BucketTargetSys {
     /// SSE-C passthrough capability verdicts keyed by target ARN. See
     /// [`SsecPassthroughCapability`]; reset alongside `arn_remotes_map`.
     ssec_passthrough_map: Arc<RwLock<HashMap<String, SsecPassthroughRecord>>>,
+    /// Version-identity verdicts keyed by target ARN. See
+    /// [`VersionIdentityCapability`]; reset alongside `arn_remotes_map`. A std
+    /// lock (never held across an await) so the replication worker can record
+    /// a verdict from inside its synchronous PUT-response audit.
+    version_identity_map: Arc<std::sync::RwLock<HashMap<String, VersionIdentityCapability>>>,
     pub targets_map: Arc<RwLock<HashMap<String, Vec<BucketTarget>>>>,
     /// Buckets whose persisted `bucket-targets.json` exists but cannot be
     /// decoded (rustfs/backlog#2282). Written under the bucket's update mutex
@@ -423,6 +451,7 @@ impl BucketTargetSys {
         Self {
             arn_remotes_map: Arc::new(RwLock::new(HashMap::new())),
             ssec_passthrough_map: Arc::new(RwLock::new(HashMap::new())),
+            version_identity_map: Arc::new(std::sync::RwLock::new(HashMap::new())),
             targets_map: Arc::new(RwLock::new(HashMap::new())),
             unreadable_targets: Arc::new(RwLock::new(HashSet::new())),
             h_mutex: Arc::new(RwLock::new(HashMap::new())),
@@ -746,10 +775,40 @@ impl BucketTargetSys {
                 arn_remotes_map.remove(&target.arn);
                 health_map.remove(&target.arn);
                 ssec_map.remove(&target.arn);
+                self.forget_version_identity_capability(&target.arn);
             }
         }
     }
 
+    /// Cached version-identity verdict for a target ARN; `Unknown` until a
+    /// replication write or a replication-check VersionFidelity probe judged
+    /// it since the target was built.
+    pub fn version_identity_capability(&self, arn: &str) -> VersionIdentityCapability {
+        self.version_identity_map
+            .read()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+            .get(arn)
+            .copied()
+            .unwrap_or_default()
+    }
+
+    /// Record a version-identity verdict for a target ARN. Written by the
+    /// replication worker after every PutObject / CompleteMultipartUpload
+    /// response and by the replication-check VersionFidelity phase.
+    pub fn record_version_identity_capability(&self, arn: &str, capability: VersionIdentityCapability) {
+        self.version_identity_map
+            .write()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+            .insert(arn.to_string(), capability);
+    }
+
+    fn forget_version_identity_capability(&self, arn: &str) {
+        self.version_identity_map
+            .write()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+            .remove(arn);
+    }
+
     /// Cached SSE-C passthrough capability for a target ARN, plus whether the
     /// verdict is older than [`SSEC_PASSTHROUGH_CAPABILITY_TTL`]. `(Unknown,
     /// false)` when no verdict has been recorded since the target was built.
@@ -1162,12 +1221,32 @@ impl BucketTargetSys {
         // Remove existing targets
         if let Some(existing_targets) = targets_map.remove(bucket) {
             let mut ssec_map = self.ssec_passthrough_map.write().await;
+            let unchanged_service: HashMap<&str, &BucketTarget> = targets
+                .map(|new_targets| {
+                    new_targets
+                        .targets
+                        .iter()
+                        .map(|target| (target.arn.as_str(), target))
+                        .collect()
+                })
+                .unwrap_or_default();
             for target in existing_targets {
                 arn_remotes_map.remove(&target.arn);
                 health_map.remove(&target.arn);
                 // A rebuilt/edited target may point at a different service:
                 // the SSE-C passthrough verdict must be re-audited from Unknown.
                 ssec_map.remove(&target.arn);
+                // The version-identity verdict survives an edit that keeps the
+                // same remote service (a resync start or a bandwidth change
+                // rewrites the entry in place): forgetting it there would make
+                // the very resync that follows re-drive every object as a
+                // duplicate on a target that mints its own version ids.
+                if unchanged_service
+                    .get(target.arn.as_str())
+                    .is_none_or(|edited| !same_replication_service(edited, &target))
+                {
+                    self.forget_version_identity_capability(&target.arn);
+                }
                 self.update_bandwidth_limit(bucket, &target.arn, 0);
             }
         }
@@ -1892,6 +1971,62 @@ impl TargetClient {
             .map_err(Box::new)
     }
 
+    /// Locate a replica by content identity on a target that mints its own
+    /// version ids: page `ListObjectVersions` under the exact key and return
+    /// the newest live version whose ETag matches `source_etag`. Delete
+    /// markers and prefix siblings never match. Bounded to
+    /// [`FIND_VERSION_BY_ETAG_MAX_PAGES`] pages so a key with a very deep
+    /// history cannot turn one convergence check into an unbounded scan; a
+    /// replica beyond that window reads as missing, which only costs a
+    /// re-PUT (today's behaviour), never a lost object.
+    pub async fn find_version_by_etag(
+        &self,
+        bucket: &str,
+        object: &str,
+        source_etag: &str,
+    ) -> Result<Option<String>, Box<SdkError<aws_sdk_s3::operation::list_object_versions::ListObjectVersionsError>>> {
+        let mut key_marker: Option<String> = None;
+        let mut version_id_marker: Option<String> = None;
+        for _ in 0..FIND_VERSION_BY_ETAG_MAX_PAGES {
+            let page = self
+                .client
+                .list_object_versions()
+                .bucket(bucket)
+                .prefix(object)
+                .max_keys(FIND_VERSION_BY_ETAG_PAGE_SIZE)
+                .set_key_marker(key_marker.take())
+                .set_version_id_marker(version_id_marker.take())
+                .send()
+                .await
+                .map_err(Box::new)?;
+            if let Some(version) = page.versions().iter().find(|version| {
+                version.key() == Some(object)
+                    && version.version_id().is_some_and(|id| !id.is_empty())
+                    && replication_etags_match(Some(source_etag), version.e_tag())
+            }) {
+                return Ok(version.version_id().map(str::to_string));
+            }
+            // Every listed key is >= the prefix; once the listing moved past
+            // the exact key there is nothing left to find.
+            if page
+                .versions()
+                .iter()
+                .any(|version| version.key().is_some_and(|key| key > object))
+            {
+                return Ok(None);
+            }
+            if !page.is_truncated().unwrap_or(false) {
+                return Ok(None);
+            }
+            key_marker = page.next_key_marker().map(str::to_string);
+            version_id_marker = page.next_version_id_marker().map(str::to_string);
+            if key_marker.is_none() {
+                return Ok(None);
+            }
+        }
+        Ok(None)
+    }
+
     /// HEAD used by the read-proxy path (GET/HEAD of an object not yet
     /// replicated locally, MinIO `proxyHeadToRepTarget`).
     ///
@@ -3221,6 +3356,64 @@ mod tests {
         assert!(message.contains("connection refused"));
     }
 
+    #[test]
+    fn same_replication_service_ignores_resync_and_bandwidth_edits() {
+        let base = BucketTarget {
+            endpoint: "target.example:9000".to_string(),
+            target_bucket: "replica".to_string(),
+            secure: true,
+            path: "on".to_string(),
+            arn: "arn:rustfs:replication:us-east-1:bucket:same".to_string(),
+            credentials: Some(Credentials {
+                access_key: "access".to_string(),
+                ..Default::default()
+            }),
+            ..Default::default()
+        };
+        let resync_edit = BucketTarget {
+            reset_id: "reset-1".to_string(),
+            bandwidth_limit: 1024,
+            ..base.clone()
+        };
+        assert!(same_replication_service(&resync_edit, &base));
+        for moved in [
+            BucketTarget {
+                endpoint: "other.example:9000".to_string(),
+                ..base.clone()
+            },
+            BucketTarget {
+                target_bucket: "other".to_string(),
+                ..base.clone()
+            },
+            BucketTarget {
+                secure: false,
+                ..base.clone()
+            },
+            BucketTarget {
+                credentials: Some(Credentials {
+                    access_key: "rotated".to_string(),
+                    ..Default::default()
+                }),
+                ..base.clone()
+            },
+        ] {
+            assert!(!same_replication_service(&moved, &base));
+        }
+    }
+
+    #[test]
+    fn version_identity_verdict_is_per_arn_and_forgotten_with_the_target() {
+        let sys = BucketTargetSys::default();
+        let arn = "arn:rustfs:replication:us-east-1:bucket:identity";
+        assert_eq!(sys.version_identity_capability(arn), VersionIdentityCapability::Unknown);
+        sys.record_version_identity_capability(arn, VersionIdentityCapability::MintsOwn);
+        assert_eq!(sys.version_identity_capability(arn), VersionIdentityCapability::MintsOwn);
+        assert_eq!(sys.version_identity_capability("other"), VersionIdentityCapability::Unknown);
+        // A rebuilt target may point at a different service.
+        sys.forget_version_identity_capability(arn);
+        assert_eq!(sys.version_identity_capability(arn), VersionIdentityCapability::Unknown);
+    }
+
     #[test]
     fn endpoint_health_key_preserves_explicit_port() {
         let url = Url::parse("https://remote.example:9443").expect("url should parse");

crates/ecstore/src/bucket/replication/mod.rs

@@ -66,6 +66,7 @@ pub(crate) use replication_lifecycle_bridge::ReplicationLifecycleBridge;
 pub(crate) use replication_migration_bridge::ReplicationMigrationBridge;
 pub use replication_object_bridge::ReplicationObjectBridge;
 pub use replication_object_config::{DeleteReplicationConfigSnapshot, ReplicationConfig};
+pub(crate) use replication_object_decision_boundary::replication_etags_match;
 pub use replication_object_decision_boundary::{
     MustReplicateOptions, ReplicationDeleteScheduleInput, ReplicationDeleteStateSource, delete_replication_state_from_config,
     delete_replication_version_id, should_schedule_delete_replication, should_use_existing_delete_replication_info,
@@ -88,5 +89,6 @@ pub use replication_state::{ReplicationStats, RuntimeReplicationTargetBacklog};
 pub use replication_stats_boundary::{BucketReplicationStat, BucketReplicationStats, BucketStats, InQueueMetric, XferStats};
 pub use replication_storage_boundary::{ReplicationObjectIO, ReplicationStorage};
 pub use replication_target_boundary::SsecPassthroughCapability;
+pub use replication_target_boundary::VersionIdentityCapability;
 pub use replication_target_boundary::{ObjectLockIntegrity, object_lock_put_integrity};
 pub(crate) use replication_target_config_bridge::ReplicationTargetConfigBridge;

crates/ecstore/src/bucket/replication/replication_resyncer.rs

@@ -56,6 +56,8 @@ use super::replication_storage_boundary::{
 };
 #[cfg(test)]
 use super::replication_storage_boundary::{NamespaceLockFence, NamespaceLockSignalTestFence, ReplicationDeletedObject};
+#[cfg(test)]
+use super::replication_target_boundary::VersionIdentityCapability;
 use super::replication_target_boundary::{
     ERR_REPLICATION_SSEC_PASSTHROUGH_UNSUPPORTED, HeadObjectSdkError, PutObjectOptions, PutObjectPartOptions,
     RemotePutObjectResponse, ReplicationTargetStore, S3ClientError, SsecPassthroughCapability, SsecPassthroughGate, TargetClient,
@@ -63,7 +65,7 @@ use super::replication_target_boundary::{
     replication_delete_marker_purge_remove_options, replication_delete_remove_options, replication_force_delete_remove_options,
     replication_object_is_ssec_encrypted, replication_put_object_header_size, replication_put_object_options,
     replication_target_head_is_newer_null_version, resolve_read_api_version_id, ssec_passthrough_evidence_present,
-    ssec_passthrough_gate, version_identity_drifted,
+    ssec_passthrough_gate, version_identity_capability_from_put, version_identity_drifted,
 };
 use super::replication_versioning_boundary::ReplicationVersioningStore;
 use super::runtime_boundary as runtime_sources;
@@ -123,6 +125,7 @@ const EVENT_DELETE_MARKER_PURGE_FAILED: &str = "replication_delete_marker_purge_
 const EVENT_DELETE_MARKER_PURGE_MRF: &str = "replication_delete_marker_purge_mrf";
 const METRIC_DELETE_MARKER_PURGE_TOTAL: &str = "rustfs_replication_delete_marker_purge_total";
 const EVENT_REPLICATION_VERSION_IDENTITY_DRIFT: &str = "replication_version_identity_drift";
+const EVENT_REPLICATION_DRIFTED_REPLICA_LOCATED: &str = "replication_drifted_replica_located";
 const EVENT_REPLICATION_OBJECT_FAILED: &str = "replication_object_failed";
 const EVENT_REPLICATION_PURGE_OBJECT_LOCK_DENIED: &str = "replication_purge_object_lock_denied";
 
@@ -332,6 +335,12 @@ fn verify_single_part_replica(
 const REPLICA_ETAG_MISMATCH_ERROR: &str = "replica etag mismatch: the target persisted different bytes than were sent";
 
 fn audit_target_version_identity(tgt_client: &TargetClient, source_version_id: &str, assigned_version_id: Option<&str>) {
+    // Every write refreshes the cached verdict, so the convergence fallback
+    // below (`replica_head_fallback`) knows whether a 404 on a
+    // version-addressed HEAD can mean "replica missing" on this target.
+    if let Some(capability) = version_identity_capability_from_put(source_version_id, assigned_version_id) {
+        ReplicationTargetStore::record_version_identity_capability(&tgt_client.arn, capability);
+    }
     if !version_identity_drifted(source_version_id, assigned_version_id) {
         return;
     }
@@ -404,11 +413,70 @@ async fn head_object_fallback(
 ) -> std::result::Result<Option<HeadObjectOutput>, HeadObjectSdkError> {
     match head_object_for_worker(tgt_client, &tgt_client.bucket, object, None).await {
         Ok(oi) => Ok(Some(oi)),
-        Err(e) if e.as_service_error().is_some_and(|se| se.is_not_found()) || has_raw_status(&e, 404) => Ok(None),
+        Err(e) if head_object_not_found(&e) => Ok(None),
         Err(e) => Err(e),
     }
 }
 
+fn head_object_not_found(err: &HeadObjectSdkError) -> bool {
+    err.as_service_error().is_some_and(|se| se.is_not_found()) || has_raw_status(err, 404)
+}
+
+/// Second look at a replica whose version-addressed HEAD failed, for the two
+/// target shapes where that failure is not a verdict on the replica:
+///
+/// - AWS-style 400/403 (the RustFS uuid is rejected as malformed): HEAD the
+///   current version without a version id; callers compare ETags.
+/// - 404 on a target known to mint its own version ids (the Wasabi shape,
+///   rustfs/backlog#2340): the source id never existed there, so locate the
+///   replica by exact key and ETag through ListObjectVersions and HEAD the id
+///   the target assigned. Without this, every heal, MRF retry and
+///   existing-object resync re-drive PUTs the object again and mints one
+///   more target version.
+///
+/// `None` when the error stands as-is: a real miss on an adopting target, or
+/// a target whose identity contract is still unknown. A failed lookup is
+/// returned as a HEAD-shaped error so callers keep their "target operation
+/// failed" handling (retry later) instead of re-driving the PUT.
+async fn replica_head_fallback(
+    tgt_client: &TargetClient,
+    object: &str,
+    source_etag: Option<&str>,
+    err: &HeadObjectSdkError,
+) -> Option<std::result::Result<Option<HeadObjectOutput>, HeadObjectSdkError>> {
+    if is_version_id_format_mismatch(err) {
+        return Some(head_object_fallback(tgt_client, object).await);
+    }
+    if !head_object_not_found(err)
+        || !ReplicationTargetStore::version_identity_capability(&tgt_client.arn).version_addressing_unreliable()
+    {
+        return None;
+    }
+    let etag = source_etag.filter(|etag| !etag.trim().is_empty())?;
+    Some(match tgt_client.find_version_by_etag(&tgt_client.bucket, object, etag).await {
+        Ok(Some(assigned_version_id)) => {
+            debug!(
+                event = EVENT_REPLICATION_DRIFTED_REPLICA_LOCATED,
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+                bucket = %tgt_client.bucket,
+                object = %object,
+                arn = %tgt_client.arn,
+                assigned_version_id = %assigned_version_id,
+                "Located replica by content identity on a target that mints its own version ids"
+            );
+            match head_object_for_worker(tgt_client, &tgt_client.bucket, object, Some(assigned_version_id)).await {
+                Ok(oi) => Ok(Some(oi)),
+                // The located version disappeared between LIST and HEAD.
+                Err(e) if head_object_not_found(&e) => Ok(None),
+                Err(e) => Err(e),
+            }
+        }
+        Ok(None) => Ok(None),
+        Err(list_err) => Err(Box::new(SdkError::construction_failure(*list_err))),
+    })
+}
+
 /// Resolve the N2 fail-closed gate for an SSE-C passthrough attempt against
 /// this target. Returns `Some(audit_required)` when replication may proceed;
 /// on a freshly-flagged header-dropping target it settles `rinfo` as FAILED
@@ -1400,31 +1468,26 @@ async fn verify_resync_head_result(
                 (0, None)
             }
         }
-        Err(err) if is_version_id_format_mismatch(&err) => {
-            // AWS-style target rejects the RustFS UUID versionId
-            // (400). Re-verify without the versionId before
-            // concluding the object failed to replicate, instead
-            // of counting a well-replicated object as failed.
-            match head_object_fallback(target_client.as_ref(), &roi.name).await {
-                Ok(Some(_)) => {
+        Err(err) => {
+            // A version-addressed HEAD is not the last word on every target:
+            // re-verify through the fallback before counting a well-replicated
+            // object as failed (see `replica_head_fallback`).
+            match replica_head_fallback(target_client.as_ref(), &roi.name, roi.etag.as_deref(), &err).await {
+                Some(Ok(Some(_))) => {
                     st.replicated_count += 1;
                     st.replicated_size += roi.size;
                     (roi.size, None)
                 }
-                Ok(None) => {
+                Some(Ok(None)) | None => {
                     st.failed_count += 1;
                     (0, Some(err))
                 }
-                Err(e2) => {
+                Some(Err(e2)) => {
                     st.failed_count += 1;
                     (0, Some(e2))
                 }
             }
         }
-        Err(err) => {
-            st.failed_count += 1;
-            (0, Some(err))
-        }
     }
 }
 
@@ -3597,11 +3660,8 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
                 }
             }
             Err(e) => {
-                if e.as_service_error().is_some_and(|se| se.is_not_found()) || has_raw_status(&e, 404) {
-                    // Object not on target yet → fall through to PUT.
-                } else if is_version_id_format_mismatch(&e) {
-                    // Version-ID format mismatch: retry without versionId and compare ETags.
-                    match head_object_fallback(&tgt_client, &object).await {
+                if let Some(fallback) = replica_head_fallback(&tgt_client, &object, object_info.etag.as_deref(), &e).await {
+                    match fallback {
                         Ok(Some(oi)) if replication_etags_match(object_info.etag.as_deref(), oi.e_tag.as_deref()) => {
                             if ssec_audit_required
                                 && !settle_ssec_passthrough_evidence(&oi, &tgt_client, &bucket, &object, &mut rinfo).await
@@ -3631,6 +3691,8 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
                             return rinfo;
                         }
                     }
+                } else if head_object_not_found(&e) {
+                    // Object not on target yet → fall through to PUT.
                 } else {
                     rinfo.error = Some(e.to_string());
                     warn!(
@@ -4230,9 +4292,8 @@ async fn resolve_replicate_all_action(
             }
         }
         Err(e) => {
-            if is_version_id_format_mismatch(&e) {
-                // Version-ID format mismatch: retry without versionId and compare ETags.
-                match head_object_fallback(tgt_client, object).await {
+            if let Some(fallback) = replica_head_fallback(tgt_client, object, object_info.etag.as_deref(), &e).await {
+                match fallback {
                     Ok(Some(oi)) => {
                         let etags_match = replication_etags_match(object_info.etag.as_deref(), oi.e_tag.as_deref());
                         if require_existing_target && !etags_match {
@@ -4284,7 +4345,7 @@ async fn resolve_replicate_all_action(
                         return None;
                     }
                 }
-            } else if e.as_service_error().is_some_and(|se| se.is_not_found()) || has_raw_status(&e, 404) {
+            } else if head_object_not_found(&e) {
                 if require_existing_target {
                     rinfo.error = Some("replica metadata target does not contain this object version".to_string());
                     rinfo.duration = (OffsetDateTime::now_utc() - start_time).unsigned_abs();
@@ -5383,6 +5444,178 @@ mod tests {
         ReplicationTargetStore::register_test_target(target).await;
     }
 
+    const DRIFTED_ASSIGNED_VERSION_ID: &str = "001788697733811332140-fR6j6uXKV-";
+    const DRIFTED_ETAG: &str = "9a0364b9e99bb480dd25e1f0284c8555";
+
+    /// The Wasabi shape (rustfs/backlog#2340): a version-addressed HEAD with
+    /// the source uuid answers 404 (not the AWS 400), ListObjectVersions shows
+    /// the id the target minted, and a HEAD by that id succeeds. Serves exactly
+    /// `requests` connections and returns the request lines it saw.
+    fn spawn_drifted_target_server(requests: usize) -> (String, std::thread::JoinHandle<Vec<String>>) {
+        use std::io::{Read, Write};
+
+        let listener = std::net::TcpListener::bind(("127.0.0.1", 0)).expect("test HTTP listener should bind");
+        let endpoint = format!("http://{}", listener.local_addr().expect("test HTTP listener should have an address"));
+        let handle = std::thread::spawn(move || {
+            let mut seen = Vec::new();
+            for _ in 0..requests {
+                let (mut stream, _) = listener.accept().expect("test HTTP client should connect");
+                let mut request = [0_u8; 8192];
+                let bytes_read = stream.read(&mut request).expect("test HTTP request should be read");
+                let text = String::from_utf8_lossy(&request[..bytes_read]).to_string();
+                let request_line = text.lines().next().unwrap_or_default().to_string();
+                let response = if request_line.starts_with("HEAD ") {
+                    if request_line.contains(&format!("versionId={DRIFTED_ASSIGNED_VERSION_ID}")) {
+                        format!(
+                            "HTTP/1.1 200 OK\r\nETag: \"{DRIFTED_ETAG}\"\r\nContent-Length: 4\r\nLast-Modified: Sun, 06 Sep 2026 10:00:00 GMT\r\nConnection: close\r\n\r\n"
+                        )
+                    } else {
+                        "HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".to_string()
+                    }
+                } else if request_line.starts_with("GET ") && request_line.contains("versions") {
+                    let body = format!(
+                        "<?xml version=\"1.0\" encoding=\"UTF-8\"?><ListVersionsResult xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\"><Name>target-bucket</Name><Prefix>object</Prefix><MaxKeys>1000</MaxKeys><IsTruncated>false</IsTruncated><Version><Key>object</Key><VersionId>{DRIFTED_ASSIGNED_VERSION_ID}</VersionId><IsLatest>true</IsLatest><LastModified>2026-09-06T10:00:00.000Z</LastModified><ETag>&quot;{DRIFTED_ETAG}&quot;</ETag><Size>4</Size><StorageClass>STANDARD</StorageClass></Version></ListVersionsResult>"
+                    );
+                    format!(
+                        "HTTP/1.1 200 OK\r\nContent-Type: application/xml\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
+                        body.len()
+                    )
+                } else {
+                    "HTTP/1.1 500 Unexpected\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".to_string()
+                };
+                stream
+                    .write_all(response.as_bytes())
+                    .expect("test HTTP response should be written");
+                seen.push(request_line);
+            }
+            seen
+        });
+        (endpoint, handle)
+    }
+
+    fn drifted_roi_and_object() -> (ReplicateObjectInfo, ObjectInfo) {
+        let roi = ReplicateObjectInfo {
+            bucket: "source".to_string(),
+            name: "object".to_string(),
+            version_id: Some(Uuid::new_v4()),
+            op_type: ReplicationType::Heal,
+            replication_status: ReplicationStatusType::Pending,
+            etag: Some(DRIFTED_ETAG.to_string()),
+            size: 4,
+            ..Default::default()
+        };
+        let object_info = ObjectInfo {
+            bucket: roi.bucket.clone(),
+            name: roi.name.clone(),
+            version_id: roi.version_id,
+            etag: Some(DRIFTED_ETAG.to_string()),
+            size: 4,
+            ..Default::default()
+        };
+        (roi, object_info)
+    }
+
+    #[tokio::test]
+    async fn heal_redrive_locates_replica_by_etag_on_target_that_mints_own_version_ids() {
+        let (endpoint, server) = spawn_drifted_target_server(3);
+        let target = test_target_client(endpoint);
+        ReplicationTargetStore::record_version_identity_capability(&target.arn, VersionIdentityCapability::MintsOwn);
+        let (roi, object_info) = drifted_roi_and_object();
+        let mut rinfo = replicate_all_target_info(&roi, &target);
+
+        let action = resolve_replicate_all_action(
+            ReplicateAllActionContext {
+                roi: &roi,
+                tgt_client: &target,
+                bucket: &roi.bucket,
+                object: &roi.name,
+                start_time: OffsetDateTime::now_utc(),
+                ssec_audit_required: false,
+            },
+            object_info,
+            &mut rinfo,
+        )
+        .await;
+
+        assert!(
+            matches!(action, Some((ReplicationAction::None, _))),
+            "a replica located by content identity must not be re-driven: {action:?}"
+        );
+        assert!(rinfo.error.is_none(), "{:?}", rinfo.error);
+        let seen = server.join().expect("test HTTP server should finish");
+        assert_eq!(seen.len(), 3, "HEAD by source id, ListObjectVersions, HEAD by assigned id: {seen:?}");
+        assert!(seen[0].starts_with("HEAD ") && seen[0].contains(&roi.version_id.unwrap().to_string()));
+        assert!(seen[1].starts_with("GET ") && seen[1].contains("prefix=object"), "{}", seen[1]);
+        assert!(seen[2].starts_with("HEAD ") && seen[2].contains(DRIFTED_ASSIGNED_VERSION_ID));
+    }
+
+    #[tokio::test]
+    async fn head_not_found_still_replicates_when_identity_contract_is_unknown() {
+        // Same 404, but the target never revealed whether it adopts version
+        // ids: a 404 keeps meaning "replica missing" (adopting targets, e.g.
+        // RustFS/MinIO peers, must not skip a genuinely missing version).
+        let (endpoint, server) = spawn_head_status_server(404);
+        let target = test_target_client(endpoint);
+        let (roi, object_info) = drifted_roi_and_object();
+        let mut rinfo = replicate_all_target_info(&roi, &target);
+
+        let action = resolve_replicate_all_action(
+            ReplicateAllActionContext {
+                roi: &roi,
+                tgt_client: &target,
+                bucket: &roi.bucket,
+                object: &roi.name,
+                start_time: OffsetDateTime::now_utc(),
+                ssec_audit_required: false,
+            },
+            object_info,
+            &mut rinfo,
+        )
+        .await;
+
+        assert!(matches!(action, Some((ReplicationAction::All, _))));
+        server.join().expect("test HTTP server should finish");
+    }
+
+    #[tokio::test]
+    async fn resync_verification_counts_drifted_replica_as_replicated() {
+        let (endpoint, server) = spawn_drifted_target_server(3);
+        let target = test_target_client(endpoint);
+        ReplicationTargetStore::record_version_identity_capability(&target.arn, VersionIdentityCapability::MintsOwn);
+        let (roi, _) = drifted_roi_and_object();
+        let mut st = TargetReplicationResyncStatus::default();
+
+        let head_result =
+            head_object_for_worker(target.as_ref(), &target.bucket, &roi.name, roi.version_id.map(|v| v.to_string())).await;
+        let (size, err) = verify_resync_head_result(head_result, &roi, &mut st, &target).await;
+
+        assert!(err.is_none(), "{err:?}");
+        assert_eq!((size, st.replicated_count, st.failed_count), (4, 1, 0));
+        server.join().expect("test HTTP server should finish");
+    }
+
+    #[test]
+    fn put_response_audit_records_identity_verdict() {
+        let target = test_target_client("http://127.0.0.1:1".to_string());
+        let source = Uuid::new_v4().to_string();
+        audit_target_version_identity(&target, &source, Some(DRIFTED_ASSIGNED_VERSION_ID));
+        assert_eq!(
+            ReplicationTargetStore::version_identity_capability(&target.arn),
+            VersionIdentityCapability::MintsOwn
+        );
+        audit_target_version_identity(&target, &source, Some(&source));
+        assert_eq!(
+            ReplicationTargetStore::version_identity_capability(&target.arn),
+            VersionIdentityCapability::Adopts
+        );
+        // An unversioned write carries no contract and must not overwrite it.
+        audit_target_version_identity(&target, "null", None);
+        assert_eq!(
+            ReplicationTargetStore::version_identity_capability(&target.arn),
+            VersionIdentityCapability::Adopts
+        );
+    }
+
     #[test]
     fn resync_admission_configuration_is_bounded() {
         assert_eq!(ENV_REPL_RESYNC_MAX_JOBS, "RUSTFS_REPL_RESYNC_MAX_JOBS");

crates/ecstore/src/bucket/replication/replication_target_boundary.rs

@@ -48,6 +48,7 @@ pub use rustfs_replication::{ObjectLockIntegrity, object_lock_put_integrity};
 pub(crate) use rustfs_replication::{
     SsecPassthroughGate, is_replication_target_offline_error, ssec_passthrough_gate, version_identity_drifted,
 };
+pub use rustfs_replication::{VersionIdentityCapability, version_identity_capability_from_put};
 
 use super::replication_config_store::ReplicationConfigStore;
 use super::replication_error_boundary::{Error, Result};
@@ -192,6 +193,14 @@ impl ReplicationTargetStore {
             .await
     }
 
+    pub(crate) fn version_identity_capability(arn: &str) -> VersionIdentityCapability {
+        BucketTargetSys::get().version_identity_capability(arn)
+    }
+
+    pub(crate) fn record_version_identity_capability(arn: &str, capability: VersionIdentityCapability) {
+        BucketTargetSys::get().record_version_identity_capability(arn, capability)
+    }
+
     #[cfg(test)]
     pub(crate) async fn register_test_target(target_client: &Arc<TargetClient>) {
         BucketTargetSys::get().arn_remotes_map.write().await.insert(