fix(replication): locate encrypted replicas without comparing ETags (#8602)

Since 1.0.2 an encrypted object's ETag is a keyed fingerprint, not the plaintext MD5. A target that computes its own ETag (AWS S3, Wasabi) never reports it, so every "find the existing replica by exact ETag" step missed encrypted objects on targets that mint their own version ids: - a heal / MRF re-drive after a lost PUT response wrote a second version; - an existing-object resync re-PUT a replica its ledger already named (the ledger-located HEAD was confirmed by ETag only), and on the AWS 400 shape the ledger was not consulted at all; - resync verification counted such replicas as failed; - a ledger-less version purge reported the replica as already gone. Encrypted sources now recognise their replica by shape: the byte count the replication PUT sent (plaintext for managed SSE, ciphertext for SSE-C) and a target write time no earlier than the source version. An ETag match still wins (RustFS/MinIO targets). The ledger is consulted first on both the 404 and the AWS 400 shape. Without a ledger, only a unique shape match among the unclaimed ListObjectVersions candidates is adopted and recorded; several shape matches mean re-PUT for convergence and a refused purge. The AWS current-version fallback stays ETag-only, and a listing the target refuses on the 400/403 shape keeps that fallback (re-PUT) rather than stalling. An unversioned target's `null` version is never matched by shape: it holds whichever generation was written last. Plaintext sources keep exact ETag identity. Builds on the ETag-computing fake target mode and the ComputesOwnEtags matrix row from rustfs#8583. Adds an SSE-S3 variant of the mint-own-ids re-drive test (red before this change: the heal left two target versions), and both variants now also assert the purge.
2026-10-28 15:02:49 +00:00 · 2026-10-16 13:44:33 +08:00
parent f5ff8783f4
commit 741de6f81c
6 changed files with 811 additions and 131 deletions
@@ -303,14 +303,40 @@ fn expectation(mode: TargetMode, shape: ObjectShape) -> Expectation {
 /// - the first PUT lands but its response is lost, so the object is FAILED
 ///   and the scanner heal pass re-drives it;
 /// - an existing-object resync re-drives a COMPLETED object unconditionally.
+///
+/// A permanent delete of the healed version then has to remove exactly the
+/// replica the heal pass located.
 #[tokio::test]
 async fn matrix_mint_own_version_ids_redrive_does_not_duplicate() -> TestResult {
+    run_redrive_does_not_duplicate(RedriveObjects::Plaintext).await
+}
+
+/// The same re-drives for SSE-S3 objects against a target that also computes
+/// its own ETags (AWS S3). The source ETag is a keyed fingerprint
+/// (docs/operations/encrypted-etags.md) while the replica reports the MD5 of
+/// the plaintext it received, so no exact-ETag lookup can ever find the
+/// replica: the lost-response heal, the resync and the purge must locate it
+/// without comparing ETags.
+#[tokio::test]
+async fn matrix_mint_own_version_ids_encrypted_redrive_does_not_duplicate() -> TestResult {
+    run_redrive_does_not_duplicate(RedriveObjects::SseS3OnEtagComputingTarget).await
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum RedriveObjects {
+    Plaintext,
+    SseS3OnEtagComputingTarget,
+}
+
+async fn run_redrive_does_not_duplicate(objects: RedriveObjects) -> TestResult {
    init_logging();
+    let encrypted = objects == RedriveObjects::SseS3OnEtagComputingTarget;

    let target = FakeS3Target::start().await?;
    let target_bucket = "matrix-mint-own-redrive-dst".to_string();
    target.create_bucket_with_object_lock(target_bucket.clone());
    target.assign_own_version_ids(true);
+    target.ignore_source_etags(encrypted);

    let mut env_vars = replication_fast_env();
    env_vars.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV);
@@ -324,6 +350,7 @@ async fn matrix_mint_own_version_ids_redrive_does_not_duplicate() -> TestResult
    ]);
    let env = OdmTestEnv::start_with(OdmEnvOptions {
        env: env_vars,
+        local_kms: encrypted,
        ..OdmEnvOptions::default()
    })
    .await?;
@@ -352,45 +379,63 @@ async fn matrix_mint_own_version_ids_redrive_does_not_duplicate() -> TestResult
        },
    )
    .await?;
-    put_bucket_replication(source_env, source_bucket, &target_arn).await?;
+    put_bucket_replication_with_delete_statuses(source_env, source_bucket, &target_arn, "Enabled", Some("Enabled")).await?;
+
+    let put = |key: &'static str, fill: u8, len: usize| {
+        let request = source_client
+            .put_object()
+            .bucket(source_bucket)
+            .key(key)
+            .body(ByteStream::from(payload(len, fill)));
+        let request = if encrypted {
+            request.server_side_encryption(ServerSideEncryption::Aes256)
+        } else {
+            request
+        };
+        async move { request.send().await }
+    };

    // Teach the worker the target's identity contract with one ordinary
    // write, exactly as production learns it (the PUT response carries the
    // minted id).
    let probe_key = "redrive/identity-probe.bin";
-    source_client
-        .put_object()
-        .bucket(source_bucket)
-        .key(probe_key)
-        .body(ByteStream::from(payload(4 * 1024, 0x01)))
-        .send()
-        .await?;
+    let probe = put(probe_key, 0x01, 4 * 1024).await?;
    assert_eq!(
        wait_for_terminal_replication_status(&source_client, source_bucket, probe_key).await?,
        "COMPLETED"
    );
+    if encrypted {
+        // The variant only proves something while the replica's ETag differs
+        // from the source's encrypted-ETag fingerprint.
+        let replica = single_target_version(&target, &target_bucket, probe_key)?;
+        let replica_etag = fake_source_client(&target)
+            .head_object()
+            .bucket(&target_bucket)
+            .key(probe_key)
+            .version_id(&replica)
+            .send()
+            .await?
+            .e_tag()
+            .map(str::to_string);
+        if replica_etag.is_none() || replica_etag.as_deref() == probe.e_tag() {
+            return Err(format!(
+                "the replica ETag {replica_etag:?} must differ from the source fingerprint {:?}",
+                probe.e_tag()
+            )
+            .into());
+        }
+    }

    // Shape 1: the PUT is stored, its response never arrives, heal re-drives.
    let heal_key = "redrive/heal.bin";
    target.inject_for_key(FakeTargetOperation::PutObject, heal_key, FakeTargetFault::DisconnectAfterResponse, 1);
-    source_client
-        .put_object()
-        .bucket(source_bucket)
-        .key(heal_key)
-        .body(ByteStream::from(payload(8 * 1024, 0x02)))
-        .send()
-        .await?;
+    let healed = put(heal_key, 0x02, 8 * 1024).await?;
+    let healed_version = healed.version_id().ok_or("source PUT returned no version id")?.to_string();
    wait_for_replication_status_and_single_version(&source_client, source_bucket, &target, &target_bucket, heal_key).await?;

    // Shape 2: an existing-object resync re-drives a COMPLETED object.
    let resync_key = "redrive/resync.bin";
-    source_client
-        .put_object()
-        .bucket(source_bucket)
-        .key(resync_key)
-        .body(ByteStream::from(payload(8 * 1024, 0x03)))
-        .send()
-        .await?;
+    put(resync_key, 0x03, 8 * 1024).await?;
    assert_eq!(
        wait_for_terminal_replication_status(&source_client, source_bucket, resync_key).await?,
        "COMPLETED"
@@ -421,6 +466,20 @@ async fn matrix_mint_own_version_ids_redrive_does_not_duplicate() -> TestResult
        );
    }

+    // The heal pass recorded the replica it located; a permanent delete of
+    // that version must remove exactly it.
+    source_client
+        .delete_object()
+        .bucket(source_bucket)
+        .key(heal_key)
+        .version_id(&healed_version)
+        .send()
+        .await?;
+    wait_until("permanent delete of the healed replica", || async {
+        Ok(target.stored_versions(&target_bucket, heal_key).is_empty())
+    })
+    .await?;
+
    target.shutdown().await;
    Ok(())
 }
@@ -142,11 +142,11 @@ fn same_replication_service(edited: &BucketTarget, previous: &BucketTarget) -> b
        && access_key(edited) == access_key(previous)
 }

-/// Page size and page budget for [`TargetClient::locate_replica_by_etag`].
+/// Page size and page budget for [`TargetClient::replica_candidates`].
 const FIND_VERSION_BY_ETAG_PAGE_SIZE: i32 = 1000;
 const FIND_VERSION_BY_ETAG_MAX_PAGES: usize = 8;
-/// Candidate cap for [`TargetClient::replica_candidates_by_etag`]: more than
-/// this many same-content versions of one key is ambiguity by any measure.
+/// Candidate cap for [`TargetClient::replica_candidates`]: more than this
+/// many matching versions of one key is ambiguity by any measure.
 const FIND_VERSION_BY_ETAG_MAX_MATCHES: usize = 16;
 pub type GetObjectSdkError = Box<SdkError<GetObjectError>>;
 pub type GetObjectTaggingSdkError = Box<SdkError<GetObjectTaggingError>>;
@@ -2030,27 +2030,28 @@ impl TargetClient {

    /// Candidate replicas by content identity on a target that mints its own
    /// version ids: page `ListObjectVersions` under the exact key and report
-    /// the live versions whose ETag matches `source_etag`, newest first.
-    /// Delete markers and prefix siblings never match. Bounded to
+    /// the live versions that match `identity`, newest first. Delete markers
+    /// and prefix siblings never match. Bounded to
    /// [`FIND_VERSION_BY_ETAG_MAX_PAGES`] pages and
-    /// [`FIND_VERSION_BY_ETAG_MAX_MATCHES`] candidates 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.
+    /// [`FIND_VERSION_BY_ETAG_MAX_MATCHES`] candidates per kind 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.
    ///
    /// Content identity is not version identity: two source generations with
-    /// the same bytes have the same ETag. Callers drop the candidates other
-    /// source versions already claim through their ledgers and refuse an
-    /// [`ReplicaLocation::Ambiguous`] remainder before mutating or deleting.
-    pub async fn replica_candidates_by_etag(
+    /// the same bytes have the same ETag, and two with the same size have the
+    /// same shape. Callers drop the candidates other source versions already
+    /// claim through their ledgers and decide through
+    /// [`ReplicaCandidates::locate`].
+    pub async fn replica_candidates(
        &self,
        bucket: &str,
        object: &str,
-        source_etag: &str,
-    ) -> Result<Vec<String>, Box<SdkError<aws_sdk_s3::operation::list_object_versions::ListObjectVersionsError>>> {
+        identity: &ReplicaIdentity<'_>,
+    ) -> Result<ReplicaCandidates, 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;
-        let mut matches: Vec<String> = Vec::new();
+        let mut candidates = ReplicaCandidates::default();
        for _ in 0..FIND_VERSION_BY_ETAG_MAX_PAGES {
            let page = self
                .client
@@ -2063,19 +2064,27 @@ impl TargetClient {
                .send()
                .await
                .map_err(Box::new)?;
-            matches.extend(
-                page.versions()
-                    .iter()
-                    .filter(|version| {
-                        version.key() == Some(object)
-                            && version.version_id().is_some_and(|id| !id.is_empty())
-                            && replication_etags_match(Some(source_etag), version.e_tag())
-                    })
-                    .filter_map(|version| version.version_id().map(str::to_string)),
-            );
+            for version in page.versions() {
+                let Some(version_id) = version.version_id().filter(|id| !id.is_empty()) else {
+                    continue;
+                };
+                if version.key() != Some(object) {
+                    continue;
+                }
+                match identity.evidence(version.e_tag(), version.size(), version.last_modified()) {
+                    Some(ReplicaEvidence::Etag) => candidates.exact.push(version_id.to_string()),
+                    // An unversioned target overwrites the key in place: its
+                    // `null` version holds whichever generation was written
+                    // last, so a shape match there identifies nothing.
+                    Some(ReplicaEvidence::Shape) if version_id != "null" => candidates.shaped.push(version_id.to_string()),
+                    Some(ReplicaEvidence::Shape) => {}
+                    None => {}
+                }
+            }
            // A listing that moved past the exact key (every listed key is >=
-            // the prefix), ended, or already filled the candidate cap decides.
-            if matches.len() >= FIND_VERSION_BY_ETAG_MAX_MATCHES
+            // the prefix), ended, or already filled a candidate cap decides.
+            if candidates.exact.len() >= FIND_VERSION_BY_ETAG_MAX_MATCHES
+                || candidates.shaped.len() >= FIND_VERSION_BY_ETAG_MAX_MATCHES
                || page
                    .versions()
                    .iter()
@@ -2090,8 +2099,9 @@ impl TargetClient {
                break;
            }
        }
-        matches.truncate(FIND_VERSION_BY_ETAG_MAX_MATCHES);
-        Ok(matches)
+        candidates.exact.truncate(FIND_VERSION_BY_ETAG_MAX_MATCHES);
+        candidates.shaped.truncate(FIND_VERSION_BY_ETAG_MAX_MATCHES);
+        Ok(candidates)
    }

    /// PutObjectRetention against a replica version on a target that does not
@@ -2614,19 +2624,18 @@ impl TargetClient {
 }

 /// Where a replica stands on a target that mints its own version ids, by
-/// content identity (exact key + ETag) after the candidates other source
-/// versions claim were removed. See
-/// [`TargetClient::replica_candidates_by_etag`].
+/// content identity after the candidates other source versions claim were
+/// removed. See [`TargetClient::replica_candidates`].
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum ReplicaLocation {
-    /// No live version under the key carries the source ETag.
+    /// No live version under the key matches the source.
    Missing,
-    /// Exactly one live version carries it: safe to address.
+    /// Exactly one live version matches: safe to address.
    Unique(String),
-    /// More than one live version carries it (same bytes replicated for
-    /// several source generations). `newest` is the most recently listed
-    /// one — good enough to prove the replica exists, never good enough to
-    /// pick which one to mutate or delete.
+    /// More than one live version carries the source ETag (same bytes
+    /// replicated for several source generations). `newest` is the most
+    /// recently listed one — good enough to prove the replica exists, never
+    /// good enough to pick which one to mutate or delete.
    Ambiguous { newest: String },
 }

@@ -2652,6 +2661,107 @@ impl ReplicaLocation {
    }
 }

+/// What a replica of an encrypted source version still has in common with
+/// it on a target that computes its own ETags. The source ETag is a keyed
+/// fingerprint (docs/operations/encrypted-etags.md) or a legacy plaintext
+/// MD5 the target's re-encrypted replica never reports, so the replica is
+/// recognised by the byte count the replication PUT sent and by a write time
+/// no earlier than the source version's own.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct EncryptedReplicaShape {
+    /// Bytes on the replication wire: the plaintext size for managed SSE,
+    /// the stored ciphertext size for an SSE-C passthrough.
+    pub size: i64,
+    /// The source version's mtime in whole seconds; listings and HEAD report
+    /// second-precision times.
+    pub not_before_secs: i64,
+}
+
+impl EncryptedReplicaShape {
+    pub fn matches(&self, size: Option<i64>, last_modified: Option<&aws_sdk_s3::primitives::DateTime>) -> bool {
+        size == Some(self.size) && last_modified.is_some_and(|written| written.secs() >= self.not_before_secs)
+    }
+}
+
+/// How a source version recognises its replica among a key's target
+/// versions.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ReplicaIdentity<'a> {
+    /// Unencrypted source: its ETag is the MD5 of the replicated bytes, which
+    /// every target reports back — exact identity.
+    Etag(&'a str),
+    /// Encrypted source: a target that adopts the source ETag (RustFS,
+    /// MinIO) still reports `etag`; any other target only keeps the shape.
+    Encrypted {
+        etag: Option<&'a str>,
+        shape: EncryptedReplicaShape,
+    },
+}
+
+/// Which kind of evidence tied a target version to the source.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ReplicaEvidence {
+    /// The target reports the source ETag.
+    Etag,
+    /// Encrypted source on a target that computes its own ETag: same size,
+    /// written no earlier than the source version.
+    Shape,
+}
+
+impl ReplicaIdentity<'_> {
+    pub fn evidence(
+        &self,
+        etag: Option<&str>,
+        size: Option<i64>,
+        last_modified: Option<&aws_sdk_s3::primitives::DateTime>,
+    ) -> Option<ReplicaEvidence> {
+        match *self {
+            Self::Etag(source_etag) => replication_etags_match(Some(source_etag), etag).then_some(ReplicaEvidence::Etag),
+            Self::Encrypted {
+                etag: source_etag,
+                shape,
+            } => {
+                if replication_etags_match(source_etag, etag) {
+                    Some(ReplicaEvidence::Etag)
+                } else {
+                    shape.matches(size, last_modified).then_some(ReplicaEvidence::Shape)
+                }
+            }
+        }
+    }
+}
+
+/// Content-identity candidates of one key, newest first, by evidence kind.
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct ReplicaCandidates {
+    /// Versions that report the source ETag.
+    pub exact: Vec<String>,
+    /// Encrypted sources only: versions without the source ETag whose shape
+    /// matches.
+    pub shaped: Vec<String>,
+}
+
+impl ReplicaCandidates {
+    pub fn is_empty(&self) -> bool {
+        self.exact.is_empty() && self.shaped.is_empty()
+    }
+
+    /// Drop the candidates other source versions claim and decide. An ETag
+    /// match outranks any shape match. A shape is weaker evidence than an
+    /// ETag — two generations of one size share it — so several shape
+    /// matches are never resolved by picking one: they read as
+    /// [`ReplicaLocation::Missing`] for convergence (the caller re-PUTs, a
+    /// duplicate at worst) and as ambiguous for a purge (it refuses).
+    pub fn locate(mut self, claimed: &HashSet<String>) -> (ReplicaLocation, ReplicaEvidence) {
+        self.exact.retain(|candidate| !claimed.contains(candidate));
+        if !self.exact.is_empty() {
+            return (ReplicaLocation::from_matches(self.exact), ReplicaEvidence::Etag);
+        }
+        self.shaped.retain(|candidate| !claimed.contains(candidate));
+        (ReplicaLocation::from_matches(self.shaped), ReplicaEvidence::Shape)
+    }
+}
+
 #[derive(Debug)]
 pub enum BucketTargetError {
    BucketRemoteTargetNotFound {
@@ -36,9 +36,9 @@ use super::replication_object_decision_boundary::{
    MustReplicateOptions, ReplicationMultipartPartInput, delete_marker_purge_mrf_entry, delete_marker_purge_version_id,
    delete_replication_creates_marker, delete_replication_target_version_id, heal_uses_delete_replication_path,
    is_object_lock_denied_delete, is_retryable_delete_replication_head_error, is_version_delete_replication,
-    replicate_delete_outcome, replication_etags_match, replication_multipart_complete_actual_size,
-    replication_multipart_part_plan, replication_single_put_size_error, resync_existing_delete_replication_info,
-    should_retry_delete_marker_purge, single_part_replica_etag_mismatch,
+    replicate_delete_outcome, replication_multipart_complete_actual_size, replication_multipart_part_plan,
+    replication_single_put_size_error, resync_existing_delete_replication_info, should_retry_delete_marker_purge,
+    single_part_replica_etag_mismatch,
 };
 use super::replication_queue_boundary::{DeletedObjectReplicationInfo, ReplicationQueueAdmission};
 use super::replication_resync_boundary::ResyncStatusType;
@@ -60,14 +60,15 @@ use super::replication_storage_boundary::{
 use super::replication_storage_boundary::{NamespaceLockFence, NamespaceLockSignalTestFence, ReplicationDeletedObject};
 use super::replication_tagging_boundary::ReplicationTagFilter;
 use super::replication_target_boundary::{
-    ERR_REPLICATION_SSEC_PASSTHROUGH_UNSUPPORTED, HeadObjectSdkError, PutObjectOptions, PutObjectPartOptions,
-    RemotePutObjectResponse, ReplicaLocation, ReplicationTargetStore, S3ClientError, SsecPassthroughCapability,
-    SsecPassthroughGate, TargetClient, VersionIdentityCapability, is_replication_target_offline_error,
-    replication_action_for_target_head, replication_complete_multipart_options, 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_source_etag_is_comparable,
-    replication_target_head_is_newer_null_version, resolve_read_api_version_id, ssec_passthrough_evidence_present,
-    ssec_passthrough_gate, version_identity_capability_from_put, version_identity_drifted,
+    ERR_REPLICATION_SSEC_PASSTHROUGH_UNSUPPORTED, EncryptedReplicaShape, HeadObjectSdkError, PutObjectOptions,
+    PutObjectPartOptions, RemotePutObjectResponse, ReplicaCandidates, ReplicaEvidence, ReplicaIdentity, ReplicaLocation,
+    ReplicationTargetStore, S3ClientError, SsecPassthroughCapability, SsecPassthroughGate, TargetClient,
+    VersionIdentityCapability, is_replication_target_offline_error, replication_action_for_target_head,
+    replication_complete_multipart_options, 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_source_etag_is_comparable, replication_target_head_is_newer_null_version,
+    resolve_read_api_version_id, ssec_passthrough_evidence_present, ssec_passthrough_gate, version_identity_capability_from_put,
+    version_identity_drifted,
 };
 use super::replication_versioning_boundary::ReplicationVersioningStore;
 use super::runtime_boundary as runtime_sources;
@@ -490,6 +491,7 @@ async fn head_object_fallback(
        Ok(head) => Ok(Some(LocatedReplica {
            version_id: head.version_id.clone(),
            head,
+            by: LocatedBy::CurrentVersion,
        })),
        Err(e) if head_object_not_found(&e) => Ok(None),
        Err(e) => Err(e),
@@ -503,6 +505,63 @@ async fn head_object_fallback(
 struct LocatedReplica {
    head: HeadObjectOutput,
    version_id: Option<String>,
+    by: LocatedBy,
+}
+
+/// How [`replica_head_fallback`] reached a replica, which decides what may
+/// confirm it (see [`located_replica_matches`]).
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum LocatedBy {
+    /// The target version the source ledger recorded for this version.
+    Ledger,
+    /// The one candidate content identity left (`ListObjectVersions`).
+    ContentIdentity,
+    /// Whatever version is current under the key (AWS-style fallback):
+    /// nothing ties it to this source version but an ETag match.
+    CurrentVersion,
+}
+
+/// How this source version recognises its replica on a target. `None` when
+/// the source carries nothing to compare (no ETag on a plaintext source).
+fn replica_identity(object_info: &ObjectInfo) -> Option<ReplicaIdentity<'_>> {
+    let etag = object_info.etag.as_deref().filter(|etag| !etag.trim().is_empty());
+    if replication_source_etag_is_comparable(&object_info.user_defined) {
+        return etag.map(ReplicaIdentity::Etag);
+    }
+    // Same byte count the replication PUT sends: SSE-C passes the stored
+    // ciphertext through, managed SSE sends the decrypted plaintext.
+    let size = if replication_object_is_ssec_encrypted(&object_info.user_defined) {
+        object_info.size
+    } else {
+        object_info.get_actual_size().ok()?
+    };
+    match object_info.mod_time {
+        Some(mod_time) if size >= 0 => Some(ReplicaIdentity::Encrypted {
+            etag,
+            shape: EncryptedReplicaShape {
+                size,
+                not_before_secs: mod_time.unix_timestamp(),
+            },
+        }),
+        _ => etag.map(ReplicaIdentity::Etag),
+    }
+}
+
+/// Whether a replica reached through [`replica_head_fallback`] holds this
+/// source version. An ETag match always does. For an encrypted source on a
+/// target that computes its own ETag, the shape does too — but only for a
+/// replica the ledger or content identity singled out, never for whichever
+/// version happens to be current. A plaintext source still needs its ETag.
+fn located_replica_matches(identity: Option<&ReplicaIdentity<'_>>, located: &LocatedReplica) -> bool {
+    let Some(identity) = identity else {
+        return false;
+    };
+    let head = &located.head;
+    match identity.evidence(head.e_tag(), head.content_length(), head.last_modified()) {
+        Some(ReplicaEvidence::Etag) => true,
+        Some(ReplicaEvidence::Shape) => located.by != LocatedBy::CurrentVersion,
+        None => false,
+    }
 }

 fn head_object_not_found(err: &HeadObjectSdkError) -> bool {
@@ -584,15 +643,20 @@ fn ledger_target_version_id(user_defined: &HashMap<String, String>, arn: &str) -
 /// 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.
+/// - AWS-style 400/403 (the RustFS uuid is rejected as malformed) and
 /// - 404 on a target that mints its own version ids (the Wasabi shape,
-///   rustfs/backlog#2340): the source id never existed there. HEAD the id
-///   the source ledger recorded for this target when it has one; otherwise,
-///   once the target is known to mint ids, 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.
+///   rustfs/backlog#2340): the source id never existed there.
+///
+/// Both HEAD the id the source ledger recorded for this target first. Without
+/// one, the AWS shape HEADs the current version without a version id
+/// (callers compare ETags); the Wasabi shape, once the target is known to
+/// mint ids, locates the replica by exact key and content identity through
+/// ListObjectVersions and HEADs the id the target assigned. An encrypted
+/// source takes that listing on the AWS shape too: its ETag never matches a
+/// target-computed one (docs/operations/encrypted-etags.md), so only a unique
+/// shape match can stand in for it. Without this, every heal, MRF retry and
+/// existing-object resync re-drive PUTs the object again and mints one more
+/// target version.
 ///
 /// `content_identity` is the sibling-ledger view used to disown candidates
 /// other source versions already claim; `None` disables content identity
@@ -609,7 +673,7 @@ struct ReplicaSource<'a> {
    bucket: &'a str,
    object: &'a str,
    version_id: Option<Uuid>,
-    etag: Option<&'a str>,
+    identity: Option<ReplicaIdentity<'a>>,
    ledger_version_id: Option<&'a str>,
 }

@@ -623,13 +687,11 @@ async fn replica_head_fallback(
        bucket: source_bucket,
        object,
        version_id: source_version_id,
-        etag: source_etag,
+        identity,
        ledger_version_id,
    } = source;
-    if is_version_id_format_mismatch(err) {
-        return Some(head_object_fallback(tgt_client, object).await);
-    }
-    if !head_object_not_found(err) {
+    let format_mismatch = is_version_id_format_mismatch(err);
+    if !format_mismatch && !head_object_not_found(err) {
        return None;
    }
    if let Some(ledger_version_id) = ledger_version_id {
@@ -641,24 +703,45 @@ async fn replica_head_fallback(
                return Some(Ok(Some(LocatedReplica {
                    head,
                    version_id: Some(ledger_version_id.to_string()),
+                    by: LocatedBy::Ledger,
                })));
            }
            // The recorded version is gone: fall through to content identity
            // before concluding that the replica is missing.
-            Err(e) if head_object_not_found(&e) => {}
+            Err(e) if head_object_not_found(&e) || is_version_id_format_mismatch(&e) => {}
            Err(e) => return Some(Err(e)),
        }
    }
-    if !ReplicationTargetStore::version_identity_capability(&tgt_client.arn).version_addressing_unreliable() {
+    let encrypted = matches!(identity, Some(ReplicaIdentity::Encrypted { .. }));
+    if format_mismatch {
+        if !(encrypted && content_identity.is_some()) {
+            return Some(head_object_fallback(tgt_client, object).await);
+        }
+    } else if !ReplicationTargetStore::version_identity_capability(&tgt_client.arn).version_addressing_unreliable() {
        return None;
    }
    let siblings = content_identity?;
-    let etag = source_etag.filter(|etag| !etag.trim().is_empty())?;
-    Some(match tgt_client.replica_candidates_by_etag(&tgt_client.bucket, object, etag).await {
+    let identity = identity?;
+    let candidates = match tgt_client.replica_candidates(&tgt_client.bucket, object, &identity).await {
+        Ok(candidates) => Ok(candidates),
+        // A target that refused the version-addressed HEAD (a bodiless 403
+        // can be a permission denial) may refuse the listing too: keep the
+        // current-version fallback, which re-PUTs, instead of stalling.
+        Err(_) if format_mismatch => return Some(head_object_fallback(tgt_client, object).await),
+        Err(list_err) => Err(list_err),
+    };
+    Some(match candidates {
        Ok(candidates) => {
-            let location =
+            let (location, evidence) =
                disowned_replica_location(siblings, &tgt_client.arn, source_bucket, object, source_version_id, candidates).await;
-            match location.any_version_id() {
+            // Several same-ETag versions all hold the source bytes, so any of
+            // them proves the replica exists. Several same-shape versions do
+            // not: re-PUT rather than adopt the wrong generation.
+            let assigned_version_id = match (&location, evidence) {
+                (ReplicaLocation::Ambiguous { .. }, ReplicaEvidence::Shape) => None,
+                _ => location.any_version_id(),
+            };
+            match assigned_version_id {
                Some(assigned_version_id) => {
                    debug!(
                        event = EVENT_REPLICATION_DRIFTED_REPLICA_LOCATED,
@@ -669,6 +752,7 @@ async fn replica_head_fallback(
                        arn = %tgt_client.arn,
                        assigned_version_id = %assigned_version_id,
                        ambiguous = matches!(location, ReplicaLocation::Ambiguous { .. }),
+                        evidence = ?evidence,
                        "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.to_string()))
@@ -677,6 +761,7 @@ async fn replica_head_fallback(
                        Ok(head) => Ok(Some(LocatedReplica {
                            head,
                            version_id: Some(assigned_version_id.to_string()),
+                            by: LocatedBy::ContentIdentity,
                        })),
                        // The located version disappeared between LIST and HEAD.
                        Err(e) if head_object_not_found(&e) => Ok(None),
@@ -699,16 +784,15 @@ async fn disowned_replica_location(
    source_bucket: &str,
    object: &str,
    source_version_id: Option<Uuid>,
-    mut candidates: Vec<String>,
-) -> ReplicaLocation {
+    candidates: ReplicaCandidates,
+) -> (ReplicaLocation, ReplicaEvidence) {
    if candidates.is_empty() {
-        return ReplicaLocation::Missing;
+        return (ReplicaLocation::Missing, ReplicaEvidence::Etag);
    }
    let claimed = siblings
        .claimed_target_versions(source_bucket, object, source_version_id, arn)
        .await;
-    candidates.retain(|candidate| !claimed.contains(candidate));
-    ReplicaLocation::from_matches(candidates)
+    candidates.locate(&claimed)
 }

 /// The id the target assigned to a located replica when it is not the source
@@ -1855,10 +1939,14 @@ fn spawn_resync_walk_task<S: ReplicationStorage>(

 /// Classify the target HEAD verification result for one resynced object,
 /// updating the per-object status counters and returning the accounted size
-/// together with any verification error.
+/// together with any verification error. `source` is the source version as
+/// stored after the resync replicated it (its ledger and encryption decide
+/// how the replica is recognised); without it the queued ETag is all there
+/// is to go on.
 async fn verify_resync_head_result(
    head_result: std::result::Result<HeadObjectOutput, HeadObjectSdkError>,
    roi: &ReplicateObjectInfo,
+    source: Option<&ObjectInfo>,
    st: &mut TargetReplicationResyncStatus,
    target_client: &Arc<TargetClient>,
 ) -> (i64, Option<HeadObjectSdkError>) {
@@ -1900,14 +1988,23 @@ async fn verify_resync_head_result(
            // 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`).
+            let ledger_version_id = source.and_then(|info| ledger_target_version_id(&info.user_defined, &target_client.arn));
+            let identity = match source {
+                Some(info) => replica_identity(info),
+                None => roi
+                    .etag
+                    .as_deref()
+                    .filter(|etag| !etag.trim().is_empty())
+                    .map(ReplicaIdentity::Etag),
+            };
            match replica_head_fallback(
                target_client.as_ref(),
                ReplicaSource {
                    bucket: &roi.bucket,
                    object: &roi.name,
                    version_id: roi.version_id,
-                    etag: roi.etag.as_deref(),
-                    ledger_version_id: None,
+                    identity,
+                    ledger_version_id: ledger_version_id.as_deref(),
                },
                Some(&NoSiblingLedger),
                &err,
@@ -1932,6 +2029,24 @@ async fn verify_resync_head_result(
    }
 }

+/// The resynced source version as stored now, ledger included, for the
+/// verification fallback. `None` when it cannot be read.
+async fn resync_source_object_info<S: ReplicationStorage>(storage: &Arc<S>, roi: &ReplicateObjectInfo) -> Option<ObjectInfo> {
+    storage
+        .get_object_info(
+            &roi.bucket,
+            &roi.name,
+            &ObjectOptions {
+                version_id: roi.version_id.map(|version_id| version_id.to_string()),
+                versioned: ReplicationVersioningStore::prefix_enabled(&roi.bucket, &roi.name).await,
+                version_suspended: ReplicationVersioningStore::prefix_suspended(&roi.bucket, &roi.name).await,
+                ..Default::default()
+            },
+        )
+        .await
+        .ok()
+}
+
 /// Replicate one existing object (or delete marker / version purge) to the
 /// resync target, verify the outcome via a target HEAD, and produce the
 /// per-object resync status update.
@@ -1966,7 +2081,12 @@ async fn resync_worker_process_object<S: ReplicationStorage>(
        roi.version_id.map(|v| v.to_string()),
    )
    .await;
-    let (size, err) = verify_resync_head_result(head_result, &roi, &mut st, target_client).await;
+    let source = if head_result.is_err() && !roi.delete_marker {
+        resync_source_object_info(storage, &roi).await
+    } else {
+        None
+    };
+    let (size, err) = verify_resync_head_result(head_result, &roi, source.as_ref(), &mut st, target_client).await;

    if err.is_some() {
        debug!(
@@ -3363,6 +3483,9 @@ fn abandoned_purge_target_info(bucket: &str, dobj: &DeletedObjectReplicationInfo
 /// to resolve the replica on a target that mints its own version ids.
 struct VersionPurgeSource {
    etag: Option<String>,
+    /// Encrypted sources: what identifies the replica when the target
+    /// computes its own ETag (see [`replica_identity`]).
+    encrypted_shape: Option<EncryptedReplicaShape>,
    ledger: HashMap<String, String>,
    ledger_corrupt: bool,
    /// Target versions other generations of the key claim, per target ARN:
@@ -3408,8 +3531,13 @@ async fn version_purge_source<S: ReplicationStorage>(
            claimed_by_siblings.insert(target.arn.clone(), claimed);
        }
    }
+    let encrypted_shape = match replica_identity(&info) {
+        Some(ReplicaIdentity::Encrypted { shape, .. }) => Some(shape),
+        _ => None,
+    };
    Some(VersionPurgeSource {
        etag: info.etag,
+        encrypted_shape,
        ledger,
        ledger_corrupt,
        claimed_by_siblings,
@@ -3427,7 +3555,8 @@ enum PurgeReplicaResolution {
    /// goal is already met there.
    Absent,
    /// Refuse: no ledger entry and more than one target version carries the
-    /// source ETag, so any pick could destroy a live generation.
+    /// source ETag (or, for an encrypted source on a target that computes its
+    /// own ETag, its shape), so any pick could destroy a live generation.
    Ambiguous,
    /// Refuse: the persisted ledger is inconsistent.
    LedgerCorrupt,
@@ -3455,15 +3584,19 @@ async fn resolve_purge_replica(
    if !force_lookup && !ReplicationTargetStore::version_identity_capability(&tgt_client.arn).version_addressing_unreliable() {
        return PurgeReplicaResolution::SourceVersion;
    }
-    let Some(etag) = source.etag.as_deref().filter(|etag| !etag.trim().is_empty()) else {
-        return PurgeReplicaResolution::SourceVersion;
+    let etag = source.etag.as_deref().filter(|etag| !etag.trim().is_empty());
+    let identity = match (source.encrypted_shape, etag) {
+        (Some(shape), etag) => ReplicaIdentity::Encrypted { etag, shape },
+        (None, Some(etag)) => ReplicaIdentity::Etag(etag),
+        (None, None) => return PurgeReplicaResolution::SourceVersion,
    };
-    match tgt_client.replica_candidates_by_etag(&tgt_client.bucket, object, etag).await {
-        Ok(mut candidates) => {
-            if let Some(claimed) = source.claimed_by_siblings.get(&tgt_client.arn) {
-                candidates.retain(|candidate| !claimed.contains(candidate));
-            }
-            match ReplicaLocation::from_matches(candidates) {
+    match tgt_client.replica_candidates(&tgt_client.bucket, object, &identity).await {
+        Ok(candidates) => {
+            let no_claims = HashSet::new();
+            let claimed = source.claimed_by_siblings.get(&tgt_client.arn).unwrap_or(&no_claims);
+            // Several shape matches stay ambiguous here: a purge never picks
+            // one to delete.
+            match candidates.locate(claimed).0 {
                ReplicaLocation::Unique(assigned) => PurgeReplicaResolution::Resolved(assigned),
                ReplicaLocation::Missing => PurgeReplicaResolution::Absent,
                ReplicaLocation::Ambiguous { .. } => PurgeReplicaResolution::Ambiguous,
@@ -4517,13 +4650,14 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
                // generation with the same bytes would be mistaken for one.
                // Only the ledger (a lost response of this very version) and
                // the AWS-style format fallback apply here.
+                let identity = replica_identity(&object_info);
                if let Some(fallback) = replica_head_fallback(
                    &tgt_client,
                    ReplicaSource {
                        bucket: &bucket,
                        object: &object,
                        version_id: self.version_id,
-                        etag: object_info.etag.as_deref(),
+                        identity,
                        ledger_version_id: ledger_version_id.as_deref(),
                    },
                    None,
@@ -4532,9 +4666,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
                .await
                {
                    match fallback {
-                        Ok(Some(located))
-                            if replication_etags_match(object_info.etag.as_deref(), located.head.e_tag.as_deref()) =>
-                        {
+                        Ok(Some(located)) if located_replica_matches(identity.as_ref(), &located) => {
                            if ssec_audit_required
                                && !settle_ssec_passthrough_evidence(&located.head, &tgt_client, &bucket, &object, &mut rinfo)
                                    .await
@@ -5182,13 +5314,14 @@ async fn resolve_replicate_all_action(
        }
        Err(e) => {
            let ledger_version_id = ledger_target_version_id(&object_info.user_defined, &tgt_client.arn);
+            let identity = replica_identity(&object_info);
            if let Some(fallback) = replica_head_fallback(
                tgt_client,
                ReplicaSource {
                    bucket,
                    object,
                    version_id: roi.version_id,
-                    etag: object_info.etag.as_deref(),
+                    identity,
                    ledger_version_id: ledger_version_id.as_deref(),
                },
                Some(siblings),
@@ -5199,20 +5332,20 @@ async fn resolve_replicate_all_action(
                match fallback {
                    Ok(Some(located)) => {
                        let oi = &located.head;
-                        let etags_match = replication_etags_match(object_info.etag.as_deref(), oi.e_tag.as_deref());
-                        if require_existing_target && !etags_match {
+                        let replica_matches = located_replica_matches(identity.as_ref(), &located);
+                        if require_existing_target && !replica_matches {
                            rinfo.error = Some("replica metadata target does not contain matching object data".to_string());
                            rinfo.duration = (OffsetDateTime::now_utc() - start_time).unsigned_abs();
                            return None;
                        }
-                        if etags_match
+                        if replica_matches
                            && ssec_audit_required
                            && !settle_ssec_passthrough_evidence(oi, tgt_client, bucket, object, rinfo).await
                        {
                            rinfo.duration = (OffsetDateTime::now_utc() - start_time).unsigned_abs();
                            return None;
                        }
-                        if etags_match && let Some(assigned) = drifted_replica_version_id(&located, roi.version_id) {
+                        if replica_matches && let Some(assigned) = drifted_replica_version_id(&located, roi.version_id) {
                            // The replica lives under an id the target minted:
                            // record it (the ledger every later mutation
                            // resolves through) and apply metadata in place —
@@ -5245,7 +5378,7 @@ async fn resolve_replicate_all_action(
                            // records the sync without running the PUT.
                            replication_action = ReplicationAction::None;
                        } else {
-                            replication_action = if etags_match {
+                            replication_action = if replica_matches {
                                ReplicationAction::None
                            } else {
                                ReplicationAction::All
@@ -6655,13 +6788,361 @@ mod tests {

        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;
+        let (size, err) = verify_resync_head_result(head_result, &roi, None, &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");
    }

+    /// What a target that computes its own ETag (AWS S3) reports for the
+    /// replica of an encrypted source: the MD5 of the plaintext it received,
+    /// never the source's encrypted-ETag fingerprint ([`DRIFTED_ETAG`] here).
+    const COMPUTED_REPLICA_ETAG: &str = "0f0e0d0c0b0a09080706050403020100";
+    /// The replica's write time on the scripted targets below.
+    const REPLICA_WRITTEN: &str = "2026-09-06T10:00:00.000Z";
+    const REPLICA_WRITTEN_HTTP: &str = "Sun, 06 Sep 2026 10:00:00 GMT";
+
+    /// An SSE-S3 source version: 4 plaintext bytes, stored as ciphertext, its
+    /// ETag a keyed fingerprint, written at `written` (unix seconds).
+    fn encrypted_roi_and_object(written: i64) -> (ReplicateObjectInfo, ObjectInfo) {
+        let (roi, mut object_info) = drifted_roi_and_object();
+        object_info.size = 36;
+        object_info.actual_size = 4;
+        object_info.mod_time = Some(OffsetDateTime::from_unix_timestamp(written).expect("valid timestamp"));
+        object_info.user_defined = HashMap::from([("x-amz-server-side-encryption".to_string(), "AES256".to_string())]).into();
+        (roi, object_info)
+    }
+
+    fn replica_written_secs() -> i64 {
+        OffsetDateTime::parse(REPLICA_WRITTEN, &Rfc3339)
+            .expect("valid replica time")
+            .unix_timestamp()
+    }
+
+    fn computed_etag_versions_response(versions: &[(&str, i64)]) -> String {
+        let versions = versions
+            .iter()
+            .enumerate()
+            .map(|(index, (version_id, size))| {
+                format!(
+                    "<Version><Key>object</Key><VersionId>{version_id}</VersionId><IsLatest>{}</IsLatest><LastModified>{REPLICA_WRITTEN}</LastModified><ETag>&quot;{COMPUTED_REPLICA_ETAG}&quot;</ETag><Size>{size}</Size><StorageClass>STANDARD</StorageClass></Version>",
+                    index == 0
+                )
+            })
+            .collect::<String>();
+        xml_response(
+            "200 OK",
+            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>{versions}</ListVersionsResult>"
+            ),
+        )
+    }
+
+    fn computed_etag_head_response(version_id: &str, size: i64) -> String {
+        format!(
+            "HTTP/1.1 200 OK\r\nETag: \"{COMPUTED_REPLICA_ETAG}\"\r\nContent-Length: {size}\r\nx-amz-version-id: {version_id}\r\nLast-Modified: {REPLICA_WRITTEN_HTTP}\r\nConnection: close\r\n\r\n"
+        )
+    }
+
+    /// A Wasabi/AWS-shaped target that computes its own ETags: 404 on the
+    /// source uuid, the listed versions, and a HEAD for any listed id.
+    fn spawn_computed_etag_target(
+        requests: usize,
+        versions: &'static [(&'static str, i64)],
+    ) -> (String, std::thread::JoinHandle<Vec<String>>) {
+        spawn_scripted_target_server(requests, move |line| {
+            if line.starts_with("GET ") && line.contains("versions") {
+                return computed_etag_versions_response(versions);
+            }
+            if line.starts_with("HEAD ")
+                && let Some((version_id, size)) = versions
+                    .iter()
+                    .find(|(version_id, _)| line.contains(&format!("versionId={version_id}")))
+            {
+                return computed_etag_head_response(version_id, *size);
+            }
+            if line.starts_with("HEAD ") {
+                return empty_response("404 Not Found");
+            }
+            empty_response("500 Unexpected")
+        })
+    }
+
+    async fn resolve_heal_action(
+        target: &Arc<TargetClient>,
+        roi: &ReplicateObjectInfo,
+        object_info: ObjectInfo,
+        rinfo: &mut ReplicatedTargetInfo,
+    ) -> Option<(ReplicationAction, ObjectInfo)> {
+        resolve_replicate_all_action(
+            ReplicateAllActionContext {
+                siblings: &NoSiblingLedger,
+                roi,
+                tgt_client: target,
+                bucket: &roi.bucket,
+                object: &roi.name,
+                start_time: OffsetDateTime::now_utc(),
+                ssec_audit_required: false,
+            },
+            object_info,
+            rinfo,
+        )
+        .await
+    }
+
+    #[tokio::test]
+    async fn heal_redrive_locates_encrypted_replica_by_shape_on_target_that_computes_own_etags() {
+        static VERSIONS: &[(&str, i64)] = &[(DRIFTED_ASSIGNED_VERSION_ID, 4)];
+        let (endpoint, server) = spawn_computed_etag_target(3, VERSIONS);
+        let target = test_target_client(endpoint);
+        ReplicationTargetStore::record_version_identity_capability(&target.arn, VersionIdentityCapability::MintsOwn);
+        let (roi, object_info) = encrypted_roi_and_object(replica_written_secs() - 1);
+        let mut rinfo = replicate_all_target_info(&roi, &target);
+
+        let action = resolve_heal_action(&target, &roi, object_info, &mut rinfo).await;
+
+        assert!(
+            matches!(action, Some((ReplicationAction::None, _))),
+            "an encrypted replica located by its shape must not be re-driven: {action:?}"
+        );
+        assert!(rinfo.error.is_none(), "{:?}", rinfo.error);
+        assert_eq!(rinfo.target_version_id.as_deref(), Some(DRIFTED_ASSIGNED_VERSION_ID));
+        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:?}");
+    }
+
+    #[tokio::test]
+    async fn heal_redrive_does_not_adopt_an_encrypted_replica_of_another_size_or_an_older_write() {
+        // Size differs from the 4 plaintext bytes.
+        static OTHER_SIZE: &[(&str, i64)] = &[(DRIFTED_ASSIGNED_VERSION_ID, 36)];
+        let (endpoint, server) = spawn_computed_etag_target(2, OTHER_SIZE);
+        let target = test_target_client(endpoint);
+        ReplicationTargetStore::record_version_identity_capability(&target.arn, VersionIdentityCapability::MintsOwn);
+        let (roi, object_info) = encrypted_roi_and_object(replica_written_secs() - 1);
+        let mut rinfo = replicate_all_target_info(&roi, &target);
+        let action = resolve_heal_action(&target, &roi, object_info, &mut rinfo).await;
+        assert!(matches!(action, Some((ReplicationAction::All, _))), "{action:?}");
+        server.join().expect("test HTTP server should finish");
+
+        // Same size, but written before the source version existed.
+        static SAME_SIZE: &[(&str, i64)] = &[(DRIFTED_ASSIGNED_VERSION_ID, 4)];
+        let (endpoint, server) = spawn_computed_etag_target(2, SAME_SIZE);
+        let target = test_target_client(endpoint);
+        ReplicationTargetStore::record_version_identity_capability(&target.arn, VersionIdentityCapability::MintsOwn);
+        let (roi, object_info) = encrypted_roi_and_object(replica_written_secs() + 1);
+        let mut rinfo = replicate_all_target_info(&roi, &target);
+        let action = resolve_heal_action(&target, &roi, object_info, &mut rinfo).await;
+        assert!(matches!(action, Some((ReplicationAction::All, _))), "{action:?}");
+        server.join().expect("test HTTP server should finish");
+    }
+
+    #[tokio::test]
+    async fn heal_redrive_does_not_pick_among_several_encrypted_shape_matches() {
+        static VERSIONS: &[(&str, i64)] = &[(DRIFTED_ASSIGNED_VERSION_ID, 4), ("another-minted-id", 4)];
+        let (endpoint, server) = spawn_computed_etag_target(2, VERSIONS);
+        let target = test_target_client(endpoint);
+        ReplicationTargetStore::record_version_identity_capability(&target.arn, VersionIdentityCapability::MintsOwn);
+        let (roi, object_info) = encrypted_roi_and_object(replica_written_secs() - 1);
+        let mut rinfo = replicate_all_target_info(&roi, &target);
+
+        let action = resolve_heal_action(&target, &roi, object_info, &mut rinfo).await;
+
+        assert!(
+            matches!(action, Some((ReplicationAction::All, _))),
+            "two same-shape versions are not evidence of either: {action:?}"
+        );
+        assert!(rinfo.target_version_id.is_none(), "{:?}", rinfo.target_version_id);
+        let seen = server.join().expect("test HTTP server should finish");
+        assert_eq!(seen.len(), 2, "HEAD by source id, ListObjectVersions, no HEAD of a guess: {seen:?}");
+    }
+
+    #[tokio::test]
+    async fn heal_redrive_keeps_exact_etag_identity_for_plaintext_sources() {
+        // A same-size replica whose ETag differs is not this plaintext
+        // version's replica: the shape criterion is for encrypted sources only.
+        static VERSIONS: &[(&str, i64)] = &[(DRIFTED_ASSIGNED_VERSION_ID, 4)];
+        let (endpoint, server) = spawn_computed_etag_target(2, VERSIONS);
+        let target = test_target_client(endpoint);
+        ReplicationTargetStore::record_version_identity_capability(&target.arn, VersionIdentityCapability::MintsOwn);
+        let (roi, mut object_info) = drifted_roi_and_object();
+        object_info.mod_time = Some(OffsetDateTime::from_unix_timestamp(replica_written_secs() - 1).expect("valid timestamp"));
+        let mut rinfo = replicate_all_target_info(&roi, &target);
+
+        let action = resolve_heal_action(&target, &roi, object_info, &mut rinfo).await;
+
+        assert!(matches!(action, Some((ReplicationAction::All, _))), "{action:?}");
+        server.join().expect("test HTTP server should finish");
+    }
+
+    #[tokio::test]
+    async fn heal_redrive_never_adopts_an_unversioned_null_version_by_shape() {
+        // An unversioned target's `null` version holds whatever generation
+        // was written last; a same-size newer generation must not be claimed.
+        static VERSIONS: &[(&str, i64)] = &[("null", 4)];
+        let (endpoint, server) = spawn_computed_etag_target(2, VERSIONS);
+        let target = test_target_client(endpoint);
+        ReplicationTargetStore::record_version_identity_capability(&target.arn, VersionIdentityCapability::MintsOwn);
+        let (roi, object_info) = encrypted_roi_and_object(replica_written_secs() - 1);
+        let mut rinfo = replicate_all_target_info(&roi, &target);
+
+        let action = resolve_heal_action(&target, &roi, object_info, &mut rinfo).await;
+
+        assert!(matches!(action, Some((ReplicationAction::All, _))), "{action:?}");
+        assert!(rinfo.target_version_id.is_none(), "{:?}", rinfo.target_version_id);
+        let seen = server.join().expect("test HTTP server should finish");
+        assert_eq!(seen.len(), 2, "HEAD by source id, ListObjectVersions, nothing adopted: {seen:?}");
+    }
+
+    #[tokio::test]
+    async fn encrypted_lookup_keeps_the_current_version_fallback_when_listing_is_denied() {
+        // A bodiless 403 on the version-addressed HEAD may be a permission
+        // denial; the same credentials then refuse ListObjectVersions. The
+        // object must still be re-driven, as before content identity existed.
+        let (endpoint, server) = spawn_scripted_target_server(3, |line| {
+            let listing = line.starts_with("GET ") && line.contains("versions");
+            let version_addressed_head = line.starts_with("HEAD ") && line.contains("versionId=");
+            if listing || version_addressed_head {
+                empty_response("403 Forbidden")
+            } else if line.starts_with("HEAD ") {
+                computed_etag_head_response(DRIFTED_ASSIGNED_VERSION_ID, 4)
+            } else {
+                empty_response("500 Unexpected")
+            }
+        });
+        let target = test_target_client(endpoint);
+        let (roi, object_info) = encrypted_roi_and_object(replica_written_secs() - 1);
+        let mut rinfo = replicate_all_target_info(&roi, &target);
+
+        let action = resolve_heal_action(&target, &roi, object_info, &mut rinfo).await;
+
+        assert!(
+            matches!(action, Some((ReplicationAction::All, _))),
+            "a denied listing must fall back to the re-PUT, not stall: {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 of the current version: {seen:?}"
+        );
+        assert!(seen[2].starts_with("HEAD ") && !seen[2].contains("versionId="), "{}", seen[2]);
+    }
+
+    #[tokio::test]
+    async fn encrypted_ledger_replica_converges_on_an_aws_style_target() {
+        // AWS rejects the RustFS uuid as malformed (400). The ledger names the
+        // replica; its computed ETag cannot confirm it, its shape does.
+        let (endpoint, server) = spawn_scripted_target_server(2, |line| {
+            if line.starts_with("HEAD ") && line.contains(&format!("versionId={DRIFTED_ASSIGNED_VERSION_ID}")) {
+                computed_etag_head_response(DRIFTED_ASSIGNED_VERSION_ID, 4)
+            } else if line.starts_with("HEAD ") {
+                empty_response("400 Bad Request")
+            } else {
+                empty_response("500 Unexpected")
+            }
+        });
+        let target = test_target_client(endpoint);
+        let (roi, mut object_info) = encrypted_roi_and_object(replica_written_secs() - 1);
+        let mut user_defined = (*object_info.user_defined).clone();
+        insert_str(
+            &mut user_defined,
+            &format!("{SUFFIX_REPLICATION_TARGET_VERSION_ARN_PREFIX}{}", target.arn),
+            DRIFTED_ASSIGNED_VERSION_ID.to_string(),
+        );
+        object_info.user_defined = Arc::new(user_defined);
+        let mut rinfo = replicate_all_target_info(&roi, &target);
+
+        let action = resolve_heal_action(&target, &roi, object_info, &mut rinfo).await;
+
+        assert!(matches!(action, Some((ReplicationAction::None, _))), "{action:?}");
+        let seen = server.join().expect("test HTTP server should finish");
+        assert_eq!(seen.len(), 2, "HEAD by source id, HEAD by the ledger id: {seen:?}");
+        assert!(seen[1].contains(DRIFTED_ASSIGNED_VERSION_ID), "{}", seen[1]);
+    }
+
+    #[tokio::test]
+    async fn resync_verification_counts_encrypted_replica_by_shape() {
+        static VERSIONS: &[(&str, i64)] = &[(DRIFTED_ASSIGNED_VERSION_ID, 4)];
+        let (endpoint, server) = spawn_computed_etag_target(3, VERSIONS);
+        let target = test_target_client(endpoint);
+        ReplicationTargetStore::record_version_identity_capability(&target.arn, VersionIdentityCapability::MintsOwn);
+        let (roi, object_info) = encrypted_roi_and_object(replica_written_secs() - 1);
+        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 (_, err) = verify_resync_head_result(head_result, &roi, Some(&object_info), &mut st, &target).await;
+
+        assert!(err.is_none(), "{err:?}");
+        assert_eq!((st.replicated_count, st.failed_count), (1, 0));
+        server.join().expect("test HTTP server should finish");
+    }
+
+    fn encrypted_purge_source() -> Arc<VersionPurgeSource> {
+        let (_, object_info) = encrypted_roi_and_object(replica_written_secs() - 1);
+        let Some(ReplicaIdentity::Encrypted { shape, .. }) = replica_identity(&object_info) else {
+            panic!("an SSE-S3 source must be identified by its shape");
+        };
+        Arc::new(VersionPurgeSource {
+            etag: object_info.etag,
+            encrypted_shape: Some(shape),
+            ledger: HashMap::new(),
+            ledger_corrupt: false,
+            claimed_by_siblings: HashMap::new(),
+        })
+    }
+
+    #[tokio::test]
+    async fn encrypted_version_purge_without_ledger_deletes_the_unique_shape_match() {
+        let (endpoint, server) = spawn_scripted_target_server(2, |line| {
+            if line.starts_with("GET ") && line.contains("versions") {
+                computed_etag_versions_response(&[(DRIFTED_ASSIGNED_VERSION_ID, 4)])
+            } else if line.starts_with("DELETE ") && line.contains(&format!("versionId={DRIFTED_ASSIGNED_VERSION_ID}")) {
+                empty_response("204 No Content")
+            } else {
+                empty_response("500 Unexpected")
+            }
+        });
+        let target = test_target_client(endpoint);
+        register_test_target(&target).await;
+        ReplicationTargetStore::record_version_identity_capability(&target.arn, VersionIdentityCapability::MintsOwn);
+        let dobj = version_purge_dobj(&target.arn);
+
+        let rinfo = replicate_delete_to_target(&dobj, target.clone(), Some(encrypted_purge_source())).await;
+
+        assert_eq!(rinfo.version_purge_status, VersionPurgeStatusType::Complete, "{:?}", rinfo.error);
+        let seen = server.join().expect("test HTTP server should finish");
+        assert_eq!(seen.len(), 2, "ListObjectVersions, then DELETE by the located id: {seen:?}");
+        assert!(
+            seen[1].starts_with("DELETE ") && seen[1].contains(DRIFTED_ASSIGNED_VERSION_ID),
+            "{}",
+            seen[1]
+        );
+    }
+
+    #[tokio::test]
+    async fn encrypted_version_purge_refuses_several_shape_matches() {
+        let (endpoint, server) = spawn_scripted_target_server(1, |line| {
+            if line.starts_with("GET ") && line.contains("versions") {
+                computed_etag_versions_response(&[(DRIFTED_ASSIGNED_VERSION_ID, 4), ("another-minted-id", 4)])
+            } else {
+                empty_response("500 Unexpected")
+            }
+        });
+        let target = test_target_client(endpoint);
+        register_test_target(&target).await;
+        ReplicationTargetStore::record_version_identity_capability(&target.arn, VersionIdentityCapability::MintsOwn);
+        let dobj = version_purge_dobj(&target.arn);
+
+        let rinfo = replicate_delete_to_target(&dobj, target.clone(), Some(encrypted_purge_source())).await;
+
+        assert_eq!(rinfo.version_purge_status, VersionPurgeStatusType::Failed);
+        let seen = server.join().expect("test HTTP server should finish");
+        assert_eq!(seen.len(), 1, "no DELETE of a guessed replica: {seen:?}");
+    }
+
    fn delete_marker_roi() -> ReplicateObjectInfo {
        ReplicateObjectInfo {
            bucket: "source".to_string(),
@@ -6886,7 +7367,7 @@ mod tests {
            None,
            "the fixture must reproduce the codeless 405 the SDK yields for a bodiless HEAD"
        );
-        let (size, err) = verify_resync_head_result(head_result, &roi, &mut st, &target).await;
+        let (size, err) = verify_resync_head_result(head_result, &roi, None, &mut st, &target).await;

        assert!(err.is_none(), "{err:?}");
        assert_eq!((size, st.replicated_count, st.failed_count), (0, 1, 0));
@@ -6913,7 +7394,7 @@ mod tests {

            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;
+            let (size, err) = verify_resync_head_result(head_result, &roi, None, &mut st, &target).await;

            assert!(err.is_some(), "a 405 must not count as a completed purge ({status:?})");
            assert_eq!((size, st.replicated_count, st.failed_count), (0, 0, 1), "{status:?}");
@@ -6931,7 +7412,7 @@ mod tests {

        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;
+        let (size, err) = verify_resync_head_result(head_result, &roi, None, &mut st, &target).await;

        assert!(err.is_none(), "{err:?}");
        assert_eq!((size, st.replicated_count, st.failed_count), (0, 1, 0));
@@ -6948,7 +7429,7 @@ mod tests {

        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;
+        let (size, err) = verify_resync_head_result(head_result, &roi, None, &mut st, &target).await;

        assert!(err.is_some(), "a 503 must not count as a propagated delete marker");
        assert_eq!((size, st.replicated_count, st.failed_count), (0, 0, 1));
@@ -6974,7 +7455,7 @@ mod tests {

        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;
+        let (size, err) = verify_resync_head_result(head_result, &roi, None, &mut st, &target).await;

        assert!(err.is_none(), "{err:?}");
        assert_eq!((size, st.replicated_count, st.failed_count), (0, 1, 0));
@@ -6999,7 +7480,7 @@ mod tests {

        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;
+        let (size, err) = verify_resync_head_result(head_result, &roi, None, &mut st, &target).await;

        assert!(err.is_some(), "a live latest version must not verify the marker");
        assert_eq!((size, st.replicated_count, st.failed_count), (0, 0, 1));
@@ -7017,7 +7498,7 @@ mod tests {

        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;
+        let (size, err) = verify_resync_head_result(head_result, &roi, None, &mut st, &target).await;

        assert!(err.is_some(), "a 403 must not count as a propagated delete marker");
        assert_eq!((size, st.replicated_count, st.failed_count), (0, 0, 1));
@@ -7134,6 +7615,7 @@ mod tests {
    fn purge_source(ledger: &[(&str, &str)]) -> Arc<VersionPurgeSource> {
        Arc::new(VersionPurgeSource {
            etag: Some(DRIFTED_ETAG.to_string()),
+            encrypted_shape: None,
            ledger: ledger
                .iter()
                .map(|(arn, version_id)| (arn.to_string(), version_id.to_string()))
@@ -7338,6 +7820,7 @@ mod tests {
        let dobj = version_purge_dobj(&target.arn);
        let source = Arc::new(VersionPurgeSource {
            etag: Some(DRIFTED_ETAG.to_string()),
+            encrypted_shape: None,
            ledger: HashMap::new(),
            ledger_corrupt: false,
            claimed_by_siblings: HashMap::from([(target.arn.clone(), HashSet::from([DRIFTED_ASSIGNED_VERSION_ID.to_string()]))]),
@@ -7425,6 +7908,7 @@ mod tests {
        let dobj = version_purge_dobj(&target.arn);
        let source = Arc::new(VersionPurgeSource {
            etag: Some(DRIFTED_ETAG.to_string()),
+            encrypted_shape: None,
            ledger: HashMap::new(),
            ledger_corrupt: true,
            claimed_by_siblings: HashMap::new(),
@@ -37,8 +37,9 @@ use time::OffsetDateTime;
 use time::format_description::well_known::Rfc3339;

 pub(crate) use crate::bucket::bucket_target_sys::{
-    AdvancedPutOptions, HeadObjectSdkError, PutObjectOptions, PutObjectPartOptions, RemotePutObjectResponse, RemoveObjectOptions,
-    ReplicaLocation, S3ClientError, TargetClient, resolve_read_api_version_id,
+    AdvancedPutOptions, EncryptedReplicaShape, HeadObjectSdkError, PutObjectOptions, PutObjectPartOptions,
+    RemotePutObjectResponse, RemoveObjectOptions, ReplicaCandidates, ReplicaEvidence, ReplicaIdentity, ReplicaLocation,
+    S3ClientError, TargetClient, resolve_read_api_version_id,
 };
 #[cfg(test)]
 pub(crate) use crate::bucket::target::BucketTarget;
@@ -36,6 +36,32 @@ the replication PUT either, so a target that transforms their payload is not
 detected. See
 `docs/operations/replication-outbound-transport.md`.

+The same mismatch affects finding an existing replica on a target that mints
+its own version ids (AWS S3, Wasabi), which a heal, MRF retry or
+existing-object resync does before writing again:
+
+- A replica recorded in the source's target-version ledger is addressed by
+  that id, on a 404 and on an AWS-style 400 for the source id alike. For an
+  encrypted source the replica counts when its size matches the bytes
+  replication sent (plaintext for SSE-S3 / SSE-KMS, ciphertext for SSE-C) and
+  it was written no earlier than the source version.
+- Without a ledger entry (for example a PUT whose response was lost), the
+  replica is located through `ListObjectVersions` by the same size and
+  write-time rule, after removing versions that other generations of the key
+  already claim. Only a single remaining match is adopted and recorded in the
+  ledger. Several matches are never resolved by picking one: convergence writes
+  the replica again, and a version purge is refused with the usual backoff.
+  The `null` version of an unversioned target is never matched this way, and
+  a target that refuses the listing keeps the earlier current-version check,
+  which writes the replica again.
+- Plaintext sources keep exact ETag identity. A target that adopts the source
+  ETag still matches encrypted sources by ETag first.
+
+Two same-size generations of a key whose own ledgers were both lost can still
+be confused, and a version purge with no ledger and no shape match reports the
+replica as already gone. Delete a leftover replica on the target by hand if a
+`replication-check` VersionFidelity report shows one.
+
 ## Existing data and migration

 The upgrade does not rewrite existing object metadata or erase fingerprints
@@ -18,7 +18,7 @@
 | Rejects or mis-stores `aws-chunked` bodies (SeaweedFS 3.97) | Handled by the plain-payload default above. | Outbound target matrix, `RejectAwsChunked` mode |
 | Requires `Content-MD5` or `x-amz-checksum-*` on a PutObject with Object Lock parameters (AWS S3, MinIO, Impossible Cloud, most compatible stores) | Satisfied: a locked single PUT carries `Content-MD5` derived from the source ETag (plaintext objects whose ETag is the MD5 of the wire bytes) or an SDK CRC32 checksum (multipart-layout ETags, managed SSE, SSE-C passthrough — this one is an `aws-chunked` trailer, so a target that also rejects that framing cannot take such objects). Releases before this fix (`1.0.0-rc.5`) need `RUSTFS_REPLICATION_STREAMING_CHECKSUMS=true` as a workaround. | Outbound target matrix, `RequireChecksumWithObjectLock` mode |
 | Stores `x-amz-checksum-*` from a PutObject and returns it on `HEAD ?ChecksumMode=ENABLED` (AWS S3, Wasabi, RustFS) | Satisfied for single-part objects: the replica answers with the source's checksum. Before this fix (`1.0.0-rc.5`) the checksum left the source as `x-amz-meta-<algorithm>` user metadata and no replica carried it (rustfs/backlog#2340). | Outbound target matrix, `Checksummed` shape |
-| Mints its own version ids (AWS S3, Wasabi, Impossible Cloud) | Data lands, and later mutations converge through the target-version ledger: the id the target assigned is recorded on the source version (internal key `replication-target-version-<arn>`) and version deletes, tag and Object Lock updates address it (tag/retention/legal-hold through the metadata-only APIs, never a re-PUT). A replica written before the ledger existed is located by exact key + ETag, minus the candidates other generations of the key already claim; an ambiguous remainder is refused with a one-hour backoff (`replication_purge_replica_unresolved`). `NoSuchVersion` on a version-addressed DELETE counts as purged. See rustfs/backlog#2340, rustfs/backlog#2085 and `docs/operations/replication-check.md` (VersionFidelity). | `replication-check`, outbound target matrix, `MintOwnVersionIds` mode (also models Wasabi's 404 `NoSuchVersion` on unknown ids) |
+| Mints its own version ids (AWS S3, Wasabi, Impossible Cloud) | Data lands, and later mutations converge through the target-version ledger: the id the target assigned is recorded on the source version (internal key `replication-target-version-<arn>`) and version deletes, tag and Object Lock updates address it (tag/retention/legal-hold through the metadata-only APIs, never a re-PUT). A replica the ledger does not name (written before the ledger existed, or a PUT whose response was lost) is located by exact key + ETag, minus the candidates other generations of the key already claim; an ambiguous remainder is refused with a one-hour backoff (`replication_purge_replica_unresolved`). An encrypted source whose ETag the target does not report is matched by size and write time instead, and only a unique match is adopted (`docs/operations/encrypted-etags.md`). `NoSuchVersion` on a version-addressed DELETE counts as purged. See rustfs/backlog#2340, rustfs/backlog#2085 and `docs/operations/replication-check.md` (VersionFidelity). | `replication-check`, outbound target matrix, `MintOwnVersionIds` mode (also models Wasabi's 404 `NoSuchVersion` on unknown ids); `matrix_mint_own_version_ids_encrypted_redrive_does_not_duplicate` for SSE-S3 on an ETag-computing target |
 | Ignores `x-rustfs-source-etag` / `x-minio-source-etag` and reports the MD5 of the received plaintext as the ETag (AWS S3, generic S3 services) | Plaintext single-part replicas are verified against the source ETag. Encrypted sources skip the comparison (see above). In `1.0.2`, the first release with encrypted-ETag fingerprints, every single-part SSE-S3 object (and SSE-KMS where the target does not echo `aws:kms`) failed with `replica etag mismatch` and was retried indefinitely; the only workaround there is `RUSTFS_REPLICATION_REPLICA_ETAG_VERIFY=false`, which disables the check for every object. RustFS and MinIO targets adopt the source ETag and were not affected. | Outbound target matrix, `ComputesOwnEtags` mode, `SseS3` shape |
 | Returns an ETag that is not the content MD5 without announcing SSE | Every single-part object fails ETag verification. Set `RUSTFS_REPLICATION_REPLICA_ETAG_VERIFY=false`. | Replication status FAILED with `replica etag mismatch` |