fix(replication): forward single-part object checksums as headers

crates/e2e_test/src/replication_extension_test.rs

@@ -368,23 +368,23 @@ impl Drop for SlowReplicationTargetGuard {
 // Mirrors madmin-go `ResyncTargetsInfo`/`ResyncTarget` json tags — the same
 // shape `mc replicate resync status` decodes.
 #[derive(Debug, Clone, serde::Deserialize)]
-pub(crate) struct ReplicationResetStatusResponse {
+struct ReplicationResetStatusResponse {
     #[serde(rename = "target", default)]
-    pub(crate) targets: Vec<ReplicationResetStatusTarget>,
+    targets: Vec<ReplicationResetStatusTarget>,
 }
 
 #[derive(Debug, Clone, serde::Deserialize)]
-pub(crate) struct ReplicationResetStatusTarget {
+struct ReplicationResetStatusTarget {
     #[serde(rename = "arn", default)]
-    pub(crate) arn: String,
+    arn: String,
     #[serde(rename = "resetid", default)]
-    pub(crate) reset_id: String,
+    reset_id: String,
     #[serde(rename = "resyncStatus", default)]
-    pub(crate) status: String,
+    status: String,
     #[serde(rename = "replicationCount", default)]
-    pub(crate) replicated_count: i64,
+    replicated_count: i64,
     #[serde(rename = "object", default)]
-    pub(crate) object: String,
+    object: String,
 }
 
 fn extract_xml_tag(xml: &str, tag: &str) -> Option<String> {
@@ -2294,7 +2294,7 @@ async fn site_replication_state_edit(
 /// return the target `(arn, reset_id)`, asserting the response carries the
 /// madmin `ResyncTargetsInfo` shape (`target[0].arn` / `target[0].resetid`)
 /// that `mc replicate resync start` decodes.
-pub(crate) async fn start_bucket_replication_reset(
+async fn start_bucket_replication_reset(
     env: &RustFSTestEnvironment,
     bucket: &str,
 ) -> Result<(String, String), Box<dyn Error + Send + Sync>> {
@@ -2314,7 +2314,7 @@ pub(crate) async fn start_bucket_replication_reset(
     Ok((arn, reset_id))
 }
 
-pub(crate) async fn get_replication_reset_status(
+async fn get_replication_reset_status(
     env: &RustFSTestEnvironment,
     bucket: &str,
     arn: &str,
@@ -3837,6 +3837,164 @@ async fn test_bucket_replication_converges_delete_marker_and_version_purge() ->
     Ok(())
 }
 
+/// Regression for rustfs/backlog#2340 (not Wasabi specific): a directory
+/// marker (`prefix/` with a body) in a versioned bucket is stored as the null
+/// version, like MinIO (`putOpts`: "for directory objects skip creating new
+/// versions"), and must still replicate to completion instead of staying
+/// `PENDING`.
+#[tokio::test]
+async fn test_bucket_replication_replicates_directory_marker_in_versioned_bucket() -> TestResult {
+    init_logging();
+
+    let mut source_env = RustFSTestEnvironment::new().await?;
+    let mut source_env_vars = replication_fast_env();
+    source_env_vars.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV);
+    source_env.start_rustfs_server_with_env(vec![], &source_env_vars).await?;
+
+    let mut target_env = RustFSTestEnvironment::new().await?;
+    target_env.start_rustfs_server_without_cleanup(vec![]).await?;
+
+    let source_bucket = "replication-dir-marker-src";
+    let target_bucket = "replication-dir-marker-dst";
+    let source_client = source_env.create_s3_client();
+    let target_client = target_env.create_s3_client();
+
+    source_client.create_bucket().bucket(source_bucket).send().await?;
+    target_client.create_bucket().bucket(target_bucket).send().await?;
+    enable_bucket_versioning(&source_env, source_bucket).await?;
+    enable_bucket_versioning(&target_env, target_bucket).await?;
+    let target_arn = set_replication_target(&source_env, source_bucket, &target_env, target_bucket).await?;
+    put_bucket_replication(&source_env, source_bucket, &target_arn).await?;
+
+    let marker_key = "dir/trailing/";
+    let body = b"directory marker body";
+    let put = source_client
+        .put_object()
+        .bucket(source_bucket)
+        .key(marker_key)
+        .body(ByteStream::from_static(body))
+        .send()
+        .await?;
+    assert!(
+        put.version_id()
+            .is_none_or(|id| id == "null" || id == uuid::Uuid::nil().to_string()),
+        "a directory marker is the null version even in a versioned bucket: {:?}",
+        put.version_id()
+    );
+
+    wait_for_source_replication_status(&source_client, source_bucket, marker_key, "COMPLETED", false).await?;
+
+    let replica = target_client
+        .get_object()
+        .bucket(target_bucket)
+        .key(marker_key)
+        .send()
+        .await?;
+    assert_eq!(replica.body.collect().await?.into_bytes().as_ref(), body);
+    let listed = target_client
+        .list_object_versions()
+        .bucket(target_bucket)
+        .prefix(marker_key)
+        .send()
+        .await?;
+    let marker_versions: Vec<_> = listed.versions().iter().filter(|v| v.key() == Some(marker_key)).collect();
+    assert_eq!(marker_versions.len(), 1, "the marker must land exactly once: {marker_versions:?}");
+    assert_eq!(
+        marker_versions[0].version_id(),
+        Some("null"),
+        "the replica keeps the null version identity"
+    );
+
+    Ok(())
+}
+
+/// Regression for rustfs/backlog#2340 (not Wasabi specific): a single-part
+/// object uploaded with `x-amz-checksum-*` must reach the target with the same
+/// checksum. The outbound options keyed the stored record by algorithm name,
+/// which the target client sent as `x-amz-meta-*` user metadata, so a replica
+/// never carried a checksum although the source HEAD returned one.
+#[tokio::test]
+async fn test_bucket_replication_forwards_single_part_object_checksums() -> TestResult {
+    init_logging();
+
+    let mut source_env = RustFSTestEnvironment::new().await?;
+    let mut source_env_vars = replication_fast_env();
+    source_env_vars.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV);
+    source_env.start_rustfs_server_with_env(vec![], &source_env_vars).await?;
+
+    let mut target_env = RustFSTestEnvironment::new().await?;
+    target_env.start_rustfs_server_without_cleanup(vec![]).await?;
+
+    let source_bucket = "replication-checksum-src";
+    let target_bucket = "replication-checksum-dst";
+    let source_client = source_env.create_s3_client();
+    let target_client = target_env.create_s3_client();
+
+    source_client.create_bucket().bucket(source_bucket).send().await?;
+    target_client.create_bucket().bucket(target_bucket).send().await?;
+    enable_bucket_versioning(&source_env, source_bucket).await?;
+    enable_bucket_versioning(&target_env, target_bucket).await?;
+    let target_arn = set_replication_target(&source_env, source_bucket, &target_env, target_bucket).await?;
+    put_bucket_replication(&source_env, source_bucket, &target_arn).await?;
+
+    let body = b"123456789";
+    let crc32_key = "checksum-crc32.txt";
+    let sha256_key = "checksum-sha256.txt";
+    let crc32_put = source_client
+        .put_object()
+        .bucket(source_bucket)
+        .key(crc32_key)
+        .body(ByteStream::from_static(body))
+        .checksum_algorithm(aws_sdk_s3::types::ChecksumAlgorithm::Crc32)
+        .send()
+        .await?;
+    let expected_crc32 = crc32_put.checksum_crc32().ok_or("source PUT omitted CRC32")?.to_string();
+    let sha256_put = source_client
+        .put_object()
+        .bucket(source_bucket)
+        .key(sha256_key)
+        .body(ByteStream::from_static(body))
+        .checksum_algorithm(aws_sdk_s3::types::ChecksumAlgorithm::Sha256)
+        .send()
+        .await?;
+    let expected_sha256 = sha256_put.checksum_sha256().ok_or("source PUT omitted SHA256")?.to_string();
+
+    for key in [crc32_key, sha256_key] {
+        wait_for_source_replication_status(&source_client, source_bucket, key, "COMPLETED", false).await?;
+    }
+
+    let replica = target_client
+        .head_object()
+        .bucket(target_bucket)
+        .key(crc32_key)
+        .checksum_mode(aws_sdk_s3::types::ChecksumMode::Enabled)
+        .send()
+        .await?;
+    assert_eq!(replica.checksum_crc32(), Some(expected_crc32.as_str()), "replica lost the CRC32 checksum");
+    let replica = target_client
+        .head_object()
+        .bucket(target_bucket)
+        .key(sha256_key)
+        .checksum_mode(aws_sdk_s3::types::ChecksumMode::Enabled)
+        .send()
+        .await?;
+    assert_eq!(
+        replica.checksum_sha256(),
+        Some(expected_sha256.as_str()),
+        "replica lost the SHA256 checksum"
+    );
+    // The bare algorithm name must not leak as user metadata either.
+    assert!(
+        replica
+            .metadata()
+            .is_none_or(|meta| !meta.keys().any(|k| k.eq_ignore_ascii_case("sha256"))),
+        "replica carries the checksum as user metadata: {:?}",
+        replica.metadata()
+    );
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn test_bucket_replication_disabled_delete_marker_does_not_propagate() -> TestResult {
     init_logging();

crates/e2e_test/src/replication_target_matrix_test.rs

@@ -33,16 +33,17 @@
 
 use crate::common::{init_logging, replication_fast_env};
 use crate::fake_s3_target::{BucketMode, FAKE_ACCESS_KEY, FAKE_SECRET_KEY};
-use crate::fake_s3_target::{FakeS3Target, FaultAction as FakeTargetFault, Operation as FakeTargetOperation, RequestRecord};
+use crate::fake_s3_target::{FakeS3Target, Operation as FakeTargetOperation, RequestRecord};
 use crate::on_demand_migration::common::{OdmEnvOptions, OdmTestEnv, fake_source_client};
 use crate::replication_extension_test::{
-    LOOPBACK_REPLICATION_TARGET_ENV, ReplicationTargetOptions, enable_bucket_versioning, get_replication_reset_status,
-    put_bucket_replication, set_replication_target_with_options, start_bucket_replication_reset,
+    LOOPBACK_REPLICATION_TARGET_ENV, ReplicationTargetOptions, enable_bucket_versioning, put_bucket_replication,
+    set_replication_target_with_options,
 };
 use aws_sdk_s3::Client;
 use aws_sdk_s3::primitives::{ByteStream, DateTime};
 use aws_sdk_s3::types::{
-    Checksum, CompletedMultipartUpload, CompletedPart, ObjectAttributes, ObjectLockLegalHoldStatus, ObjectLockMode,
+    Checksum, ChecksumAlgorithm, CompletedMultipartUpload, CompletedPart, ObjectAttributes, ObjectLockLegalHoldStatus,
+    ObjectLockMode,
 };
 use bytes::Bytes;
 use std::error::Error;
@@ -114,10 +115,13 @@ enum ObjectShape {
     LockedMultipart,
     /// ODM stores two local parts while preserving a single-PUT source's MD5 ETag.
     OdmPreservedMd5Multipart,
+    /// Single-part object uploaded with `x-amz-checksum-sha256`; the replica
+    /// must carry the same header (rustfs/backlog#2340).
+    Checksummed,
 }
 
 impl ObjectShape {
-    const ALL: [ObjectShape; 7] = [
+    const ALL: [ObjectShape; 8] = [
         ObjectShape::Empty,
         ObjectShape::Plain,
         ObjectShape::Retention,
@@ -125,6 +129,7 @@ impl ObjectShape {
         ObjectShape::Multipart,
         ObjectShape::LockedMultipart,
         ObjectShape::OdmPreservedMd5Multipart,
+        ObjectShape::Checksummed,
     ];
 
     fn key(self) -> &'static str {
@@ -136,6 +141,16 @@ impl ObjectShape {
             ObjectShape::Multipart => "matrix/multipart.bin",
             ObjectShape::LockedMultipart => "matrix/locked-multipart.bin",
             ObjectShape::OdmPreservedMd5Multipart => "matrix/odm-preserved-md5.bin",
+            ObjectShape::Checksummed => "matrix/checksummed.bin",
+        }
+    }
+
+    /// The `x-amz-checksum-*` header the source stored and every upload of
+    /// the replica must repeat.
+    fn forwarded_checksum_header(self) -> Option<&'static str> {
+        match self {
+            ObjectShape::Checksummed => Some("x-amz-checksum-sha256"),
+            _ => None,
         }
     }
 
@@ -198,6 +213,18 @@ impl ObjectShape {
             ObjectShape::Multipart => multipart_put(client, bucket, key, 0x44, false).await,
             ObjectShape::LockedMultipart => multipart_put(client, bucket, key, 0x55, true).await,
             ObjectShape::OdmPreservedMd5Multipart => odm_preserved_md5_multipart(env, bucket, key).await,
+            ObjectShape::Checksummed => {
+                let body = payload(40 * 1024, 0x66);
+                client
+                    .put_object()
+                    .bucket(bucket)
+                    .key(key)
+                    .body(ByteStream::from(body.clone()))
+                    .checksum_algorithm(ChecksumAlgorithm::Sha256)
+                    .send()
+                    .await?;
+                Ok(body)
+            }
         }
     }
 }
@@ -227,170 +254,6 @@ fn expectation(mode: TargetMode, shape: ObjectShape) -> Expectation {
         .unwrap_or(Expectation::Completed)
 }
 
-/// rustfs/backlog#2340: a target that mints its own version ids (Wasabi,
-/// AWS S3) answers 404 to a HEAD by the source uuid, which the worker used to
-/// read as "replica missing" and re-drive the PUT — one more target version
-/// per heal, MRF retry or resync. Two re-drive shapes, both must converge on
-/// the single version the first PUT created:
-/// - 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.
-#[tokio::test]
-async fn matrix_mint_own_version_ids_redrive_does_not_duplicate() -> TestResult {
-    init_logging();
-
-    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);
-
-    let mut env_vars = replication_fast_env();
-    env_vars.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV);
-    env_vars.extend_from_slice(&[
-        ("NO_PROXY", "127.0.0.1,localhost"),
-        ("HTTP_PROXY", ""),
-        ("HTTPS_PROXY", ""),
-        // The scanner heal pass is what re-drives a FAILED object.
-        ("RUSTFS_SCANNER_CYCLE", "1"),
-        ("RUSTFS_SCANNER_START_DELAY_SECS", "1"),
-    ]);
-    let env = OdmTestEnv::start_with(OdmEnvOptions {
-        env: env_vars,
-        ..OdmEnvOptions::default()
-    })
-    .await?;
-    let source_env = &env.rustfs;
-
-    let source_bucket = "matrix-mint-own-redrive-src";
-    let source_client = source_env.create_s3_client();
-    source_client
-        .create_bucket()
-        .bucket(source_bucket)
-        .object_lock_enabled_for_bucket(true)
-        .send()
-        .await?;
-    enable_bucket_versioning(source_env, source_bucket).await?;
-    let target_arn = set_replication_target_with_options(
-        source_env,
-        source_bucket,
-        ReplicationTargetOptions {
-            endpoint: &target.address(),
-            access_key: FAKE_ACCESS_KEY,
-            secret_key: FAKE_SECRET_KEY,
-            target_bucket: &target_bucket,
-            secure: false,
-            skip_tls_verify: false,
-            ca_cert_pem: None,
-        },
-    )
-    .await?;
-    put_bucket_replication(source_env, source_bucket, &target_arn).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?;
-    assert_eq!(
-        wait_for_terminal_replication_status(&source_client, source_bucket, probe_key).await?,
-        "COMPLETED"
-    );
-
-    // 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?;
-    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?;
-    assert_eq!(
-        wait_for_terminal_replication_status(&source_client, source_bucket, resync_key).await?,
-        "COMPLETED"
-    );
-    let (reset_arn, _reset_id) = start_bucket_replication_reset(source_env, source_bucket).await?;
-    assert_eq!(reset_arn, target_arn);
-    let resync = async {
-        loop {
-            let status = get_replication_reset_status(source_env, source_bucket, &target_arn).await?;
-            if let Some(entry) = status.targets.iter().find(|entry| entry.arn == target_arn)
-                && entry.status == "Completed"
-            {
-                return Ok::<_, Box<dyn Error + Send + Sync>>(entry.replicated_count);
-            }
-            sleep(Duration::from_millis(250)).await;
-        }
-    };
-    let replicated = timeout(Duration::from_secs(90), resync)
-        .await
-        .map_err(|_| "existing-object resync did not complete within 90 seconds")??;
-    assert!(replicated >= 3, "resync must count the located replicas as replicated, got {replicated}");
-    for key in [probe_key, heal_key, resync_key] {
-        let versions = target.stored_versions(&target_bucket, key);
-        assert_eq!(
-            versions.len(),
-            1,
-            "{key}: a re-drive against a target that mints its own version ids must not mint another one: {versions:?}"
-        );
-    }
-
-    target.shutdown().await;
-    Ok(())
-}
-
-/// Wait until `key` is COMPLETED on the source and, for the observation
-/// window after that, the target still holds exactly one live version of it.
-async fn wait_for_replication_status_and_single_version(
-    source_client: &Client,
-    source_bucket: &str,
-    target: &FakeS3Target,
-    target_bucket: &str,
-    key: &str,
-) -> TestResult {
-    // The lost PUT response first settles the object FAILED; only the next
-    // scanner heal pass can turn that into COMPLETED, so FAILED is transient
-    // here and the wait is for COMPLETED alone.
-    let converged = async {
-        loop {
-            let head = source_client.head_object().bucket(source_bucket).key(key).send().await?;
-            if head.replication_status().is_some_and(|status| status.as_str() == "COMPLETED") {
-                return Ok::<_, Box<dyn Error + Send + Sync>>(());
-            }
-            sleep(Duration::from_millis(250)).await;
-        }
-    };
-    timeout(Duration::from_secs(90), converged)
-        .await
-        .map_err(|_| format!("{key}: heal re-drive did not converge to COMPLETED within 90 seconds"))??;
-    // The heal pass keeps visiting the key for a few scanner cycles; a
-    // duplicate would show up here as a second stored version.
-    for _ in 0..12 {
-        let versions = target.stored_versions(target_bucket, key);
-        assert_eq!(versions.len(), 1, "{key}: target minted another version on re-drive: {versions:?}");
-        sleep(Duration::from_millis(500)).await;
-    }
-    Ok(())
-}
-
 #[tokio::test]
 async fn matrix_baseline_target() -> TestResult {
     run_row(TargetMode::Baseline).await
@@ -614,6 +477,19 @@ async fn check_completed_cell(
     }) {
         return Err(format!("a locked PutObject went out without any integrity header (rustfs#7082): {bare:?}").into());
     }
+    // rustfs/backlog#2340 contract: a source checksum reaches the target as
+    // the `x-amz-checksum-*` header, not as user metadata; every PutObject of
+    // the shape carries it.
+    if let Some(header) = shape.forwarded_checksum_header()
+        && let Some(missing) = uploads.iter().find(|record| {
+            record.operation == FakeTargetOperation::PutObject
+                && !record.transport.checksum_headers.iter().any(|name| name == header)
+        })
+    {
+        return Err(
+            format!("a PutObject went out without the source's {header} header (rustfs/backlog#2340): {missing:?}").into(),
+        );
+    }
     Ok(())
 }
 

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, VersionIdentityCapability, append_version_id_query,
+            SsecPassthroughCapability, TargetClient, 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, replication_etags_match};
+use crate::bucket::replication::{ObjectLockIntegrity, object_lock_put_integrity};
 use crate::bucket::replication::{ReplicationStatusType, ReplicationTargetConfigBridge};
 use crate::bucket::target::ARN;
 use crate::bucket::target::BucketTargetType;
@@ -126,22 +126,6 @@ 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>>;
@@ -365,13 +349,6 @@ 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.
 ///
@@ -398,11 +375,6 @@ 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
@@ -451,7 +423,6 @@ 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())),
@@ -775,40 +746,10 @@ 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.
@@ -1221,32 +1162,12 @@ 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);
             }
         }
@@ -1971,62 +1892,6 @@ 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`).
     ///
@@ -2199,7 +2064,15 @@ impl TargetClient {
             }
         }
 
-        match builder
+        // A forwarded source checksum is this PUT's integrity header. In
+        // streaming-checksum mode (`RUSTFS_REPLICATION_STREAMING_CHECKSUMS`)
+        // the SDK would still add its default CRC32 trailer, and a target that
+        // receives both keeps the trailer's algorithm: a forwarded SHA256
+        // vanished from the replica while the source reported COMPLETED. Pin
+        // this request to WhenRequired so nothing is sent beside the source's
+        // own checksum.
+        let forwards_source_checksum = headers.keys().any(|name| name.as_str().starts_with("x-amz-checksum-"));
+        let mut operation = builder
             .bucket(bucket)
             .key(object)
             .content_length(size)
@@ -2219,10 +2092,14 @@ impl TargetClient {
                 }
 
                 Result::<_, aws_smithy_types::error::operation::BuildError>::Ok(req)
-            })
-            .send()
-            .await
-        {
+            });
+        if forwards_source_checksum {
+            operation = operation.config_override(
+                aws_sdk_s3::config::Builder::new()
+                    .request_checksum_calculation(aws_sdk_s3::config::RequestChecksumCalculation::WhenRequired),
+            );
+        }
+        match operation.send().await {
             Ok(output) => {
                 // Under SSE-KMS/DSSE or SSE-C the target's ETag is not the MD5
                 // of the stored plaintext, so it cannot be compared against the
@@ -2642,13 +2519,21 @@ mod tests {
     }
 
     fn header_recording_target_client(response_headers: Vec<(String, String)>) -> (TargetClient, RecordedHeaders) {
+        header_recording_target_client_with_checksums(response_headers, replication_request_checksum_calculation())
+    }
+
+    fn header_recording_target_client_with_checksums(
+        response_headers: Vec<(String, String)>,
+        checksums: RequestChecksumCalculation,
+    ) -> (TargetClient, RecordedHeaders) {
         let request_headers: RecordedHeaders = Arc::new(std::sync::Mutex::new(Vec::new()));
         let connector = SharedHttpConnector::new(RecordingHeaderConnector {
             request_headers: Arc::clone(&request_headers),
             response_headers,
         });
         let http_client = http_client_fn(move |_settings, _components| connector.clone());
-        let client = s3_client_for_test(443, Some(http_client));
+        let client =
+            s3_client_for_endpoint_test_with_checksums("https://localhost:443".to_string(), Some(http_client), checksums);
         (
             TargetClient {
                 endpoint: "https://localhost:443".to_string(),
@@ -2815,6 +2700,47 @@ mod tests {
         }
     }
 
+    /// With streaming checksums enabled the SDK adds a CRC32 trailer to every
+    /// upload. A PUT that forwards the source's checksum must not get that
+    /// second algorithm: a target that receives both keeps the trailer's and
+    /// the forwarded SHA256 never reaches the replica (rustfs/backlog#2340).
+    #[tokio::test]
+    async fn streaming_put_object_with_forwarded_checksum_sends_no_sdk_checksum() {
+        let (client, recorded) =
+            header_recording_target_client_with_checksums(Vec::new(), RequestChecksumCalculation::WhenSupported);
+        let mut forwarded = PutObjectOptions::default();
+        forwarded.user_metadata.insert(
+            "x-amz-checksum-sha256".to_string(),
+            "OoJ3yNhRwv3wwtZoGqEIPrPX9xwTnfLl+ka0wStN1g0=".to_string(),
+        );
+        client
+            .put_object("target-bucket", "object", 4, streaming_test_body(b"data"), &forwarded)
+            .await
+            .expect("recorded put_object should succeed");
+        client
+            .put_object("target-bucket", "object", 4, streaming_test_body(b"data"), &PutObjectOptions::default())
+            .await
+            .expect("recorded put_object should succeed");
+        let recorded = recorded.lock().expect("recorded header lock should not be poisoned");
+        let with_forwarded = &recorded[0];
+        assert_eq!(
+            recorded_header(with_forwarded, "x-amz-checksum-sha256"),
+            Some("OoJ3yNhRwv3wwtZoGqEIPrPX9xwTnfLl+ka0wStN1g0=")
+        );
+        assert_eq!(
+            recorded_header(with_forwarded, "x-amz-trailer"),
+            None,
+            "the SDK must not add a trailer checksum"
+        );
+        assert_eq!(recorded_header(with_forwarded, "x-amz-sdk-checksum-algorithm"), None);
+        // Control: the same client still streams a trailer when nothing is forwarded.
+        let without_forwarded = &recorded[1];
+        assert!(
+            recorded_header(without_forwarded, "x-amz-trailer").is_some(),
+            "streaming mode must still apply to uploads without a forwarded checksum: {without_forwarded:?}"
+        );
+    }
+
     /// A forwarded source checksum already satisfies the rule; nothing is added.
     #[tokio::test]
     async fn locked_put_object_keeps_a_forwarded_source_checksum() {
@@ -3180,6 +3106,14 @@ mod tests {
     }
 
     fn s3_client_for_endpoint_test(endpoint: String, http_client: Option<SharedHttpClient>) -> S3Client {
+        s3_client_for_endpoint_test_with_checksums(endpoint, http_client, replication_request_checksum_calculation())
+    }
+
+    fn s3_client_for_endpoint_test_with_checksums(
+        endpoint: String,
+        http_client: Option<SharedHttpClient>,
+        checksums: RequestChecksumCalculation,
+    ) -> S3Client {
         let credentials = SdkCredentials::builder()
             .access_key_id("test-access")
             .secret_access_key("test-secret")
@@ -3193,7 +3127,7 @@ mod tests {
             .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest())
             // Mirror the production remote-target builder so recorded requests
             // exercise the same checksum/framing behavior (#6853).
-            .request_checksum_calculation(replication_request_checksum_calculation());
+            .request_checksum_calculation(checksums);
         if let Some(http_client) = http_client {
             config = config.http_client(http_client);
         }
@@ -3356,64 +3290,6 @@ 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,7 +66,6 @@ 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,
@@ -89,6 +88,5 @@ 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,8 +56,6 @@ 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,
@@ -65,7 +63,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_capability_from_put, version_identity_drifted,
+    ssec_passthrough_gate, version_identity_drifted,
 };
 use super::replication_versioning_boundary::ReplicationVersioningStore;
 use super::runtime_boundary as runtime_sources;
@@ -125,7 +123,6 @@ 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";
 
@@ -335,12 +332,6 @@ 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;
     }
@@ -413,70 +404,11 @@ 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 head_object_not_found(&e) => Ok(None),
+        Err(e) if e.as_service_error().is_some_and(|se| se.is_not_found()) || has_raw_status(&e, 404) => 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
@@ -1468,26 +1400,31 @@ async fn verify_resync_head_result(
                 (0, None)
             }
         }
-        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(_))) => {
+        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(_)) => {
                     st.replicated_count += 1;
                     st.replicated_size += roi.size;
                     (roi.size, None)
                 }
-                Some(Ok(None)) | None => {
+                Ok(None) => {
                     st.failed_count += 1;
                     (0, Some(err))
                 }
-                Some(Err(e2)) => {
+                Err(e2) => {
                     st.failed_count += 1;
                     (0, Some(e2))
                 }
             }
         }
+        Err(err) => {
+            st.failed_count += 1;
+            (0, Some(err))
+        }
     }
 }
 
@@ -3660,8 +3597,11 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
                 }
             }
             Err(e) => {
-                if let Some(fallback) = replica_head_fallback(&tgt_client, &object, object_info.etag.as_deref(), &e).await {
-                    match fallback {
+                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 {
                         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
@@ -3691,8 +3631,6 @@ 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!(
@@ -4292,8 +4230,9 @@ async fn resolve_replicate_all_action(
             }
         }
         Err(e) => {
-            if let Some(fallback) = replica_head_fallback(tgt_client, object, object_info.etag.as_deref(), &e).await {
-                match fallback {
+            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 {
                     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 {
@@ -4345,7 +4284,7 @@ async fn resolve_replicate_all_action(
                         return None;
                     }
                 }
-            } else if head_object_not_found(&e) {
+            } else if e.as_service_error().is_some_and(|se| se.is_not_found()) || has_raw_status(&e, 404) {
                 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();
@@ -5444,178 +5383,6 @@ 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,7 +48,6 @@ 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};
@@ -193,14 +192,6 @@ 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(
@@ -290,12 +281,6 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo)
             // record may only add multipart-ness, never take it away.
             is_multipart = base_is_multipart || checksum_record_is_multipart;
 
-            for (key, value) in checksum_meta.iter() {
-                if key != AMZ_CHECKSUM_TYPE {
-                    meta.insert(key.clone(), value.clone());
-                }
-            }
-
             if !base_is_multipart
                 && checksum_meta
                     .get(AMZ_CHECKSUM_TYPE)
@@ -303,6 +288,26 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo)
             {
                 is_multipart = false;
             }
+
+            // The record keys each checksum by algorithm name ("CRC32"); the
+            // target only reads `x-amz-checksum-<algorithm>`. Inserting the bare
+            // name here made `PutObjectOptions::header()` send it as user
+            // metadata (`x-amz-meta-crc32`), so no replica ever carried the
+            // source checksum (rustfs/backlog#2340). The object-level record
+            // describes one PUT body: a multipart replica is rebuilt part by
+            // part, and its CreateMultipartUpload must not announce a checksum
+            // the parts do not carry, so the record is forwarded on the
+            // single-PUT route only (MinIO `getCRCMeta` parity).
+            if !is_multipart {
+                for (key, value) in checksum_meta.iter() {
+                    if key == AMZ_CHECKSUM_TYPE {
+                        continue;
+                    }
+                    if let Some(header) = rustfs_rio::ChecksumType::from_string(key).key() {
+                        meta.insert(header.to_string(), value.clone());
+                    }
+                }
+            }
         }
     }
 
@@ -1477,12 +1482,63 @@ mod tests {
                 ..Default::default()
             };
 
-            let (opts, _is_multipart) = replication_put_object_options("", &object_info).expect("build replication put options");
+            let (opts, is_multipart) = replication_put_object_options("", &object_info).expect("build replication put options");
 
+            assert!(!is_multipart, "{name}: a single-part checksum record must keep the single-PUT route");
+            let header = ty.key().expect("every forwarded algorithm has an x-amz-checksum header");
             assert_eq!(
-                opts.user_metadata.get(name),
+                opts.user_metadata.get(header),
                 Some(&checksum.encoded),
-                "replication must forward the {name} checksum into user_metadata identically to the classic algorithms"
+                "replication must forward the {name} checksum as the {header} header"
+            );
+            assert!(
+                !opts.user_metadata.contains_key(name),
+                "{name}: the bare algorithm name would leave as x-amz-meta user metadata"
+            );
+        }
+    }
+
+    /// The object-level record of a multipart upload (composite or full-object)
+    /// must not become a PutObject checksum header: the replica is rebuilt
+    /// through CreateMultipartUpload/UploadPart, and a checksum announced there
+    /// that the parts do not carry would be rejected by the target.
+    #[test]
+    fn replication_put_object_options_keeps_multipart_checksum_records_off_the_wire() {
+        let mut composite_type = rustfs_rio::ChecksumType::from_string("crc32");
+        composite_type
+            .merge(rustfs_rio::ChecksumType::MULTIPART)
+            .merge(rustfs_rio::ChecksumType::INCLUDES_MULTIPART);
+        let mut combined = Vec::new();
+        for part in [b"part-one".as_slice(), b"part-two".as_slice()] {
+            let part_checksum =
+                rustfs_rio::Checksum::new_from_data(rustfs_rio::ChecksumType::from_string("crc32"), part).expect("part checksum");
+            combined.extend_from_slice(part_checksum.raw.as_slice());
+        }
+        let composite = rustfs_rio::Checksum::new_from_data(composite_type, &combined)
+            .expect("composite checksum")
+            .to_bytes(&combined);
+
+        for (label, checksum, etag) in [
+            ("composite", composite, "0123456789abcdef0123456789abcdef-2"),
+            (
+                "full-object",
+                full_object_multipart_checksum_record(),
+                "0123456789abcdef0123456789abcdef-3",
+            ),
+        ] {
+            let object_info = ObjectInfo {
+                etag: Some(etag.to_string()),
+                checksum: Some(checksum),
+                ..Default::default()
+            };
+            let (opts, is_multipart) = replication_put_object_options("", &object_info).expect("build replication put options");
+            assert!(is_multipart, "{label}: a multipart object must keep the multipart route");
+            assert!(
+                opts.user_metadata
+                    .keys()
+                    .all(|key| !key.starts_with("x-amz-checksum-") && key != "CRC32"),
+                "{label}: no object-level checksum may reach the target's CreateMultipartUpload: {:?}",
+                opts.user_metadata
             );
         }
     }

crates/replication/src/lib.rs

@@ -65,10 +65,9 @@ pub use multipart::{
 };
 pub use object::{
     ObjectLockIntegrity, ReplicationSourceObject, ReplicationTargetObject, SsecPassthroughCapability, SsecPassthroughGate,
-    VersionIdentityCapability, content_matches_by_etag, is_replication_target_offline_error, object_lock_put_integrity,
-    replication_action_for_target, replication_etags_match, single_part_replica_etag_mismatch, ssec_passthrough_evidence_present,
-    ssec_passthrough_gate, target_is_newer_than_source_null_version, version_identity_capability_from_put,
-    version_identity_drifted,
+    content_matches_by_etag, is_replication_target_offline_error, object_lock_put_integrity, replication_action_for_target,
+    replication_etags_match, single_part_replica_etag_mismatch, ssec_passthrough_evidence_present, ssec_passthrough_gate,
+    target_is_newer_than_source_null_version, version_identity_drifted,
 };
 pub use operation::{
     MustReplicateOptions, ReplicationDeleteScheduleInput, ReplicationDeleteSource, ReplicationDeleteStateSource,

crates/replication/src/object.rs

@@ -234,62 +234,18 @@ fn comparable_metadata(metadata: Option<&HashMap<String, String>>) -> HashMap<St
 /// real (non-nil) version uuid, and drift means the target answered with
 /// anything else — including nothing at all.
 pub fn version_identity_drifted(source_version_id: &str, assigned_version_id: Option<&str>) -> bool {
-    version_identity_capability_from_put(source_version_id, assigned_version_id) == Some(VersionIdentityCapability::MintsOwn)
-}
-
-/// Whether a replication target adopts the source version id it is handed on
-/// PutObject / CompleteMultipartUpload, or mints its own.
-///
-/// A target that mints its own ids (AWS S3, Wasabi, Impossible Cloud) still
-/// stores the bytes, but every later version-addressed request from the
-/// source names an id the target never had. Its HEAD then answers 404 —
-/// indistinguishable from a replica that is really missing — so a heal, MRF
-/// retry or existing-object resync re-drive would PUT the object again and
-/// mint yet another target version (rustfs/backlog#2340). The replication
-/// worker learns the verdict from each PUT response (and replication-check's
-/// VersionFidelity phase) and, once `MintsOwn` is known, locates a replica by
-/// exact key and ETag before concluding that it is missing. The verdict cache
-/// is owned by the runtime's bucket target system; this crate owns only the
-/// vocabulary and the judgment.
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
-pub enum VersionIdentityCapability {
-    #[default]
-    Unknown,
-    Adopts,
-    MintsOwn,
-}
-
-impl VersionIdentityCapability {
-    /// True when a 404 from a version-addressed HEAD on this target cannot be
-    /// read as "replica missing": the source-side id was never the target's.
-    pub fn version_addressing_unreliable(self) -> bool {
-        self == VersionIdentityCapability::MintsOwn
-    }
-}
-
-/// Judge the identity contract from one replication write: `None` when no
-/// contract applies (the source addressed no real version — an empty or nil
-/// uuid travels as the literal "null", unversioned-source semantics),
-/// otherwise whether the target echoed the source id or answered with
-/// anything else — including nothing at all.
-pub fn version_identity_capability_from_put(
-    source_version_id: &str,
-    assigned_version_id: Option<&str>,
-) -> Option<VersionIdentityCapability> {
     if source_version_id.is_empty() {
-        return None;
+        return false;
     }
+    // A nil source uuid travels as the literal "null" (unversioned-source
+    // semantics); no identity contract applies to it.
     if uuid::Uuid::parse_str(source_version_id)
         .map(|uuid| uuid.is_nil())
         .unwrap_or(true)
     {
-        return None;
+        return false;
     }
-    Some(if assigned_version_id == Some(source_version_id) {
-        VersionIdentityCapability::Adopts
-    } else {
-        VersionIdentityCapability::MintsOwn
-    })
+    assigned_version_id != Some(source_version_id)
 }
 
 const REPLICATION_TARGET_OFFLINE_ERROR_MARKERS: &[&str] = &[
@@ -421,9 +377,9 @@ mod tests {
 
     use super::{
         ReplicationSourceObject, ReplicationTargetObject, SsecPassthroughCapability, SsecPassthroughGate,
-        VersionIdentityCapability, content_matches_by_etag, is_replication_target_offline_error, replication_action_for_target,
-        replication_etags_match, single_part_replica_etag_mismatch, ssec_passthrough_evidence_present, ssec_passthrough_gate,
-        target_is_newer_than_source_null_version, version_identity_capability_from_put, version_identity_drifted,
+        content_matches_by_etag, is_replication_target_offline_error, replication_action_for_target, replication_etags_match,
+        single_part_replica_etag_mismatch, ssec_passthrough_evidence_present, ssec_passthrough_gate,
+        target_is_newer_than_source_null_version, version_identity_drifted,
     };
     use crate::filemeta::{ReplicationAction, ReplicationType};
     use crate::http::AMZ_OBJECT_LOCK_MODE;
@@ -552,32 +508,6 @@ mod tests {
         }
     }
 
-    #[test]
-    fn version_identity_capability_is_judged_only_for_real_source_versions() {
-        let source = "8e4d2f4c-2d5c-4f1b-9d0a-9c8b7a6f5e4d";
-        assert_eq!(
-            version_identity_capability_from_put(source, Some(source)),
-            Some(VersionIdentityCapability::Adopts)
-        );
-        // Wasabi / AWS shape: a minted id, or no id at all, both mean the
-        // source-side id is not addressable on the target.
-        assert_eq!(
-            version_identity_capability_from_put(source, Some("001788697733811332140-fR6j6uXKV-")),
-            Some(VersionIdentityCapability::MintsOwn)
-        );
-        assert_eq!(
-            version_identity_capability_from_put(source, None),
-            Some(VersionIdentityCapability::MintsOwn)
-        );
-        // No contract for an unversioned source write.
-        assert_eq!(version_identity_capability_from_put("", Some("anything")), None);
-        assert_eq!(version_identity_capability_from_put("00000000-0000-0000-0000-000000000000", None), None);
-        assert_eq!(version_identity_capability_from_put("null", Some("null")), None);
-        assert!(VersionIdentityCapability::MintsOwn.version_addressing_unreliable());
-        assert!(!VersionIdentityCapability::Adopts.version_addressing_unreliable());
-        assert!(!VersionIdentityCapability::Unknown.version_addressing_unreliable());
-    }
-
     #[test]
     fn replication_target_offline_error_classifier_is_network_scoped() {
         assert!(is_replication_target_offline_error("put_object dispatch failure: connector error"));

docs/architecture/s3-compatibility-matrix.md

@@ -62,6 +62,7 @@ Object keys are stored as file-system paths under each drive (`{drive}/{bucket}/
 | Behavior | RustFS | AWS S3 | Why |
 |---|---|---|---|
 | Object key with a `.` or `..` path segment, or an empty segment (`//`), such as `a//b/./c/../d` | `400 InvalidArgument` (`check_object_args` in `crates/ecstore/src/bucket/utils.rs`, mirroring MinIO `IsValidObjectPrefix`) | Accepted as an opaque key | A `..` segment would resolve to a parent directory and `.`/`//` segments would alias other keys on disk; encoding them would change the MinIO-compatible on-disk format. |
+| Directory marker (key ending in `/`, with or without a body) in a versioned bucket | Stored as the null version: `PutObject`/`HeadObject` report version id `00000000-0000-0000-0000-000000000000`, `ListObjectVersions` reports `null`, and a later PUT of the same key overwrites in place (`put_opts` in `rustfs/src/storage/options.rs`, mirroring MinIO `putOpts`: "for directory objects skip creating new versions") | A real version id per PUT, with a version history | The marker only exists to make an empty prefix listable; keeping a history for it would leave hidden versions behind every prefix delete. Replication still copies the marker as its null version (`test_bucket_replication_replicates_directory_marker_in_versioned_bucket` in `crates/e2e_test/src/replication_extension_test.rs`). |
 
 ## Update Rule
 

docs/operations/replication-outbound-transport.md

@@ -6,7 +6,7 @@
 ## What a replication PUT carries by default
 
 - A plain signed body with an exact `Content-Length`. The SDK does not add a streaming trailer checksum, so the body is never wrapped in `aws-chunked` framing (rustfs#6853: a target that does not decode that framing stored the frames verbatim while RustFS recorded COMPLETED).
-- Any object-level checksum the source object was uploaded with, forwarded as its `x-amz-checksum-*` header.
+- For a single-part object, the checksum the source object was uploaded with, forwarded as its `x-amz-checksum-<algorithm>` header (the value the source verified on upload). A multipart replica is rebuilt through CreateMultipartUpload/UploadPart and carries no object-level checksum header. Managed-SSE objects forward none.
 - On a PUT that carries Object Lock parameters and no forwarded checksum: `Content-MD5` derived from the source ETag, or an SDK CRC32 checksum when the ETag is not the MD5 of the wire bytes (rustfs#7082).
 - The source ETag, mtime and version id on `x-rustfs-source-*` headers (with `x-minio-source-*` twins), and the Object Lock mode, retain-until date and legal hold of the source version when present.
 - After the PUT, the target's ETag is compared with the source ETag when both are plain single-part MD5s; a mismatch fails the replication instead of reporting a corrupted replica as COMPLETED.
@@ -17,6 +17,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; version-addressed convergence does not. See rustfs/backlog#2085 and `docs/operations/replication-check.md` (VersionFidelity). | `replication-check`, outbound target matrix, `MintOwnVersionIds` mode |
 | 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` |
 
@@ -24,7 +25,7 @@
 
 | Variable | Default | Meaning |
 | --- | --- | --- |
-| `RUSTFS_REPLICATION_STREAMING_CHECKSUMS` | unset (plain payloads) | `true` or `1` restores SDK trailer checksums (`RequestChecksumCalculation::WhenSupported`). Every streaming upload is then `aws-chunked` with an `x-amz-trailer`; use only when every target decodes that framing. |
+| `RUSTFS_REPLICATION_STREAMING_CHECKSUMS` | unset (plain payloads) | `true` or `1` restores SDK trailer checksums (`RequestChecksumCalculation::WhenSupported`). Every streaming upload is then `aws-chunked` with an `x-amz-trailer`, except a single-part PUT that forwards the source's `x-amz-checksum-*` header, which is sent plain so the target does not receive a second algorithm; use only when every target decodes that framing. |
 | `RUSTFS_REPLICATION_REPLICA_ETAG_VERIFY` | enabled | `false` or `0` disables the post-PUT ETag comparison for targets whose 32-hex ETags are legitimately not the content MD5. |
 
 Both knobs are read by the RustFS process that owns the replication target, at client build time; restart the server after changing them.

rustfs/src/admin/router.rs

@@ -18,7 +18,7 @@ use super::storage_api::bucket::replication::{self, BucketReplicationResyncStatu
 use super::storage_api::bucket::target::{BucketTarget, BucketTargetType, BucketTargets};
 use super::storage_api::bucket::target_sys::{
     BucketTargetSys, PutObjectOptions, RemoveObjectOptions, S3ClientError, SsecPassthroughCapability, TargetClient,
-    VersionIdentityCapability, append_version_id_query,
+    append_version_id_query,
 };
 use super::storage_api::bucket::versioning_sys::BucketVersioningSys;
 use super::storage_api::bucket::{AdminReplicationConfigExt as _, AdminVersioningConfigExt as _};
@@ -2100,18 +2100,6 @@ async fn check_replication_target(
         }
         _ => {}
     }
-    // Same for the identity verdict: once a target is known to mint its own
-    // version ids, the worker locates replicas by content identity instead of
-    // re-driving PUTs whenever a version-addressed HEAD answers 404.
-    match (result.phases.version_fidelity.status, result.phases.version_fidelity.code) {
-        ("OK", _) => {
-            BucketTargetSys::get().record_version_identity_capability(&target.arn, VersionIdentityCapability::Adopts);
-        }
-        ("FAILED", Some(REPLICATION_CHECK_CODE_VERSION_MISMATCH)) => {
-            BucketTargetSys::get().record_version_identity_capability(&target.arn, VersionIdentityCapability::MintsOwn);
-        }
-        _ => {}
-    }
 
     result
 }

rustfs/src/admin/storage_api.rs

@@ -211,7 +211,6 @@ pub(crate) mod bucket_target_sys {
     pub(crate) type RemoveObjectOptions = super::ecstore_bucket::bucket_target_sys::RemoveObjectOptions;
     pub(crate) type S3ClientError = super::ecstore_bucket::bucket_target_sys::S3ClientError;
     pub(crate) type SsecPassthroughCapability = super::ecstore_bucket::bucket_target_sys::SsecPassthroughCapability;
-    pub(crate) type VersionIdentityCapability = super::ecstore_bucket::bucket_target_sys::VersionIdentityCapability;
     pub(crate) type TargetClient = super::ecstore_bucket::bucket_target_sys::TargetClient;
 }