From 8023cf3e26eef40233ac788f753b07d0a25d78de Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=94=90=E5=B0=8F=E9=B8=AD?= Date: Thu, 3 Sep 2026 14:57:05 +0800 Subject: [PATCH] test(e2e): add the outbound target matrix and #7082 postmortem (#7092) test(e2e): add the outbound target matrix and the replication checksum postmortem Defense work for rustfs#7082, the regression rustfs#6895 introduced while fixing rustfs#6853: a fix for one target class changed a client default for every target class and nothing in tree modeled the other classes. - docs/postmortems: timeline, root cause, why four defense layers missed it, and the SOP for changing any outbound client default; AGENTS.md and the adversarial compatibility lens point at it; the two env knobs from rustfs#6895 are documented in docs/operations. - fake_s3_target: reject_aws_chunked_uploads, require_checksum_for_object_lock (Content-MD5 always verified), create_bucket_with_object_lock with a GetObjectLockConfiguration handler, and a TransportSnapshot on every journal record. - replication_target_matrix_test: six object shapes against four target modes with an explicit expectation table; the two rustfs#7082 cells are pinned KnownFailing and fail with an XPASS message once the fix lands. Wired into e2e-repl-nightly, excluded from e2e-full. --- .../references/compatibility.md | 14 + .config/e2e-full-selection.txt | 4 +- .config/e2e-repl-nightly-selection.txt | 2 +- .config/e2e-smoke-selection.txt | 2 +- .config/nextest.toml | 7 +- .gitignore | 2 + AGENTS.md | 6 + crates/e2e_test/src/fake_s3_target/README.md | 4 +- crates/e2e_test/src/fake_s3_target/mod.rs | 381 ++++++++++++- crates/e2e_test/src/lib.rs | 6 + .../src/replication_extension_test.rs | 27 +- .../src/replication_target_matrix_test.rs | 507 ++++++++++++++++++ .../replication-outbound-transport.md | 33 ++ ...replication-checksum-default-regression.md | 63 +++ docs/testing/README.md | 1 + scripts/check_doc_paths.sh | 2 +- 16 files changed, 1036 insertions(+), 25 deletions(-) create mode 100644 crates/e2e_test/src/replication_target_matrix_test.rs create mode 100644 docs/operations/replication-outbound-transport.md create mode 100644 docs/postmortems/2026-09-03-replication-checksum-default-regression.md diff --git a/.agents/skills/adversarial-validation/references/compatibility.md b/.agents/skills/adversarial-validation/references/compatibility.md index 6b2b9b73b..6946ea647 100644 --- a/.agents/skills/adversarial-validation/references/compatibility.md +++ b/.agents/skills/adversarial-validation/references/compatibility.md @@ -22,3 +22,17 @@ fixtures and encrypted migration data. - Compatibility shims use `RUSTFS_COMPAT_TODO()`, have a removal condition, and default toward reading old data safely. + +## Outbound targets + +- A change to what the replication or migration client sends by default + (checksum policy, payload framing, headers, version-id addressing) is judged + against every target class, not the one it fixes. Name each target-side rule + the current default satisfies — checksum required with Object Lock + parameters, `aws-chunked` decoding, version-id adoption, ETag equals content + MD5 — and show which cell of + `crates/e2e_test/src/replication_target_matrix_test.rs` covers each. +- A test that asserts the fix ("no trailer header") is not evidence; the + matrix cell that asserts the target accepted and stored the object is. +- Every new environment escape hatch appears in + `docs/operations/replication-outbound-transport.md` in the same diff. diff --git a/.config/e2e-full-selection.txt b/.config/e2e-full-selection.txt index ac83c30cb..204565e83 100644 --- a/.config/e2e-full-selection.txt +++ b/.config/e2e-full-selection.txt @@ -1,2 +1,2 @@ -sha256-darwin=a45665d370be1d49301599aff8fb3e1ffe349f47fd2e6734f844a850a15d08a6 -sha256-linux=86e69337ad1440252a2ee20a12063c989ed12442d3b1ddf9e9233acf0f2ec089 +sha256-darwin=21f2a5484bb10bed1f11bd6397fe80f20d60a870e4ed5016a1f3acbdb4e6f735 +sha256-linux=29fd001e70d0114b59d65dd2a516c1f96a5fea1d6bb8fb315550a27f94d7b39b diff --git a/.config/e2e-repl-nightly-selection.txt b/.config/e2e-repl-nightly-selection.txt index 46bcb0862..b95fe0eaf 100644 --- a/.config/e2e-repl-nightly-selection.txt +++ b/.config/e2e-repl-nightly-selection.txt @@ -1 +1 @@ -sha256=8d5517f5f2fc32d561782dfccd51b7f746f5e25b2835e37e100c883f7f18777d +sha256=6b4c126c0b590e5768bcb9928e9bce025ddc2efeddcc8b6cda5ee4855730c469 diff --git a/.config/e2e-smoke-selection.txt b/.config/e2e-smoke-selection.txt index 2797c9973..cda5ece21 100644 --- a/.config/e2e-smoke-selection.txt +++ b/.config/e2e-smoke-selection.txt @@ -1 +1 @@ -sha256=56a9de1c0f000954f1c9ec691e8f24cc2ee387e73cf8b0e684c2baf70317d910 +sha256=d90fa941cb737fe6fcf39ca13d45440183e77a7e047880c025af7e6bfe1d0517 diff --git a/.config/nextest.toml b/.config/nextest.toml index 5ad679fd0..e85a10ebc 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -454,6 +454,10 @@ slow-timeout = { period = "60s", terminate-after = 2, grace-period = "10s" } # until it is explicitly promoted to the fast PR subset — no replication test # is ever silently left out of CI. # +# replication_target_matrix_test (the outbound target matrix: every object +# shape against every remote-target failure mode the fake target models) runs +# here in full; its expectation table pins known-red cells to open issues. +# # #[serial] does NOT serialize under nextest (process-per-test; see the file # header). These tests need no cross-test serialization: each spawns its own # server(s) on random ports with isolated temp dirs, so they are parallel-safe @@ -471,7 +475,7 @@ slow-timeout = { period = "60s", terminate-after = 2, grace-period = "10s" } [profile.e2e-repl-nightly] default-filter = """ package(e2e_test) - & test(/^replication_extension_test::/) + & (test(/^replication_extension_test::/) | test(/^replication_target_matrix_test::/)) & !test(/^replication_extension_test::(test_replication_check_succeeds_with_remote_target|test_replication_check_rejects_target_without_object_lock|test_set_remote_target_rejects_unversioned_source_bucket|test_replication_check_rejects_unversioned_source_bucket|test_replication_check_rejects_missing_replication_config|test_replication_check_rejects_invalid_bucket|test_set_remote_target_rejects_same_bucket_on_same_deployment|test_set_remote_target_rejects_unversioned_target_bucket|test_set_remote_target_update_requires_arn|test_set_remote_target_update_rejects_missing_target|test_set_remote_target_rejects_invalid_target_url|test_set_remote_target_rejects_self_signed_https_target_without_skip_tls_verify|test_set_remote_target_rejects_private_ca_https_target_without_ca_cert_pem|test_list_remote_targets_rejects_empty_bucket|test_list_remote_targets_rejects_invalid_bucket|test_remove_remote_target_rejects_missing_target|test_remove_remote_target_rejects_missing_arn|test_remove_remote_target_rejects_invalid_bucket|test_remove_remote_target_rejects_target_used_by_replication|test_delete_bucket_replication_removes_remote_target)$/) """ fail-fast = false @@ -560,6 +564,7 @@ default-filter = """ & !test(/^protocols::/) & !test(/^(admin_timeout_regression_test|cluster_concurrency_test|cluster_multidrive_pool_test|degraded_listing_availability_test|heal_erasure_disk_rebuild_test|namespace_lock_quorum_test|object_lambda_test|stale_multipart_cleanup_cluster_test)::/) & !test(/^replication_extension_test::/) + & !test(/^replication_target_matrix_test::/) & !test(/^on_demand_migration::(concurrency_test|fault_test|real_source_test)::/) """ fail-fast = false diff --git a/.gitignore b/.gitignore index c6d670f50..b8d2eecbf 100644 --- a/.gitignore +++ b/.gitignore @@ -55,6 +55,8 @@ docs/* !docs/architecture/** !docs/operations/ !docs/operations/** +!docs/postmortems/ +!docs/postmortems/** !docs/testing/ !docs/testing/** .codegraph/* diff --git a/AGENTS.md b/AGENTS.md index 757e0c969..a821ad2d2 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -162,6 +162,12 @@ Risk and review shape: S3-visible semantics. Cover all applicable lenses using exactly two independent reviewers when delegation is explicitly authorized. Split the lenses between them. Otherwise perform two fresh sequential passes. +- **Outbound client defaults:** what `TargetClient`, `PutObjectOptions`, or + the remote SDK configuration sends to every replication or migration target + is high risk for every target class even when the change fixes one. Follow + the SOP in `docs/postmortems/2026-09-03-replication-checksum-default-regression.md`: + run the outbound target matrix, document each new env knob in the same PR, + and list verified and unverified target classes in the PR Impact section. Available domain lenses are security, concurrency/durability, compatibility, and performance. Select `.agents/skills/adversarial-validation/SKILL.md` for an diff --git a/crates/e2e_test/src/fake_s3_target/README.md b/crates/e2e_test/src/fake_s3_target/README.md index 4546d9a66..533f97fb5 100644 --- a/crates/e2e_test/src/fake_s3_target/README.md +++ b/crates/e2e_test/src/fake_s3_target/README.md @@ -6,10 +6,12 @@ This module is the shared failure-injection boundary for replication end-to-end Supported data operations are HeadBucket, GetBucketVersioning, ListObjectsV2, PUT/GET/HEAD/DELETE Object, Get/Put/Delete ObjectTagging (tags live per version; Put replaces the whole set, Delete clears it), and create/upload/complete/abort multipart upload. `create_bucket` models general-purpose buckets in S3's shared global namespace; account-regional namespace buckets and their `-an` names are intentionally out of scope. Buckets created with `create_bucket` are versioned: PUT creates a version, DELETE without `versionId` creates a delete marker, and DELETE with `versionId` removes exactly that version. Internal source version IDs must be UUIDs and are stored canonically. Source mtime is honored only for source-replication PUT/DELETE requests; absent or invalid values use receipt time, matching RustFS, while multipart completion always uses receipt time. Replicated versions are ordered newest-first by source mtime so late older versions and delete markers do not become current. Equal mtimes prefer objects over delete markers, then canonical UUID order; RustFS's internal FileMeta signature tie-break is intentionally out of scope because it is not part of the target S3 protocol. Multipart part numbers follow S3's `1..=10000` range, and every completed part except the final part must be at least 5 MiB. +`create_bucket_with_object_lock(name)` creates a versioned bucket whose GetObjectLockConfiguration reports `Enabled`; every other bucket answers `ObjectLockConfigurationNotFoundError`, the code RustFS's replication-check classifies as "not enabled". Three switches model remote-target behaviors the fleet has shown, so the outbound target matrix (`crates/e2e_test/src/replication_target_matrix_test.rs`) can replicate every object shape against each: `assign_own_version_ids(true)` ignores the source version id and mints its own (AWS S3 / Wasabi); `reject_aws_chunked_uploads(true)` refuses any PutObject or UploadPart announcing `aws-chunked` framing (`Content-Encoding: aws-chunked`, an `x-amz-trailer`, or a `STREAMING-*` payload hash) with `InvalidRequest` before the body is read (SeaweedFS 3.97, rustfs#6853); `require_checksum_for_object_lock(true)` rejects a PutObject carrying any `x-amz-object-lock-*` header unless it also carries `Content-MD5`, an `x-amz-checksum-*` header, or `x-amz-sdk-checksum-algorithm` (AWS S3 / MinIO, rustfs#7082). Independently of that switch, a `Content-MD5` header is always verified against the body and a mismatch answers `BadDigest`. + `create_bucket_with_mode(name, BucketMode::Unversioned)` models a plain migration source: PUT overwrites in place, DELETE removes the key without a delete marker, GetBucketVersioning reports no status, and no `x-amz-version-id` is returned by PUT, GET, HEAD, tagging, or multipart completion. The only `versionId` such a bucket accepts is `null`; any other value is rejected with `InvalidArgument`. The mode is fixed at creation. ListObjectsV2 lists current versions only (a key whose newest version is a delete marker is hidden) in byte order and supports `prefix`, `delimiter`, `max-keys` (clamped to 1000), `start-after`, and `continuation-token`; common prefixes count toward `max-keys`, `IsTruncated` / `NextContinuationToken` / `KeyCount` follow S3, and continuation tokens are opaque. `encoding-type` and `fetch-owner` are accepted but ignored, and ListObjects (v1) is not implemented. GET and HEAD honor `Range` in the `bytes=first-last`, `bytes=first-`, and `bytes=-suffix` forms with a 206 status, exact `Content-Range`, and `Accept-Ranges: bytes`; unsatisfiable ranges answer 416 `InvalidRange` with `Content-Range: bytes */`. PUT and CreateMultipartUpload accept `Content-Type`, `Content-Encoding`, `Content-Disposition`, `Content-Language`, `Cache-Control`, `Expires`, and `x-amz-meta-*` (names stored lowercased), and HEAD/GET replay them verbatim together with `Last-Modified` and the ETag (hex MD5 for single PUTs, `-` for multipart objects). `put_seed_object` stores an object directly, bypassing the wire, the fault script, and the journal, so a source can be seeded without polluting the assertions a scenario later makes. -Fault actions cover HTTP 401/403/503 responses (`Status`), any 4xx/5xx status paired with the matching S3 error code (`ResponseStatus`), pre-dispatch delay, holding a fully computed successful response before its first byte (`Stall`), connection abort when a logical request-body threshold is reached, GetObject bodies cut off after N bytes while `Content-Length` announces the full size (`TruncateBodyAt`), streaming slow drain, and a deliberately wrong response ETag (including multipart-complete XML). `requests()` returns the ordered, credential-free request journal for assertions and `count_requests(operation, key)` counts entries for one exact key. Each record journals the `Range` and `User-Agent` request headers, the ListObjectsV2 `prefix` and `continuation-token` query values, and a `ProxyHeaderSnapshot` — the read-proxy anti-loop marker (`x-{rustfs,minio}-source-proxy-request`), the replication-check exemption header, and the client SSE-C header family (algorithm and key-MD5 values; for the key itself only its presence) — so proxy tests can pin the exact wire contract. +Fault actions cover HTTP 401/403/503 responses (`Status`), any 4xx/5xx status paired with the matching S3 error code (`ResponseStatus`), pre-dispatch delay, holding a fully computed successful response before its first byte (`Stall`), connection abort when a logical request-body threshold is reached, GetObject bodies cut off after N bytes while `Content-Length` announces the full size (`TruncateBodyAt`), streaming slow drain, and a deliberately wrong response ETag (including multipart-complete XML). `requests()` returns the ordered, credential-free request journal for assertions and `count_requests(operation, key)` counts entries for one exact key. Each record journals the `Range` and `User-Agent` request headers, the ListObjectsV2 `prefix` and `continuation-token` query values, a `TransportSnapshot` — whether the body was announced as `aws-chunked`, the verbatim `Content-MD5`, the sorted `x-amz-checksum-*` / `x-amz-sdk-checksum-algorithm` header names, and whether any `x-amz-object-lock-*` header was present — and a `ProxyHeaderSnapshot` — the read-proxy anti-loop marker (`x-{rustfs,minio}-source-proxy-request`), the replication-check exemption header, and the client SSE-C header family (algorithm and key-MD5 values; for the key itself only its presence) — so proxy tests can pin the exact wire contract. The listener is loopback-only. It admits at most 64 active connections and two concurrently buffered request bodies; authenticated multipart-complete XML collection and assembly take both body permits. Keep-alive is disabled, request-header reads are bounded to 30 seconds, a parsed request is bounded to 65 seconds, and the complete connection lifetime is bounded to 100 seconds. It retains at most 256 buckets, 4,096 journal entries, 4,096 scripted faults, 4,096 object versions, 256 multipart uploads, and 10,000 multipart parts. Retained identifiers are capped at 1 KiB, user metadata at 2 KiB, and content type and each standard object header at 1 KiB. By default a PUT or uploaded part is capped at 64 MiB and a completed multipart object and all stored object/part data are capped at 128 MiB; `FakeS3Target::start_with_options(FakeS3TargetOptions { max_object_bytes })` raises the object cap up to 256 MiB, and the total budget then becomes twice the object cap (never below 128 MiB). Body drain, body-permit waits, delay, stall, and slow-drain execution are bounded to 30 seconds; each slow-drain slice delay must be below that bound. diff --git a/crates/e2e_test/src/fake_s3_target/mod.rs b/crates/e2e_test/src/fake_s3_target/mod.rs index d9be301d4..7a04419cf 100644 --- a/crates/e2e_test/src/fake_s3_target/mod.rs +++ b/crates/e2e_test/src/fake_s3_target/mod.rs @@ -18,6 +18,7 @@ //! See [`README.md`](README.md) for the supported protocol and fault surface. use async_trait::async_trait; +use base64_simd::STANDARD as BASE64_STANDARD; use bytes::{Bytes, BytesMut}; use futures::StreamExt; use http::header::{CONTENT_LENGTH, ETAG, LAST_MODIFIED, RANGE, USER_AGENT}; @@ -33,10 +34,12 @@ use s3s::dto::{ AbortMultipartUploadInput, AbortMultipartUploadOutput, CommonPrefix, CompleteMultipartUploadInput, CompleteMultipartUploadOutput, CreateMultipartUploadInput, CreateMultipartUploadOutput, DeleteMarkerEntry, DeleteObjectInput, DeleteObjectOutput, DeleteObjectTaggingInput, DeleteObjectTaggingOutput, ETag, GetBucketVersioningInput, - GetBucketVersioningOutput, GetObjectInput, GetObjectOutput, GetObjectTaggingInput, GetObjectTaggingOutput, HeadBucketInput, - HeadBucketOutput, HeadObjectInput, HeadObjectOutput, ListObjectVersionsInput, ListObjectVersionsOutput, ListObjectsV2Input, - ListObjectsV2Output, Object, ObjectStorageClass, ObjectVersionId, PutObjectInput, PutObjectOutput, PutObjectTaggingInput, - PutObjectTaggingOutput, Range, StreamingBlob, Tag, TagSet, Timestamp, TimestampFormat, UploadPartInput, UploadPartOutput, + GetBucketVersioningOutput, GetObjectInput, GetObjectLockConfigurationInput, GetObjectLockConfigurationOutput, + GetObjectOutput, GetObjectTaggingInput, GetObjectTaggingOutput, HeadBucketInput, HeadBucketOutput, HeadObjectInput, + HeadObjectOutput, ListObjectVersionsInput, ListObjectVersionsOutput, ListObjectsV2Input, ListObjectsV2Output, Object, + ObjectLockConfiguration, ObjectLockEnabled, ObjectStorageClass, ObjectVersionId, PutObjectInput, PutObjectOutput, + PutObjectTaggingInput, PutObjectTaggingOutput, Range, StreamingBlob, Tag, TagSet, Timestamp, TimestampFormat, + UploadPartInput, UploadPartOutput, }; use s3s::service::{S3Service, S3ServiceBuilder}; use s3s::validation::{AwsNameValidation, NameValidation}; @@ -116,6 +119,7 @@ pub const FAKE_SECRET_KEY: &str = "fake-secret"; pub enum Operation { HeadBucket, GetBucketVersioning, + GetObjectLockConfiguration, PutObject, GetObject, HeadObject, @@ -350,6 +354,9 @@ pub struct RequestRecord { pub consumed_bytes: Option, pub replication_timestamps: ReplicationTimestampHeaders, pub proxy_headers: ProxyHeaderSnapshot, + /// Integrity and framing headers of the request, so outbound-transport + /// tests can pin what the sender actually put on the wire. + pub transport: TransportSnapshot, /// Verbatim `Range` request header, so range-forwarding tests can pin the /// exact wire syntax a migrating server sent to its source. pub range: Option, @@ -362,6 +369,50 @@ pub struct RequestRecord { pub fault: Option, } +/// Integrity and framing headers of an upload. A remote target's acceptance +/// rules key off exactly these (rustfs#6853: `aws-chunked` framing stored +/// verbatim; rustfs#7082: Object Lock PUTs need `Content-MD5` or +/// `x-amz-checksum-*`), so the journal records them for every request. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct TransportSnapshot { + /// The body was announced as `aws-chunked`: `Content-Encoding` names it, + /// an `x-amz-trailer` is declared, or the payload hash is `STREAMING-*`. + pub aws_chunked: bool, + /// Verbatim `Content-MD5` request header. + pub content_md5: Option, + /// Sorted names of every `x-amz-checksum-*` header, plus + /// `x-amz-sdk-checksum-algorithm` when present (the header a trailer + /// checksum announces itself with). + pub checksum_headers: Vec, + /// Any `x-amz-object-lock-*` header was present. + pub object_lock_params: bool, +} + +impl TransportSnapshot { + fn from_headers(headers: &HeaderMap) -> Self { + let content_encoding_chunked = headers.get_all("content-encoding").iter().any(|value| { + value + .to_str() + .is_ok_and(|value| value.split(',').any(|token| token.trim().eq_ignore_ascii_case("aws-chunked"))) + }); + let streaming_payload_hash = header_value(headers, &["x-amz-content-sha256"]) + .is_some_and(|value| value.to_ascii_uppercase().starts_with("STREAMING-")); + let mut checksum_headers: Vec = headers + .keys() + .map(|name| name.as_str().to_string()) + .filter(|name| name.starts_with("x-amz-checksum-") || name == "x-amz-sdk-checksum-algorithm") + .collect(); + checksum_headers.sort(); + checksum_headers.dedup(); + Self { + aws_chunked: content_encoding_chunked || streaming_payload_hash || headers.contains_key("x-amz-trailer"), + content_md5: header_value(headers, &["content-md5"]).map(bounded_journal_value), + checksum_headers, + object_lock_params: headers.keys().any(|name| name.as_str().starts_with("x-amz-object-lock-")), + } + } +} + /// Request headers journaled with every record, captured before the fault /// script is consulted. #[derive(Debug, Clone, Default, PartialEq, Eq)] @@ -369,6 +420,7 @@ struct JournaledHeaders { content_length: Option, replication_timestamps: ReplicationTimestampHeaders, proxy_headers: ProxyHeaderSnapshot, + transport: TransportSnapshot, range: Option, user_agent: Option, } @@ -382,6 +434,7 @@ impl JournaledHeaders { .and_then(|value| value.parse().ok()), replication_timestamps: ReplicationTimestampHeaders::from_headers(headers), proxy_headers: ProxyHeaderSnapshot::from_headers(headers), + transport: TransportSnapshot::from_headers(headers), range: header_value(headers, &[RANGE.as_str()]).map(bounded_journal_value), user_agent: header_value(headers, &[USER_AGENT.as_str()]).map(bounded_journal_value), } @@ -404,6 +457,14 @@ struct StoreState { /// transport headers instead of storing them (see /// [`REPLICATION_SSE_TRANSPORT_PREFIX`]). drop_unlisted_replication_headers: bool, + /// Models a target that cannot decode `aws-chunked` request bodies + /// (SeaweedFS 3.97, rustfs#6853): a PutObject or UploadPart announcing that + /// framing is refused with `InvalidRequest` before its body is read. + reject_aws_chunked_uploads: bool, + /// Models AWS S3 / MinIO / most compatible stores (rustfs#7082): a + /// PutObject carrying any `x-amz-object-lock-*` header must also carry + /// `Content-MD5` or an `x-amz-checksum-*` header. + require_checksum_for_object_lock: bool, limits: StoreLimits, buckets: HashMap, uploads: HashMap, @@ -414,6 +475,8 @@ struct StoreState { struct BucketState { versioned: bool, + /// Object Lock enabled at creation; reported by GetObjectLockConfiguration. + object_lock: bool, /// Versions per key, newest first (see `upsert_version`); an unversioned /// bucket holds exactly one version per key. objects: HashMap>, @@ -498,6 +561,7 @@ struct FakeBackend { struct FaultAccess { control: Arc>, body_limit: Arc, + store: Arc>, } #[derive(Debug, Clone)] @@ -580,6 +644,7 @@ impl FakeS3Target { builder.set_access(FaultAccess { control: Arc::clone(&control), body_limit, + store: Arc::clone(&backend.store), }); builder.build() }; @@ -681,6 +746,7 @@ impl FakeS3Target { bucket, BucketState { versioned, + object_lock: false, objects: HashMap::new(), }, ); @@ -688,6 +754,19 @@ impl FakeS3Target { } } + /// Pre-create a versioned bucket with Object Lock enabled, the shape a + /// RustFS source with Object Lock requires of its replication target + /// (`replication-check` probes GetObjectLockConfiguration for it). + pub fn create_bucket_with_object_lock(&self, bucket: impl Into) { + let bucket = bucket.into(); + self.create_bucket_with_mode(bucket.clone(), BucketMode::Versioned); + lock(&self.backend.store) + .buckets + .get_mut(&bucket) + .expect("bucket was just created") + .object_lock = true; + } + /// Store an object directly, bypassing the wire, the fault script, and /// the request journal. Seeds a migration source without polluting the /// journal that the scenario under test later asserts on. Returns the @@ -791,6 +870,22 @@ impl FakeS3Target { lock(&self.backend.store).drop_unlisted_replication_headers = enabled; } + /// Refuse every PutObject / UploadPart whose body is announced as + /// `aws-chunked` (SeaweedFS-like target, rustfs#6853). The refusal is an + /// `InvalidRequest` issued before the body is read, so a sender that + /// frames its uploads sees a hard failure instead of a corrupted replica. + pub fn reject_aws_chunked_uploads(&self, enabled: bool) { + lock(&self.backend.store).reject_aws_chunked_uploads = enabled; + } + + /// Require `Content-MD5` or an `x-amz-checksum-*` header on every + /// PutObject that carries Object Lock parameters (AWS S3 / MinIO rule, + /// rustfs#7082). `Content-MD5`, when present, is always verified against + /// the body regardless of this mode. + pub fn require_checksum_for_object_lock(&self, enabled: bool) { + lock(&self.backend.store).require_checksum_for_object_lock = enabled; + } + pub fn assign_own_multipart_version_ids(&self, enabled: bool) { lock(&self.backend.store).assign_own_multipart_version_ids = enabled; } @@ -942,10 +1037,20 @@ impl S3Access for FaultAccess { let parsed = parse_request(context.method(), context.uri()); let operation = operation_from_s3_name(context.s3_op().name()); let journaled = JournaledHeaders::from_headers(context.headers()); + let aws_chunked = journaled.transport.aws_chunked; let fault = record_request(&self.control, operation, context.method().clone(), parsed, journaled); if let Some(status) = fault.as_ref().and_then(|fault| scripted_status(&fault.action)) { return Err(scripted_status_error(status)); } + if aws_chunked + && matches!(operation, Operation::PutObject | Operation::UploadPart) + && lock(&self.store).reject_aws_chunked_uploads + { + return Err(s3s::s3_error!( + InvalidRequest, + "this target does not decode aws-chunked request bodies; send a plain signed payload with an exact Content-Length" + )); + } let prebody_permit = if operation == Operation::CompleteMultipartUpload { Some( timeout( @@ -973,6 +1078,7 @@ fn operation_from_s3_name(name: &str) -> Operation { match name { "HeadBucket" => Operation::HeadBucket, "GetBucketVersioning" => Operation::GetBucketVersioning, + "GetObjectLockConfiguration" => Operation::GetObjectLockConfiguration, "PutObject" => Operation::PutObject, "GetObject" => Operation::GetObject, "HeadObject" => Operation::HeadObject, @@ -1021,6 +1127,7 @@ fn record_request( consumed_bytes: None, replication_timestamps: headers.replication_timestamps, proxy_headers: headers.proxy_headers, + transport: headers.transport, range: headers.range, user_agent: headers.user_agent, prefix: parsed.prefix.map(bounded_journal_value), @@ -1846,6 +1953,34 @@ impl S3 for FakeBackend { )) } + async fn get_object_lock_configuration( + &self, + req: S3Request, + ) -> S3Result> { + let fault = request_fault(&req); + apply_non_body_fault(fault.as_ref(), &self.control).await?; + let object_lock = lock(&self.store) + .buckets + .get(&req.input.bucket) + .map(|bucket| bucket.object_lock) + .ok_or_else(|| s3s::s3_error!(NoSuchBucket, "bucket does not exist"))?; + if !object_lock { + return Err(s3s::s3_error!( + ObjectLockConfigurationNotFoundError, + "Object Lock configuration does not exist for this bucket" + )); + } + Ok(apply_response_fault( + S3Response::new(GetObjectLockConfigurationOutput { + object_lock_configuration: Some(ObjectLockConfiguration { + object_lock_enabled: Some(ObjectLockEnabled::from_static(ObjectLockEnabled::ENABLED)), + ..Default::default() + }), + }), + fault.as_ref(), + )) + } + /// Current versions only (a key whose newest version is a delete marker /// is hidden), keys in byte order, `delimiter` folding into common /// prefixes that count toward `max-keys`, and `start-after` / @@ -1996,17 +2131,37 @@ impl S3 for FakeBackend { .map_err(|_| s3s::s3_error!(ServiceUnavailable, "fake target body limiter closed"))?; let headers = req.headers; let input = req.input; - let (assign_own, drop_unlisted, limits, versioned) = { + let (assign_own, drop_unlisted, require_checksum, limits, versioned) = { let state = lock(&self.store); ( state.assign_own_version_ids, state.drop_unlisted_replication_headers, + state.require_checksum_for_object_lock, state.limits, bucket_versioned(&state, &input.bucket), ) }; + let transport = TransportSnapshot::from_headers(&headers); + if require_checksum + && transport.object_lock_params + && transport.content_md5.is_none() + && transport.checksum_headers.is_empty() + { + return Err(s3s::s3_error!( + InvalidRequest, + "Content-MD5 OR x-amz-checksum- HTTP header is required for Put Object requests with Object Lock parameters" + )); + } let body = collect_stream(input.body, input.content_length, fault.as_ref(), &self.control, limits.max_object_bytes).await?; + let mut body_permit = _body_permit; + if let Some(expected) = &transport.content_md5 { + let (digest, permit) = md5_digest(body.clone(), body_permit).await?; + body_permit = permit; + if BASE64_STANDARD.encode_to_string(digest) != expected.trim() { + return Err(s3s::s3_error!(BadDigest, "The Content-MD5 you specified did not match what we received.")); + } + } validate_stored_metadata(&input.content_type, &input.metadata)?; let standard_headers = StandardHeaders { cache_control: input.cache_control, @@ -2024,10 +2179,13 @@ impl S3 for FakeBackend { let e_tag = match source_etag(&headers)? { Some(value) => value, None => { - let (digest, _body_permit) = md5_digest(body.clone(), _body_permit).await?; + let (digest, permit) = md5_digest(body.clone(), body_permit).await?; + body_permit = permit; hex_simd::encode_to_string(digest, hex_simd::AsciiCase::Lower) } }; + // Held until the version is stored, like every other body-bearing op. + let _body_permit = body_permit; let version = ObjectVersion { version_id: version_id.clone(), body, @@ -2683,6 +2841,217 @@ mod tests { .await?) } + /// A client that sends no SDK-computed checksum, the shape RustFS's + /// replication client has had since rustfs#6895. + fn client_without_sdk_checksums(target: &FakeS3Target) -> Client { + let credentials = Credentials::new(FAKE_ACCESS_KEY, FAKE_SECRET_KEY, None, None, "fake-target"); + Client::from_conf( + aws_sdk_s3::Config::builder() + .credentials_provider(credentials) + .region(Region::new("us-east-1")) + .endpoint_url(target.endpoint()) + .force_path_style(true) + .behavior_version_latest() + .request_checksum_calculation(aws_sdk_s3::config::RequestChecksumCalculation::WhenRequired) + .retry_config(RetryConfig::standard().with_max_attempts(1)) + .http_client(SmithyHttpClientBuilder::new().build_http()) + .build(), + ) + } + + fn retain_until() -> aws_sdk_s3::primitives::DateTime { + aws_sdk_s3::primitives::DateTime::from_secs(4_102_444_800) + } + + #[tokio::test] + async fn object_lock_target_requires_a_checksum_on_locked_puts() -> Result<(), BoxError> { + use aws_sdk_s3::error::ProvideErrorMetadata; + use aws_sdk_s3::types::ObjectLockMode; + + let target = FakeS3Target::start().await?; + target.create_bucket_with_object_lock("locked"); + target.require_checksum_for_object_lock(true); + let client = client_without_sdk_checksums(&target); + let locked_put = |key: &'static str| { + client + .put_object() + .bucket("locked") + .key(key) + .body(ByteStream::from_static(b"locked payload")) + .object_lock_mode(ObjectLockMode::Governance) + .object_lock_retain_until_date(retain_until()) + }; + + // The rustfs#7082 shape: lock headers, no integrity header. + let err = locked_put("no-checksum") + .send() + .await + .expect_err("a locked PUT without Content-MD5 / x-amz-checksum-* must be rejected") + .into_service_error(); + assert_eq!(err.code(), Some("InvalidRequest")); + assert!( + err.message().unwrap_or_default().contains("Object Lock parameters"), + "unexpected message: {:?}", + err.message() + ); + assert!(!target.has_object("locked", "no-checksum")); + + // A wrong Content-MD5 is caught even though the header is present. + let err = locked_put("wrong-md5") + .customize() + .map_request(|mut request| { + request + .headers_mut() + .insert("content-md5", BASE64_STANDARD.encode_to_string(md5_bytes(b"other payload"))); + Ok::<_, std::convert::Infallible>(request) + }) + .send() + .await + .expect_err("a mismatching Content-MD5 must be rejected") + .into_service_error(); + assert_eq!(err.code(), Some("BadDigest")); + + // The correct Content-MD5 is accepted and the body is stored intact. + locked_put("good-md5") + .customize() + .map_request(|mut request| { + request + .headers_mut() + .insert("content-md5", BASE64_STANDARD.encode_to_string(md5_bytes(b"locked payload"))); + Ok::<_, std::convert::Infallible>(request) + }) + .send() + .await?; + assert_eq!( + get_bytes(&client, "locked", "good-md5", None).await?, + Bytes::from_static(b"locked payload") + ); + + // The rule is scoped to Object Lock parameters: an unlocked PUT + // without any checksum still lands. + client + .put_object() + .bucket("locked") + .key("unlocked") + .body(ByteStream::from_static(b"plain payload")) + .send() + .await?; + assert!(target.has_object("locked", "unlocked")); + + let journal = target.requests(); + let rejected = journal + .iter() + .find(|record| record.operation == Operation::PutObject && record.key.as_deref() == Some("no-checksum")) + .expect("the rejected PUT is journaled"); + assert!(rejected.transport.object_lock_params); + assert!(rejected.transport.checksum_headers.is_empty()); + assert_eq!(rejected.transport.content_md5, None); + assert!(!rejected.transport.aws_chunked); + let accepted = journal + .iter() + .find(|record| record.operation == Operation::PutObject && record.key.as_deref() == Some("good-md5")) + .expect("the accepted PUT is journaled"); + assert_eq!( + accepted.transport.content_md5.as_deref(), + Some(BASE64_STANDARD.encode_to_string(md5_bytes(b"locked payload")).as_str()) + ); + Ok(()) + } + + #[tokio::test] + async fn aws_chunked_rejecting_target_refuses_framed_uploads() -> Result<(), BoxError> { + use aws_sdk_s3::error::ProvideErrorMetadata; + + let target = FakeS3Target::start().await?; + target.create_bucket("plain"); + target.reject_aws_chunked_uploads(true); + let client = client_without_sdk_checksums(&target); + + // An upload announcing a trailer is refused before its body is read. + let err = client + .put_object() + .bucket("plain") + .key("framed") + .body(ByteStream::from_static(b"framed payload")) + .customize() + .map_request(|mut request| { + request.headers_mut().insert("x-amz-trailer", "x-amz-checksum-crc32"); + Ok::<_, std::convert::Infallible>(request) + }) + .send() + .await + .expect_err("an aws-chunked upload must be refused") + .into_service_error(); + assert_eq!(err.code(), Some("InvalidRequest")); + assert!( + err.message().unwrap_or_default().contains("aws-chunked"), + "unexpected message: {:?}", + err.message() + ); + assert!(!target.has_object("plain", "framed")); + + // A plain signed payload still lands. + client + .put_object() + .bucket("plain") + .key("unframed") + .body(ByteStream::from_static(b"plain payload")) + .send() + .await?; + assert!(target.has_object("plain", "unframed")); + + let journal = target.requests(); + let framed = journal + .iter() + .find(|record| record.key.as_deref() == Some("framed")) + .expect("the refused upload is journaled"); + assert!(framed.transport.aws_chunked); + let unframed = journal + .iter() + .find(|record| record.key.as_deref() == Some("unframed")) + .expect("the accepted upload is journaled"); + assert!(!unframed.transport.aws_chunked); + Ok(()) + } + + #[tokio::test] + async fn object_lock_configuration_reports_bucket_mode() -> Result<(), BoxError> { + use aws_sdk_s3::error::ProvideErrorMetadata; + + let target = FakeS3Target::start().await?; + target.create_bucket_with_object_lock("locked"); + target.create_bucket("plain"); + let client = client(&target); + + let locked = client.get_object_lock_configuration().bucket("locked").send().await?; + assert_eq!( + locked + .object_lock_configuration() + .and_then(|config| config.object_lock_enabled()) + .map(|state| state.as_str()), + Some("Enabled") + ); + + // The code RustFS's replication-check classifies as "not enabled". + let err = client + .get_object_lock_configuration() + .bucket("plain") + .send() + .await + .expect_err("a bucket without Object Lock has no configuration") + .into_service_error(); + assert_eq!(err.code(), Some("ObjectLockConfigurationNotFoundError")); + assert_eq!( + target + .requests() + .iter() + .filter(|record| record.operation == Operation::GetObjectLockConfiguration) + .count(), + 2 + ); + Ok(()) + } + #[tokio::test] async fn journals_replication_timestamp_headers() -> Result<(), BoxError> { let target = FakeS3Target::start().await?; diff --git a/crates/e2e_test/src/lib.rs b/crates/e2e_test/src/lib.rs index b364c44a6..e68ee426a 100644 --- a/crates/e2e_test/src/lib.rs +++ b/crates/e2e_test/src/lib.rs @@ -80,6 +80,12 @@ mod upgrade_compatibility_test; #[cfg(test)] mod replication_lww_receiver_test; +// Outbound target matrix: every object shape against every remote-target +// failure mode the fake target models (SOP: +// docs/postmortems/2026-09-03-replication-checksum-default-regression.md). +#[cfg(test)] +mod replication_target_matrix_test; + // Data usage regression tests #[cfg(test)] mod data_usage_test; diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index 01f462c84..73f43a05a 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/crates/e2e_test/src/replication_extension_test.rs @@ -86,7 +86,7 @@ type BacklogMetricPoints = Arc Result<(String, String, String), Ok((access_key, secret_key, session_token)) } -struct ReplicationTargetOptions<'a> { - endpoint: &'a str, - access_key: &'a str, - secret_key: &'a str, - target_bucket: &'a str, - secure: bool, - skip_tls_verify: bool, - ca_cert_pem: Option<&'a str>, +pub(crate) struct ReplicationTargetOptions<'a> { + pub(crate) endpoint: &'a str, + pub(crate) access_key: &'a str, + pub(crate) secret_key: &'a str, + pub(crate) target_bucket: &'a str, + pub(crate) secure: bool, + pub(crate) skip_tls_verify: bool, + pub(crate) ca_cert_pem: Option<&'a str>, } async fn set_replication_target( @@ -434,7 +434,7 @@ async fn set_replication_target( .await } -async fn set_replication_target_with_options( +pub(crate) async fn set_replication_target_with_options( source_env: &RustFSTestEnvironment, source_bucket: &str, options: ReplicationTargetOptions<'_>, @@ -504,7 +504,7 @@ async fn send_set_replication_target_request( .await } -async fn put_bucket_replication( +pub(crate) async fn put_bucket_replication( env: &RustFSTestEnvironment, bucket: &str, target_arn: &str, @@ -643,7 +643,10 @@ async fn get_bucket_replication( signed_request(http::Method::GET, &url, &env.access_key, &env.secret_key, None, None).await } -async fn enable_bucket_versioning(env: &RustFSTestEnvironment, bucket: &str) -> Result<(), Box> { +pub(crate) async fn enable_bucket_versioning( + env: &RustFSTestEnvironment, + bucket: &str, +) -> Result<(), Box> { set_bucket_versioning(env, bucket, BucketVersioningStatus::Enabled).await } diff --git a/crates/e2e_test/src/replication_target_matrix_test.rs b/crates/e2e_test/src/replication_target_matrix_test.rs new file mode 100644 index 000000000..521e0c323 --- /dev/null +++ b/crates/e2e_test/src/replication_target_matrix_test.rs @@ -0,0 +1,507 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Outbound target matrix: every object shape RustFS replicates, against +//! every remote-target failure mode the fake target models. +//! +//! The matrix exists because a fix for one target class shipped a regression +//! for another (rustfs#6895 fixed rustfs#6853 and caused rustfs#7082; see +//! `docs/postmortems/2026-09-03-replication-checksum-default-regression.md`). +//! Each row is one target mode with its own RustFS source and fake target; +//! each cell is one object shape. [`expectation`] is the single place that +//! says what a cell must do today: +//! +//! - `Completed` cells must replicate and the target must hold the source +//! bytes; the journal must also show the wire shape the cell relies on. +//! - `KnownFailing` cells pin an open issue. They must fail for the recorded +//! reason, and the moment they start passing the test fails with an XPASS +//! message so the expectation is flipped in the same PR as the fix. +//! +//! Adding a target behavior the fleet has shown: add the mode to the fake +//! target, add a row here, and record any cell that is red before the fix. + +use crate::common::{RustFSTestEnvironment, init_logging, replication_fast_env}; +use crate::fake_s3_target::{FAKE_ACCESS_KEY, FAKE_SECRET_KEY}; +use crate::fake_s3_target::{FakeS3Target, Operation as FakeTargetOperation, RequestRecord}; +use crate::on_demand_migration::common::fake_source_client; +use crate::replication_extension_test::{ + 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::{CompletedMultipartUpload, CompletedPart, ObjectLockLegalHoldStatus, ObjectLockMode}; +use bytes::Bytes; +use std::error::Error; +use std::time::{SystemTime, UNIX_EPOCH}; +use tokio::time::{Duration, sleep, timeout}; + +type TestResult = Result<(), Box>; + +/// A remote-target behavior the fleet has shown, as the fake target models it. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum TargetMode { + /// RustFS / MinIO-like target: adopts source version ids, decodes any + /// framing, enforces no checksum rule. + Baseline, + /// SeaweedFS 3.97 (rustfs#6853): refuses `aws-chunked` bodies. A sender + /// that frames its uploads gets a hard failure here instead of a + /// silently corrupted replica. + RejectAwsChunked, + /// AWS S3 / MinIO / Impossible Cloud (rustfs#7082): a PutObject with + /// Object Lock parameters must carry `Content-MD5` or `x-amz-checksum-*`. + RequireChecksumWithObjectLock, + /// AWS S3 / Wasabi / Impossible Cloud: mints its own version ids + /// (rustfs/backlog#2085). Data must still land. + MintOwnVersionIds, +} + +impl TargetMode { + const ALL: [TargetMode; 4] = [ + TargetMode::Baseline, + TargetMode::RejectAwsChunked, + TargetMode::RequireChecksumWithObjectLock, + TargetMode::MintOwnVersionIds, + ]; + + fn apply(self, target: &FakeS3Target) { + match self { + TargetMode::Baseline => {} + TargetMode::RejectAwsChunked => target.reject_aws_chunked_uploads(true), + TargetMode::RequireChecksumWithObjectLock => target.require_checksum_for_object_lock(true), + TargetMode::MintOwnVersionIds => target.assign_own_version_ids(true), + } + } + + fn slug(self) -> &'static str { + match self { + TargetMode::Baseline => "baseline", + TargetMode::RejectAwsChunked => "reject-aws-chunked", + TargetMode::RequireChecksumWithObjectLock => "require-checksum-object-lock", + TargetMode::MintOwnVersionIds => "mint-own-version-ids", + } + } +} + +/// An object shape the replication transport treats differently. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum ObjectShape { + /// The exact rustfs#7082 reproduction: a zero-byte object. + Empty, + /// Small single-part object with no Object Lock parameters. + Plain, + /// Single-part object with a GOVERNANCE retention period. + Retention, + /// Single-part object with legal hold ON. + LegalHold, + /// Two-part multipart upload, no Object Lock parameters. + Multipart, + /// Two-part multipart upload with a GOVERNANCE retention period; the + /// lock headers travel on CreateMultipartUpload, which has no body. + LockedMultipart, +} + +impl ObjectShape { + const ALL: [ObjectShape; 6] = [ + ObjectShape::Empty, + ObjectShape::Plain, + ObjectShape::Retention, + ObjectShape::LegalHold, + ObjectShape::Multipart, + ObjectShape::LockedMultipart, + ]; + + fn key(self) -> &'static str { + match self { + ObjectShape::Empty => "matrix/empty.bin", + ObjectShape::Plain => "matrix/plain.bin", + ObjectShape::Retention => "matrix/retention.bin", + ObjectShape::LegalHold => "matrix/legal-hold.bin", + ObjectShape::Multipart => "matrix/multipart.bin", + ObjectShape::LockedMultipart => "matrix/locked-multipart.bin", + } + } + + fn carries_object_lock_params(self) -> bool { + matches!(self, ObjectShape::Retention | ObjectShape::LegalHold | ObjectShape::LockedMultipart) + } + + /// Upload the shape to the source and return the bytes the target must + /// end up holding. + async fn put(self, client: &Client, bucket: &str) -> Result> { + let key = self.key(); + match self { + ObjectShape::Empty => { + client + .put_object() + .bucket(bucket) + .key(key) + .body(ByteStream::from_static(b"")) + .send() + .await?; + Ok(Bytes::new()) + } + ObjectShape::Plain => { + let body = payload(64 * 1024, 0x11); + client + .put_object() + .bucket(bucket) + .key(key) + .body(ByteStream::from(body.clone())) + .send() + .await?; + Ok(body) + } + ObjectShape::Retention => { + let body = payload(48 * 1024, 0x22); + client + .put_object() + .bucket(bucket) + .key(key) + .body(ByteStream::from(body.clone())) + .object_lock_mode(ObjectLockMode::Governance) + .object_lock_retain_until_date(retain_until()) + .send() + .await?; + Ok(body) + } + ObjectShape::LegalHold => { + let body = payload(32 * 1024, 0x33); + client + .put_object() + .bucket(bucket) + .key(key) + .body(ByteStream::from(body.clone())) + .object_lock_legal_hold_status(ObjectLockLegalHoldStatus::On) + .send() + .await?; + Ok(body) + } + ObjectShape::Multipart => multipart_put(client, bucket, key, 0x44, false).await, + ObjectShape::LockedMultipart => multipart_put(client, bucket, key, 0x55, true).await, + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum Expectation { + /// Replicates COMPLETED and the target holds the source bytes. + Completed, + /// Replicates FAILED today for a recorded reason; pinned to an open issue. + KnownFailing(&'static str), +} + +/// The single source of truth for what every cell must do today. A fix that +/// turns a `KnownFailing` cell green must flip it here in the same PR; the +/// test refuses an unexpected pass so the table cannot go stale silently. +fn expectation(mode: TargetMode, shape: ObjectShape) -> Expectation { + match (mode, shape) { + // rustfs#7082: the replication PUT carries the lock headers but no + // Content-MD5 / x-amz-checksum-* since rustfs#6895 switched the SDK + // to plain payloads; AWS-compatible Object Lock targets reject it. + (TargetMode::RequireChecksumWithObjectLock, ObjectShape::Retention | ObjectShape::LegalHold) => { + Expectation::KnownFailing("rustfs#7082") + } + _ => Expectation::Completed, + } +} + +#[tokio::test] +async fn matrix_baseline_target() -> TestResult { + run_row(TargetMode::Baseline).await +} + +#[tokio::test] +async fn matrix_reject_aws_chunked_target() -> TestResult { + run_row(TargetMode::RejectAwsChunked).await +} + +#[tokio::test] +async fn matrix_require_checksum_with_object_lock_target() -> TestResult { + run_row(TargetMode::RequireChecksumWithObjectLock).await +} + +#[tokio::test] +async fn matrix_mint_own_version_ids_target() -> TestResult { + run_row(TargetMode::MintOwnVersionIds).await +} + +/// The expectation table must name every mode and shape exactly once, so a +/// new row or column cannot be added without deciding what it does. +#[test] +fn expectation_table_covers_every_cell() { + for mode in TargetMode::ALL { + for shape in ObjectShape::ALL { + let _ = expectation(mode, shape); + } + } + let known_failing: Vec<_> = TargetMode::ALL + .iter() + .flat_map(|mode| ObjectShape::ALL.iter().map(move |shape| (*mode, *shape))) + .filter(|(mode, shape)| matches!(expectation(*mode, *shape), Expectation::KnownFailing(_))) + .collect(); + assert_eq!( + known_failing, + vec![ + (TargetMode::RequireChecksumWithObjectLock, ObjectShape::Retention), + (TargetMode::RequireChecksumWithObjectLock, ObjectShape::LegalHold), + ], + "every known-red cell is listed here on purpose; update this list together with the expectation table" + ); +} + +async fn run_row(mode: TargetMode) -> TestResult { + init_logging(); + + let target = FakeS3Target::start().await?; + let target_bucket = format!("matrix-{}-dst", mode.slug()); + target.create_bucket_with_object_lock(target_bucket.clone()); + mode.apply(&target); + + let mut source_env = RustFSTestEnvironment::new().await?; + 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", "")]); + source_env.start_rustfs_server_with_env(vec![], &env_vars).await?; + + let source_bucket = format!("matrix-{}-src", mode.slug()); + 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?; + + let target_client = fake_source_client(&target); + let mut failures = Vec::new(); + for shape in ObjectShape::ALL { + let cell = format!("{}/{:?}", mode.slug(), shape); + let expected_body = shape.put(&source_client, &source_bucket).await?; + let status = wait_for_terminal_replication_status(&source_client, &source_bucket, shape.key()).await?; + let journal = target.requests(); + let outcome = match expectation(mode, shape) { + Expectation::Completed => { + check_completed_cell(&cell, &status, &target_client, &target_bucket, shape, &expected_body, &journal).await + } + Expectation::KnownFailing(issue) => check_known_failing_cell(&cell, issue, &status, shape, &journal), + }; + if let Err(err) = outcome { + failures.push(format!("{cell}: {err}")); + } + } + + target.shutdown().await; + if failures.is_empty() { + Ok(()) + } else { + Err(format!( + "{} matrix cell(s) violated their expectation:\n {}", + failures.len(), + failures.join("\n ") + ) + .into()) + } +} + +async fn check_completed_cell( + cell: &str, + status: &str, + target_client: &Client, + target_bucket: &str, + shape: ObjectShape, + expected_body: &Bytes, + journal: &[RequestRecord], +) -> TestResult { + if status != "COMPLETED" { + return Err(format!("expected COMPLETED, source reports {status}").into()); + } + let stored = target_client + .get_object() + .bucket(target_bucket) + .key(shape.key()) + .send() + .await + .map_err(|err| format!("target GET failed after COMPLETED: {err}"))? + .body + .collect() + .await? + .into_bytes(); + if stored != *expected_body { + return Err(format!( + "target holds {} bytes that differ from the {} source bytes (COMPLETED over a corrupted replica)", + stored.len(), + expected_body.len() + ) + .into()); + } + // The wire shape the cell relies on: plain signed payloads (rustfs#6853) + // for every upload of this key, and the lock headers present exactly when + // the shape carries them. + let uploads: Vec<&RequestRecord> = journal + .iter() + .filter(|record| { + record.key.as_deref() == Some(shape.key()) + && matches!( + record.operation, + FakeTargetOperation::PutObject | FakeTargetOperation::UploadPart | FakeTargetOperation::CreateMultipartUpload + ) + }) + .collect(); + if uploads.is_empty() { + return Err("no upload reached the target although the source reports COMPLETED".into()); + } + if let Some(framed) = uploads.iter().find(|record| record.transport.aws_chunked) { + return Err(format!("{cell}: an upload went out aws-chunked (rustfs#6853 framing): {framed:?}").into()); + } + let lock_headers_seen = uploads.iter().any(|record| record.transport.object_lock_params); + if lock_headers_seen != shape.carries_object_lock_params() { + return Err(format!( + "object lock headers on the wire: {lock_headers_seen}, shape carries them: {}", + shape.carries_object_lock_params() + ) + .into()); + } + Ok(()) +} + +fn check_known_failing_cell(cell: &str, issue: &str, status: &str, shape: ObjectShape, journal: &[RequestRecord]) -> TestResult { + if status == "COMPLETED" { + return Err(format!( + "XPASS: {cell} reached COMPLETED but the expectation table pins it to {issue}; \ + the fix landed, so flip this cell to Expectation::Completed in the same PR" + ) + .into()); + } + if status != "FAILED" { + return Err(format!("expected FAILED ({issue}), source reports {status}").into()); + } + // Fail for the recorded reason, not by accident: the PUT carried the lock + // headers and no integrity header at all. + let rejected = journal.iter().any(|record| { + record.operation == FakeTargetOperation::PutObject + && record.key.as_deref() == Some(shape.key()) + && record.transport.object_lock_params + && record.transport.content_md5.is_none() + && record.transport.checksum_headers.is_empty() + }); + if !rejected { + return Err(format!( + "FAILED, but not for the {issue} reason (a locked PUT without Content-MD5 / x-amz-checksum-*); journal: {journal:?}" + ) + .into()); + } + Ok(()) +} + +/// First terminal replication status (`COMPLETED` or `FAILED`) the source +/// reports for the key. +async fn wait_for_terminal_replication_status( + client: &Client, + bucket: &str, + key: &str, +) -> Result> { + let wait = async { + loop { + let head = client.head_object().bucket(bucket).key(key).send().await?; + match head.replication_status().map(|status| status.as_str().to_string()) { + Some(status) if status == "COMPLETED" || status == "FAILED" => return Ok(status), + _ => sleep(Duration::from_millis(200)).await, + } + } + }; + match timeout(Duration::from_secs(90), wait).await { + Ok(result) => result, + Err(_) => Err(format!("{key} reached no terminal replication status within 90 seconds").into()), + } +} + +async fn multipart_put( + client: &Client, + bucket: &str, + key: &str, + fill: u8, + locked: bool, +) -> Result> { + let part_one = payload(5 * 1024 * 1024, fill); + let part_two = payload(256 * 1024, fill.wrapping_add(1)); + let mut create = client.create_multipart_upload().bucket(bucket).key(key); + if locked { + create = create + .object_lock_mode(ObjectLockMode::Governance) + .object_lock_retain_until_date(retain_until()); + } + let upload_id = create + .send() + .await? + .upload_id() + .ok_or("CreateMultipartUpload returned no upload id")? + .to_string(); + let mut completed = Vec::new(); + for (number, part) in [(1, &part_one), (2, &part_two)] { + let etag = client + .upload_part() + .bucket(bucket) + .key(key) + .upload_id(&upload_id) + .part_number(number) + .body(ByteStream::from(part.clone())) + .send() + .await? + .e_tag() + .ok_or("UploadPart returned no ETag")? + .to_string(); + completed.push(CompletedPart::builder().part_number(number).e_tag(etag).build()); + } + client + .complete_multipart_upload() + .bucket(bucket) + .key(key) + .upload_id(&upload_id) + .multipart_upload(CompletedMultipartUpload::builder().set_parts(Some(completed)).build()) + .send() + .await?; + let mut body = Vec::with_capacity(part_one.len() + part_two.len()); + body.extend_from_slice(&part_one); + body.extend_from_slice(&part_two); + Ok(Bytes::from(body)) +} + +fn payload(len: usize, fill: u8) -> Bytes { + Bytes::from((0..len).map(|i| fill.wrapping_add((i % 251) as u8)).collect::>()) +} + +fn retain_until() -> DateTime { + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("clock after epoch") + .as_secs(); + DateTime::from_secs(now as i64 + 86_400) +} diff --git a/docs/operations/replication-outbound-transport.md b/docs/operations/replication-outbound-transport.md new file mode 100644 index 000000000..4d0c9ec36 --- /dev/null +++ b/docs/operations/replication-outbound-transport.md @@ -0,0 +1,33 @@ +# Replication outbound transport + +**Use this when:** a bucket-replication or site-replication target rejects, corrupts, or silently transforms uploads from RustFS, or you need to know which integrity headers RustFS sends to a remote target and how to change them. +**Source of truth:** `crates/ecstore/src/bucket/remote_s3_client.rs` (`replication_request_checksum_calculation`), `crates/ecstore/src/bucket/bucket_target_sys.rs` (`TargetClient::put_object`, `PutObjectOptions::header`), `crates/ecstore/src/bucket/replication/replication_resyncer.rs` (`verify_single_part_replica`). + +## 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. +- 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. + +## Target classes and their known requirements + +| Target behavior | Effect on RustFS replication | Detected by | +| --- | --- | --- | +| 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) | Objects with a retention period or legal hold fail until rustfs#7082 lands. Set `RUSTFS_REPLICATION_STREAMING_CHECKSUMS=true` as a workaround when the target also decodes `aws-chunked`. | Outbound target matrix, `RequireChecksumWithObjectLock` mode | +| 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` | + +## Environment knobs + +| 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_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. + +## Before changing any of this + +Follow the SOP in `docs/postmortems/2026-09-03-replication-checksum-default-regression.md`: inventory the target-side rules the current default satisfies, run the outbound target matrix, and document any new knob here in the same PR. diff --git a/docs/postmortems/2026-09-03-replication-checksum-default-regression.md b/docs/postmortems/2026-09-03-replication-checksum-default-regression.md new file mode 100644 index 000000000..17e4fcd65 --- /dev/null +++ b/docs/postmortems/2026-09-03-replication-checksum-default-regression.md @@ -0,0 +1,63 @@ +# Postmortem: replication checksum default flipped, Object Lock targets rejected every PUT (rustfs#7082) + +**Use this when:** you change what the replication or migration client sends to a remote target by default, you are reviewing such a change, or you are triaging a report where a fix for one target class broke another. +**Source of truth:** `crates/ecstore/src/bucket/remote_s3_client.rs` (`replication_request_checksum_calculation`), `crates/e2e_test/src/replication_target_matrix_test.rs` (the outbound target matrix), `crates/e2e_test/src/fake_s3_target/mod.rs` (target failure modes). + +## Summary + +rustfs/rustfs#6895 fixed rustfs/rustfs#6853 (a target that stores `aws-chunked` framing verbatim) by switching the outbound SDK checksum policy from `WhenSupported` to `WhenRequired` for every replication target. The SDK-computed CRC32 had been doing a second job nobody had written down: it satisfied the S3 rule that a PutObject carrying `x-amz-object-lock-*` headers must include `Content-MD5` or an `x-amz-checksum-*` header. With the checksum gone, every replication PUT of an object with a retention period or legal hold was rejected by AWS-compatible Object Lock targets. The regression shipped in `1.0.0-rc.5` and was reported by a customer two days later. + +## Timeline (UTC) + +| When | Event | +| --- | --- | +| 2026-08-29 14:49 | rustfs#6853 filed from the round-2 real-VM lab (SeaweedFS stored `aws-chunked` frames). | +| 2026-08-30 14:24 | PR rustfs#6895 opened. | +| 2026-08-30 17:43 | PR merged. Zero GitHub reviews; adversarial validation ran in-session. | +| 2026-08-31 | `1.0.0-rc.5-preview.1` tagged. | +| 2026-09-01 17:35 | `1.0.0-rc.5` tagged. | +| 2026-09-02 20:29 | rustfs#7082 filed by a customer replicating to an Object Lock bucket on Impossible Cloud. | + +Three days from an own-lab finding to a public release candidate, with no Object Lock target exercised in between. + +## Root cause + +`RequestChecksumCalculation::WhenRequired` makes the SDK add a checksum only to operations whose model marks one as required. PutObject is not such an operation, so the client now sends no integrity header at all. AWS S3, MinIO and most compatible stores enforce a server-side rule the SDK model does not know about: a PutObject with Object Lock parameters needs `Content-MD5` or `x-amz-checksum-*`. RustFS itself does not enforce that rule (the matching error constant in `crates/ecstore/src/bucket/object_lock/objectlock.rs` is unused), so RustFS-to-RustFS site replication was unaffected and the gap was invisible in every in-tree test. + +## Why four layers of defense all missed it + +The same thing was missing at every layer: an inventory of what remote targets may require of an outbound request. + +- **Change design.** A default that applies to every target was changed to satisfy one target class. The PR text considered checksums only for objects that already carried one ("object-level `x-amz-checksum-*` forwarding is unaffected") and never asked the inverse question: which target-side rules depend on a checksum being present. +- **Tests.** The new unit test asserted the fix ("no `x-amz-trailer`"), not the contract ("the target accepted the PUT"). The fake target modeled neither failure mode: it did not store `aws-chunked` verbatim and it did not demand a checksum on locked PUTs. `replication-check` has an ObjectLock phase, but its probe PUT carries no retention headers, so it would have reported OK against the customer's target. +- **Review.** The compatibility lens (`.agents/skills/adversarial-validation/references/compatibility.md`) listed only inbound and on-disk concerns: `xl.meta`, proto, MinIO fixtures. It contained no line about outbound target behavior, so the lens passed while blind to the class. +- **Release.** Two new environment escape hatches shipped undocumented. The customer's second question in rustfs#7082 was whether a configuration option exists; it did, and it was unfindable. + +## Corrective actions + +| Action | Where | +| --- | --- | +| Outbound target matrix e2e: every target failure mode times every object shape, with an explicit expectation table. Known-red cells are pinned to an open issue and fail loudly when they start passing. | `crates/e2e_test/src/replication_target_matrix_test.rs` | +| Fake target gains the three failure modes a real fleet has shown: rejects `aws-chunked` uploads, requires a checksum on Object Lock PUTs (and verifies `Content-MD5`), mints its own version ids. | `crates/e2e_test/src/fake_s3_target/mod.rs` | +| Compatibility lens gains an outbound target section. | `.agents/skills/adversarial-validation/references/compatibility.md` | +| `AGENTS.md` names outbound client defaults as high risk and requires the matrix plus documented escape hatches. | `AGENTS.md`, Adversarial Validation | +| The two escape hatches from rustfs#6895 are documented. | `docs/operations/replication-outbound-transport.md` | +| Product fix for rustfs#7082: derive `Content-MD5` from the source ETag on locked PUTs. | tracked in rustfs#7082 | +| `replication-check` probe PUT carries retention headers when the target bucket has Object Lock. | tracked in rustfs#7082 | + +## SOP: changing an outbound client default + +Applies to any change in what `TargetClient`, `PutObjectOptions`, the remote SDK configuration, or the outbound header set sends by default. Treat the change as high risk for every target class, even when it fixes one. + +1. **Inventory the dependents.** Before editing, list every target-side rule that the current default satisfies, not only the one the change targets. Done when the PR body names each rule and which target classes enforce it. +2. **Run the matrix.** Run `replication_target_matrix_test` locally against the built binary. Done when every cell matches its expectation and no expectation was edited to make it pass. +3. **Extend the matrix for the new failure mode.** If the change is motivated by a target behavior the fake does not model, add the mode to the fake first and add a cell that is red before the fix. Done when the cell is green only with the fix applied. +4. **Document every escape hatch in the same PR.** Each new environment knob gets an entry in `docs/operations/replication-outbound-transport.md`. Done when the knob name appears there. +5. **Record the coverage.** The PR Impact section lists the target classes verified and, explicitly, the ones not verified. Done when a reader can tell which cells were never run. +6. **Soak before a release candidate.** A fix that changes outbound behavior runs once through the real-target lab, which must include an Object Lock target and an AWS or MinIO target, before it is tagged into a release candidate. + +## Related + +- rustfs/backlog#2085 — product boundary for generic S3 targets (version identity). +- `docs/operations/replication-object-size-limits.md` — single PUT and multipart limits on generic targets. +- `docs/operations/replication-check.md` — the probe phases and their contract. diff --git a/docs/testing/README.md b/docs/testing/README.md index 137285d4e..3589d8ee1 100644 --- a/docs/testing/README.md +++ b/docs/testing/README.md @@ -1,3 +1,4 @@ +| Outbound target matrix | Replication of every object shape (empty, plain, retention, legal hold, multipart) against every remote-target failure mode the fake target models; an explicit expectation table pins known-red cells to an open issue | `cargo nextest run -p e2e_test -E 'test(/^replication_target_matrix_test::/)'` (build `target/debug/rustfs` first) | With `e2e-repl-nightly`; required locally for any change to outbound client defaults (SOP: [`docs/postmortems/2026-09-03-replication-checksum-default-regression.md`](../postmortems/2026-09-03-replication-checksum-default-regression.md)) | # RustFS Testing **Use this when:** you need to pick a test layer for a change, name a test so a gate keeps selecting it, understand why `#[serial]` does nothing under nextest, or handle a flaky test. diff --git a/scripts/check_doc_paths.sh b/scripts/check_doc_paths.sh index c800ce219..0e7772046 100755 --- a/scripts/check_doc_paths.sh +++ b/scripts/check_doc_paths.sh @@ -21,7 +21,7 @@ cd "$ROOT_DIR" FAILURES=0 doc_files() { - git ls-files 'AGENTS.md' '*/AGENTS.md' 'CLAUDE.md' 'ARCHITECTURE.md' 'docs/*.md' 'docs/architecture/*.md' 'docs/operations/*.md' 'docs/testing/*.md' + git ls-files 'AGENTS.md' '*/AGENTS.md' 'CLAUDE.md' 'ARCHITECTURE.md' 'docs/*.md' 'docs/architecture/*.md' 'docs/operations/*.md' 'docs/postmortems/*.md' 'docs/testing/*.md' } check_file() {