From 3b404e56c0a6eefda7d30ec133925a0abd9b8341 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=94=90=E5=B0=8F=E9=B8=AD?= Date: Mon, 7 Sep 2026 09:17:24 +0800 Subject: [PATCH] fix(replication): forward single-part object checksums as headers (#7313) --- .../src/replication_extension_test.rs | 87 +++++++++++++++++++ .../src/replication_target_matrix_test.rs | 44 +++++++++- .../ecstore/src/bucket/bucket_target_sys.rs | 83 ++++++++++++++++-- .../replication_target_boundary.rs | 83 ++++++++++++++++-- .../replication-outbound-transport.md | 5 +- 5 files changed, 282 insertions(+), 20 deletions(-) diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index f28f9c153..d6fc11e6a 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/crates/e2e_test/src/replication_extension_test.rs @@ -3988,6 +3988,93 @@ async fn test_bucket_replication_version_purge_of_non_inline_object_releases_sou 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(); diff --git a/crates/e2e_test/src/replication_target_matrix_test.rs b/crates/e2e_test/src/replication_target_matrix_test.rs index aa688f592..bace5a08e 100644 --- a/crates/e2e_test/src/replication_target_matrix_test.rs +++ b/crates/e2e_test/src/replication_target_matrix_test.rs @@ -42,7 +42,8 @@ use crate::replication_extension_test::{ 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) + } } } } @@ -614,6 +641,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(()) } diff --git a/crates/ecstore/src/bucket/bucket_target_sys.rs b/crates/ecstore/src/bucket/bucket_target_sys.rs index 9aa5afbd0..b0634c9ce 100644 --- a/crates/ecstore/src/bucket/bucket_target_sys.rs +++ b/crates/ecstore/src/bucket/bucket_target_sys.rs @@ -2199,7 +2199,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 +2227,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 +2654,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 +2835,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 +3241,14 @@ mod tests { } fn s3_client_for_endpoint_test(endpoint: String, http_client: Option) -> 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, + checksums: RequestChecksumCalculation, + ) -> S3Client { let credentials = SdkCredentials::builder() .access_key_id("test-access") .secret_access_key("test-secret") @@ -3193,7 +3262,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); } diff --git a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs index 93b6e1768..3f0f33369 100644 --- a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs @@ -290,12 +290,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 +297,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-`. 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 +1491,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 ); } } diff --git a/docs/operations/replication-outbound-transport.md b/docs/operations/replication-outbound-transport.md index f3149e21c..69996531e 100644 --- a/docs/operations/replication-outbound-transport.md +++ b/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-` 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-` 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.