fix(filemeta): keep data dir of a version awaiting purge replication

crates/e2e_test/src/replication_extension_test.rs

@@ -3908,89 +3908,82 @@ async fn test_bucket_replication_replicates_directory_marker_in_versioned_bucket
     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.
+/// Regression for rustfs/backlog#2340 (not Wasabi specific): permanently
+/// deleting a version whose payload lives in a data dir must leave the source
+/// clean once the purge replicates. Managed-SSE objects are never inlined and a
+/// plain object above the inline threshold takes the same layout. The version
+/// retained with a pending purge used to lose its data dir, so the purge state
+/// could never be applied (`VersionNotFound` on every retry) and the bucket
+/// stayed `BucketNotEmpty` while `ListObjectVersions` was already empty.
 #[tokio::test]
-async fn test_bucket_replication_forwards_single_part_object_checksums() -> TestResult {
+async fn test_bucket_replication_version_purge_of_non_inline_object_releases_source_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-checksum-src";
-    let target_bucket = "replication-checksum-dst";
+    let (source_env, target_env, source_bucket, target_bucket) = build_sse_replication_pair("purge-datadir", true, true).await?;
+    let target_arn = wait_for_remote_target_arn(&source_env, &source_bucket).await?;
+    put_bucket_replication_with_delete_statuses(&source_env, &source_bucket, &target_arn, "Enabled", Some("Enabled")).await?;
     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
+    let sse_key = "sse-object.bin";
+    let large_key = "large-object.bin";
+    let sse_put = source_client
         .put_object()
-        .bucket(source_bucket)
-        .key(crc32_key)
-        .body(ByteStream::from_static(body))
-        .checksum_algorithm(aws_sdk_s3::types::ChecksumAlgorithm::Crc32)
+        .bucket(&source_bucket)
+        .key(sse_key)
+        .body(ByteStream::from_static(b"encrypted source payload"))
+        .server_side_encryption(ServerSideEncryption::Aes256)
         .send()
         .await?;
-    let expected_crc32 = crc32_put.checksum_crc32().ok_or("source PUT omitted CRC32")?.to_string();
-    let sha256_put = source_client
+    let large_put = source_client
         .put_object()
-        .bucket(source_bucket)
-        .key(sha256_key)
-        .body(ByteStream::from_static(body))
-        .checksum_algorithm(aws_sdk_s3::types::ChecksumAlgorithm::Sha256)
+        .bucket(&source_bucket)
+        .key(large_key)
+        .body(ByteStream::from(vec![0x5a; 2 * 1024 * 1024]))
         .send()
         .await?;
-    let expected_sha256 = sha256_put.checksum_sha256().ok_or("source PUT omitted SHA256")?.to_string();
+    let purged = [
+        (sse_key, sse_put.version_id().ok_or("SSE PUT omitted version ID")?.to_string()),
+        (large_key, large_put.version_id().ok_or("large PUT omitted version ID")?.to_string()),
+    ];
+    assert_replication_converged(&source_client, &source_bucket, &target_client, &target_bucket).await?;
 
-    for key in [crc32_key, sha256_key] {
-        wait_for_source_replication_status(&source_client, source_bucket, key, "COMPLETED", false).await?;
+    for (key, version_id) in &purged {
+        source_client
+            .delete_object()
+            .bucket(&source_bucket)
+            .key(*key)
+            .version_id(version_id)
+            .send()
+            .await?;
     }
+    assert_replication_converged(&source_client, &source_bucket, &target_client, &target_bucket).await?;
+    let target_state = list_replication_state(&target_client, &target_bucket).await?;
+    assert!(target_state.is_empty(), "target retained an explicitly purged version: {target_state:?}");
 
-    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()
-    );
+    // The purge state is applied on the source asynchronously after the target
+    // acknowledges the delete; only then does the retained version go away and
+    // the bucket become deletable. A listing that is empty while DeleteBucket
+    // keeps answering BucketNotEmpty is exactly the regression.
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(60);
+    loop {
+        let listing = source_client.list_object_versions().bucket(&source_bucket).send().await?;
+        let listed = listing.versions().len() + listing.delete_markers().len();
+        match source_client.delete_bucket().bucket(&source_bucket).send().await {
+            Ok(_) => break,
+            Err(err) if err.code() == Some("BucketNotEmpty") => {
+                if tokio::time::Instant::now() >= deadline {
+                    return Err(format!(
+                        "source bucket stayed BucketNotEmpty after the version purge replicated; \
+                         ListObjectVersions shows {listed} entries"
+                    )
+                    .into());
+                }
+                sleep(Duration::from_millis(500)).await;
+            }
+            Err(err) => return Err(err.into()),
+        }
+    }
 
     Ok(())
 }

crates/e2e_test/src/replication_target_matrix_test.rs

@@ -42,8 +42,7 @@ use crate::replication_extension_test::{
 use aws_sdk_s3::Client;
 use aws_sdk_s3::primitives::{ByteStream, DateTime};
 use aws_sdk_s3::types::{
-    Checksum, ChecksumAlgorithm, CompletedMultipartUpload, CompletedPart, ObjectAttributes, ObjectLockLegalHoldStatus,
-    ObjectLockMode,
+    Checksum, CompletedMultipartUpload, CompletedPart, ObjectAttributes, ObjectLockLegalHoldStatus, ObjectLockMode,
 };
 use bytes::Bytes;
 use std::error::Error;
@@ -115,13 +114,10 @@ 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; 8] = [
+    const ALL: [ObjectShape; 7] = [
         ObjectShape::Empty,
         ObjectShape::Plain,
         ObjectShape::Retention,
@@ -129,7 +125,6 @@ impl ObjectShape {
         ObjectShape::Multipart,
         ObjectShape::LockedMultipart,
         ObjectShape::OdmPreservedMd5Multipart,
-        ObjectShape::Checksummed,
     ];
 
     fn key(self) -> &'static str {
@@ -141,16 +136,6 @@ 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,
         }
     }
 
@@ -213,18 +198,6 @@ 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)
-            }
         }
     }
 }
@@ -477,19 +450,6 @@ 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/bucket/bucket_target_sys.rs

@@ -2064,15 +2064,7 @@ impl TargetClient {
             }
         }
 
-        // 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
+        match builder
             .bucket(bucket)
             .key(object)
             .content_length(size)
@@ -2092,14 +2084,10 @@ impl TargetClient {
                 }
 
                 Result::<_, aws_smithy_types::error::operation::BuildError>::Ok(req)
-            });
-        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 {
+            })
+            .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
@@ -2519,21 +2507,13 @@ 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_endpoint_test_with_checksums("https://localhost:443".to_string(), Some(http_client), checksums);
+        let client = s3_client_for_test(443, Some(http_client));
         (
             TargetClient {
                 endpoint: "https://localhost:443".to_string(),
@@ -2700,47 +2680,6 @@ 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() {
@@ -3106,14 +3045,6 @@ 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")
@@ -3127,7 +3058,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(checksums);
+            .request_checksum_calculation(replication_request_checksum_calculation());
         if let Some(http_client) = http_client {
             config = config.http_client(http_client);
         }

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

@@ -281,6 +281,12 @@ 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)
@@ -288,26 +294,6 @@ 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());
-                    }
-                }
-            }
         }
     }
 
@@ -1482,63 +1468,12 @@ 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(header),
+                opts.user_metadata.get(name),
                 Some(&checksum.encoded),
-                "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
+                "replication must forward the {name} checksum into user_metadata identically to the classic algorithms"
             );
         }
     }

crates/filemeta/src/filemeta.rs

@@ -692,10 +692,15 @@ impl FileMeta {
                             }
                         }
 
-                        let old_dir = v.object.as_ref().map(|v| v.data_dir).unwrap_or_default();
+                        // The version stays on disk while the purge replicates
+                        // (status PENDING/FAILED); its data dir must stay with
+                        // it. Returning the dir here made the disk layer delete
+                        // it, which turned every non-inline retained version
+                        // into an unreadable zombie: the purge state could never
+                        // be applied and the bucket could never be deleted.
                         self.set_idx(i, v)?;
 
-                        return Ok(old_dir);
+                        return Ok(None);
                     }
                     found_index = Some(i);
                 }
@@ -2702,6 +2707,58 @@ mod test {
         );
     }
 
+    /// Regression for rustfs/backlog#2340: a version purge that still awaits
+    /// the replication target keeps the object version on disk with a pending
+    /// purge status. Its data dir must be retained with it; handing the dir
+    /// back here made the disk layer delete it, leaving every non-inline
+    /// retained version unreadable. The dir is released only once the purge
+    /// completes and the version itself goes away.
+    #[test]
+    fn delete_version_pending_version_purge_retains_object_data_dir() {
+        let version_id = Uuid::new_v4();
+        let data_dir = Uuid::new_v4();
+        let mut fm = FileMeta::new();
+        let mut fi = FileInfo::new("object", 2, 2);
+        fi.version_id = Some(version_id);
+        fi.data_dir = Some(data_dir);
+        fi.mod_time = Some(OffsetDateTime::now_utc());
+        fm.add_version(fi).unwrap();
+
+        let pending_purge = FileInfo {
+            name: "object".to_string(),
+            version_id: Some(version_id),
+            mark_deleted: true,
+            replication_state_internal: Some(ReplicationState {
+                version_purge_status_internal: Some("target=PENDING;".to_string()),
+                purge_targets: version_purge_statuses_map("target=PENDING;"),
+                ..Default::default()
+            }),
+            ..Default::default()
+        };
+        let freed = fm.delete_version(&pending_purge).unwrap();
+        assert_eq!(freed, None, "a pending purge must not release the retained version's data dir");
+        assert_eq!(fm.versions.len(), 1, "the version must stay until the purge replicates");
+        let retained = fm
+            .into_fileinfo("vol", "object", &version_id.to_string(), false, false, true)
+            .unwrap();
+        assert_eq!(retained.data_dir, Some(data_dir));
+        assert_eq!(retained.version_purge_status(), VersionPurgeStatusType::Pending);
+
+        let completed_purge = FileInfo {
+            name: "object".to_string(),
+            version_id: Some(version_id),
+            replication_state_internal: Some(ReplicationState {
+                version_purge_status_internal: Some("target=COMPLETE;".to_string()),
+                purge_targets: version_purge_statuses_map("target=COMPLETE;"),
+                ..Default::default()
+            }),
+            ..Default::default()
+        };
+        let freed = fm.delete_version(&completed_purge).unwrap();
+        assert_eq!(freed, Some(data_dir), "a completed purge removes the version and releases its data dir");
+        assert!(fm.versions.is_empty());
+    }
+
     #[test]
     fn delete_version_accepts_delete_only_marker_and_free_version_paths() {
         let marker_version_id = Uuid::new_v4();

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).
-- 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.
+- Any object-level checksum the source object was uploaded with, forwarded as its `x-amz-checksum-*` header.
 - 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,7 +17,6 @@
 | --- | --- | --- |
 | 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` |
 
@@ -25,7 +24,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`, 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_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_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.