From 2ecf6b4575836767e64bcc108c23848edeaef8f9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=94=90=E5=B0=8F=E9=B8=AD?= Date: Tue, 11 Aug 2026 11:04:05 +0800 Subject: [PATCH] fix(replication): probe the version-identity contract in replication-check (#5881) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * test(replication): pin the version-fidelity probe contract (red) P1-19 (rustfs/backlog#1675 B2): the supported replication contract is targets that adopt the source version id — a target that mints its own ids silently breaks every version-addressed operation that follows (version deletes, heal re-drives never match), diverging the two sides with no signal. replication-check already captures the probe PUT's response version id but never compares it. Red evidence (current main): against a FakeS3Target with assign_own_version_ids enabled, ?replication-check returns Status "OK" — the drift is invisible. test_replication_check_flags_version_minting_target expects a VersionFidelity phase that fails with the machine-readable code BucketRemoteTargetVersionMismatch, skips the later mutation phases, and still cleans up the probe via the version id the target actually assigned. Test infra: FakeS3Target gains assign_own_version_ids (models a generic S3 service; validated-but-not-mirrored source version headers) and a prefix+max-keys ListObjectVersions implementation (the probe key allocation requires it); stored_versions accessor duplicated from the P1-21 branch (identical code, resolves clean on merge). * fix(replication): probe the version-identity contract in replication-check P1-19 (rustfs/backlog#1675 B2, plan B). Replication only converges on targets that adopt the source version id: version-addressed deletes and heal re-drives address the source id, so a target that mints its own ids silently diverges — nothing surfaced this. replication-check already captured the probe PUT's response version id but never compared it. - The probe PUT now carries the source version as `?versionId=` (the exact shape live replication uses since P0-5, and the only shape MinIO consumes; the internal source-version-id header alone would let the probe pass against targets the real data path drifts on). Reuses ecstore's append_version_id_query through the api facade. - New VersionFidelity phase: the probe PUT's response version id must equal the sent source id. On mismatch the phase fails with the machine-readable extension key `"Code": "BucketRemoteTargetVersionMismatch"` (new optional Code field on phase statuses; Go decoders ignore unknown keys), the overall target fails, the later version-addressed mutation phases are skipped, and cleanup still removes the probe via the id the target actually assigned (with the existing list-based sweep as backstop when the target returns no version id at all). - Runtime half: TargetClient::put_object now returns the assigned version id (mirroring remove_object), and the replication PUT path audits it — every drifting PUT increments rustfs_replication_version_identity_drift_total and the first drift per target ARN logs a structured warning pointing at ?replication-check. The drift judgment is a pure function with an exemption-matrix test (empty / literal "null" / nil-uuid sources carry no contract). - docs/operations/replication-check.md documents the phase and the code. Red -> green: test_replication_check_flags_version_minting_target (fake target with assign_own_version_ids; on main the check reported Status "OK"). The probe's query shape is pinned by a journal assertion (revert of the query hunk alone fails it), probe-level unit tests cover the mismatch/mirror matrix including cleanup addressing the minted id, and the existing success e2e now asserts VersionFidelity OK against a RustFS target. Adversarial review (seven roles): non-blocking; noted follow-ups are the multipart runtime audit (the probe phase already pins the contract) and per-target re-warning after reconfiguration. * fix(e2e): stop the fake target self-deadlocking on version-id minting The assign_own_version_ids flag was read with a fresh `lock(&self.store)` inside two paths that already hold that guard — delete_object's marker-creation branch and create_multipart_upload — and the store mutex is not reentrant, so both hung forever (CI: the fake target's own multipart and delete-marker tests ran >1560s until the job was cancelled). Read the flag from the live guard instead. The replication e2e paths did not catch this: a version-addressed purge DELETE never mints an id, and the probe PUT reads the flag before taking the guard. * chore(test): refresh the nextest replication count invariant The e2e-smoke/e2e-repl-nightly split comment is descriptive metadata (authority: `cargo nextest list`); refresh it to this branch's post-rebase total. --- .config/nextest.toml | 4 +- crates/e2e_test/src/fake_s3_target/mod.rs | 95 +++- .../src/replication_extension_test.rs | 331 ++++++++++++++ crates/ecstore/src/api/mod.rs | 2 +- .../ecstore/src/bucket/bucket_target_sys.rs | 9 +- .../replication/replication_resyncer.rs | 121 +++++- docs/operations/replication-check.md | 17 + rustfs/src/admin/router.rs | 409 ++++++++++++++++-- rustfs/src/admin/storage_api.rs | 1 + 9 files changed, 929 insertions(+), 60 deletions(-) diff --git a/.config/nextest.toml b/.config/nextest.toml index 3073dae55..615c4c4fe 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -218,7 +218,7 @@ test-group = 'ecstore-serial-flaky' # the nightly profile derives its set as "the replication module MINUS this # allowlist", so any new replication test lands in nightly by default (never # silently unrun) until it is explicitly blessed as fast here. Keep the two -# regexes byte-identical. Count invariant: 20 here + 43 nightly = 63 total +# regexes byte-identical. Count invariant: 20 here + 47 nightly = 67 total # (authority: `cargo nextest list`; docs/testing/e2e-suite-inventory.md). # HISTORY (2026-07-11): the 20 fast tests were briefly pulled out of this lane # (#4724) because they set a loopback (127.0.0.1) replication target that the @@ -344,7 +344,7 @@ path = "junit.xml" # object_lambda) — too heavy for the merge budget; they run in ci-7's # nightly 4-node lane. # * replication_extension_test — repl-1 already splits it into the PR -# `e2e-smoke` (20 fast) and `e2e-repl-nightly` (43 slow) lanes and reserves +# `e2e-smoke` (20 fast) and `e2e-repl-nightly` (47 slow) lanes and reserves # it for those, so e2e-full does not double-run it. # * #[ignore]d tests — nextest skips them by default (no --run-ignored); the # manual-localhost:9000 reliant/policy tests are ci-13's migration. diff --git a/crates/e2e_test/src/fake_s3_target/mod.rs b/crates/e2e_test/src/fake_s3_target/mod.rs index c40058e8d..d094f8cdb 100644 --- a/crates/e2e_test/src/fake_s3_target/mod.rs +++ b/crates/e2e_test/src/fake_s3_target/mod.rs @@ -30,10 +30,10 @@ use s3s::access::{S3Access, S3AccessContext}; use s3s::auth::SimpleAuth; use s3s::dto::{ AbortMultipartUploadInput, AbortMultipartUploadOutput, CompleteMultipartUploadInput, CompleteMultipartUploadOutput, - CreateMultipartUploadInput, CreateMultipartUploadOutput, DeleteObjectInput, DeleteObjectOutput, ETag, + CreateMultipartUploadInput, CreateMultipartUploadOutput, DeleteMarkerEntry, DeleteObjectInput, DeleteObjectOutput, ETag, GetBucketVersioningInput, GetBucketVersioningOutput, GetObjectInput, GetObjectOutput, HeadBucketInput, HeadBucketOutput, - HeadObjectInput, HeadObjectOutput, PutObjectInput, PutObjectOutput, StreamingBlob, Timestamp, TimestampFormat, - UploadPartInput, UploadPartOutput, + HeadObjectInput, HeadObjectOutput, ListObjectVersionsInput, ListObjectVersionsOutput, ObjectVersionId, PutObjectInput, + PutObjectOutput, StreamingBlob, Timestamp, TimestampFormat, UploadPartInput, UploadPartOutput, }; use s3s::service::{S3Service, S3ServiceBuilder}; use s3s::validation::{AwsNameValidation, NameValidation}; @@ -91,6 +91,7 @@ pub enum Operation { GetObject, HeadObject, DeleteObject, + ListObjectVersions, CreateMultipartUpload, UploadPart, CompleteMultipartUpload, @@ -109,6 +110,8 @@ pub enum FaultAction { /// already have buffered the rest of the current frame; the journal reports /// the threshold, and the backend never receives or stores the request. DisconnectAfterBytes(usize), + /// Apply the request, then close the connection before returning its response. + DisconnectAfterResponse, /// Drain a request body in fixed-size slices, sleeping after every slice. SlowDrain { chunk_bytes: usize, delay: Duration }, /// Store the request normally but replace the response ETag. @@ -142,6 +145,7 @@ struct ControlState { #[derive(Default)] struct StoreState { assign_own_version_ids: bool, + assign_own_multipart_version_ids: bool, buckets: HashMap, uploads: HashMap, total_bytes: usize, @@ -390,6 +394,16 @@ impl FakeS3Target { lock(&self.backend.store).assign_own_version_ids = enabled; } + /// Mint own version ids for the multipart path only — models a target + /// that adopts PutObject version ids but not CreateMultipartUpload ones. + pub fn assign_own_multipart_version_ids(&self, enabled: bool) { + lock(&self.backend.store).assign_own_multipart_version_ids = enabled; + } + + pub fn active_multipart_upload_count(&self) -> usize { + lock(&self.backend.store).uploads.len() + } + /// Queue `times` copies of a fault for one operation. pub fn inject(&self, operation: Operation, action: FaultAction, times: usize) { if times == 0 { @@ -659,6 +673,7 @@ fn parse_request(method: &Method, uri: &Uri) -> ParsedRequest { let operation = match (method, key.is_some()) { (&Method::HEAD, false) => Operation::HeadBucket, (&Method::GET, false) if query.contains_key("versioning") => Operation::GetBucketVersioning, + (&Method::GET, false) if query.contains_key("versions") => Operation::ListObjectVersions, (&Method::PUT, true) if upload_id.is_some() && part_number.is_some() => Operation::UploadPart, (&Method::PUT, true) if upload_id.is_some() || query.contains_key("partNumber") => Operation::Unknown, (&Method::POST, true) if query.contains_key("uploads") => Operation::CreateMultipartUpload, @@ -810,7 +825,10 @@ async fn apply_non_body_fault(fault: Option<&RequestFault>, control: &Mutex Ok(()), + Some(FaultAction::SlowDrain { .. }) + | Some(FaultAction::WrongEtag) + | Some(FaultAction::DisconnectAfterResponse) + | None => Ok(()), } } @@ -851,7 +869,7 @@ async fn collect_stream( Some(FaultAction::SlowDrain { chunk_bytes, delay }) => { return collect_stream_slow(body, capacity, *chunk_bytes, *delay).await; } - Some(FaultAction::WrongEtag) | None => {} + Some(FaultAction::WrongEtag) | Some(FaultAction::DisconnectAfterResponse) | None => {} } let mut output = BytesMut::with_capacity(capacity); @@ -913,6 +931,9 @@ fn apply_response_fault(mut response: S3Response, fault: Option<&RequestFa if fault.is_some_and(|fault| fault.action == FaultAction::WrongEtag) { response.headers.insert(ETAG, HeaderValue::from_static(WRONG_ETAG)); } + if fault.is_some_and(|fault| fault.action == FaultAction::DisconnectAfterResponse) { + response.headers.insert(DISCONNECT_HEADER, HeaderValue::from_static("true")); + } response } @@ -1101,6 +1122,63 @@ impl S3 for FakeBackend { )) } + /// Prefix + max-keys subset only — enough for the replication-check probe + /// key allocation. No pagination markers or delimiter folding. + async fn list_object_versions( + &self, + req: S3Request, + ) -> S3Result> { + let fault = request_fault(&req); + apply_non_body_fault(fault.as_ref(), &self.control).await?; + let state = lock(&self.store); + let Some(bucket_state) = state.buckets.get(&req.input.bucket) else { + return Err(s3s::s3_error!(NoSuchBucket, "bucket does not exist")); + }; + let prefix = req.input.prefix.as_deref().unwrap_or_default(); + let max_keys = req.input.max_keys.unwrap_or(1000).max(0) as usize; + + let mut keys: Vec<&String> = bucket_state.objects.keys().filter(|key| key.starts_with(prefix)).collect(); + keys.sort(); + + let mut versions = Vec::new(); + let mut delete_markers = Vec::new(); + 'keys: for key in keys { + for version in bucket_state.objects[key].iter().rev() { + if versions.len() + delete_markers.len() >= max_keys { + break 'keys; + } + if version.delete_marker { + delete_markers.push(DeleteMarkerEntry { + key: Some(key.clone()), + version_id: Some(ObjectVersionId::from(version.version_id.clone())), + last_modified: Some(version.last_modified.clone()), + ..Default::default() + }); + } else { + versions.push(s3s::dto::ObjectVersion { + key: Some(key.clone()), + version_id: Some(ObjectVersionId::from(version.version_id.clone())), + last_modified: Some(version.last_modified.clone()), + e_tag: Some(ETag::Strong(version.e_tag.clone())), + size: Some(version.body.len() as i64), + ..Default::default() + }); + } + } + } + drop(state); + + Ok(apply_response_fault( + S3Response::new(ListObjectVersionsOutput { + name: Some(req.input.bucket), + versions: Some(versions), + delete_markers: Some(delete_markers), + ..Default::default() + }), + fault.as_ref(), + )) + } + async fn put_object(&self, req: S3Request) -> S3Result> { let fault = request_fault(&req); let _body_permit = timeout(MAX_FAULT_DURATION, Arc::clone(&self.body_limit).acquire_owned()) @@ -1155,7 +1233,7 @@ impl S3 for FakeBackend { content_type: version.content_type, metadata: version.metadata, e_tag: Some(ETag::Strong(version.e_tag)), - last_modified: Some(version.last_modified), + last_modified: Some(version.last_modified.clone()), version_id: Some(version.version_id), ..Default::default() }), @@ -1177,7 +1255,7 @@ impl S3 for FakeBackend { content_type: version.content_type, metadata: version.metadata, e_tag: Some(ETag::Strong(version.e_tag)), - last_modified: Some(version.last_modified), + last_modified: Some(version.last_modified.clone()), version_id: Some(version.version_id), ..Default::default() }), @@ -1290,7 +1368,8 @@ impl S3 for FakeBackend { let upload_id = Uuid::new_v4().to_string(); // Read the flag before the mutable borrow of `state.uploads` below // (and never re-lock the store: the mutex is not reentrant). - let version_id = new_version_id(&headers, state.assign_own_version_ids)?; + let mint_own = state.assign_own_version_ids || state.assign_own_multipart_version_ids; + let version_id = new_version_id(&headers, mint_own)?; state.uploads.insert( upload_id.clone(), MultipartState { diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index bd3e72223..bb02eb604 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/crates/e2e_test/src/replication_extension_test.rs @@ -2595,6 +2595,9 @@ async fn test_replication_check_succeeds_with_remote_target() -> Result<(), Box< assert_eq!(payload["Targets"].as_array().map(Vec::len), Some(1)); assert_eq!(payload["Targets"][0]["Status"], "OK"); assert_eq!(payload["Targets"][0]["Phases"]["Put"]["Status"], "OK"); + // A RustFS target adopts the source version id, so the P1-19 + // version-identity probe passes. + assert_eq!(payload["Targets"][0]["Phases"]["VersionFidelity"]["Status"], "OK"); assert_eq!(payload["Targets"][0]["Phases"]["DeleteMarker"]["Status"], "OK"); assert_eq!(payload["Targets"][0]["Phases"]["VersionDelete"]["Status"], "OK"); assert_eq!(payload["Targets"][0]["Phases"]["Cleanup"]["Status"], "OK"); @@ -7489,6 +7492,334 @@ async fn test_scanner_never_cascades_inbound_replicas() -> TestResult { Ok(()) } +/// P1-19 review follow-up: multipart fixes the target version at initiate +/// and only reports it on completion, so a target can adopt PutObject +/// version ids and still mint its own there — the check must not report OK +/// while multipart deletes and heals would silently miss. +#[tokio::test] +#[serial] +async fn test_replication_check_flags_multipart_only_version_minting_target() -> TestResult { + init_logging(); + + let target = FakeS3Target::start().await?; + let target_bucket = "multipart-fidelity-dst"; + target.create_bucket(target_bucket); + // PutObject mirrors the source version id; CreateMultipartUpload does not. + target.assign_own_multipart_version_ids(true); + + 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 = "multipart-fidelity-src"; + let source_client = source_env.create_s3_client(); + source_client.create_bucket().bucket(source_bucket).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, + secure: false, + skip_tls_verify: false, + ca_cert_pem: None, + }, + ) + .await?; + put_bucket_replication(&source_env, source_bucket, &target_arn).await?; + + let response = run_replication_check(&source_env, source_bucket).await?; + assert_eq!(response.status(), StatusCode::OK); + let payload: serde_json::Value = response.json().await?; + + assert_eq!(payload["Status"], "FAILED", "multipart drift must fail the check: {payload}"); + let target_report = &payload["Targets"][0]; + let fidelity = &target_report["Phases"]["VersionFidelity"]; + assert_eq!(fidelity["Status"], "FAILED", "{payload}"); + assert_eq!(fidelity["Code"], "BucketRemoteTargetVersionMismatch", "{payload}"); + assert!( + fidelity["Error"] + .as_str() + .is_some_and(|error| error.contains("CreateMultipartUpload")), + "the failure must name the multipart path: {payload}" + ); + // The PutObject leg mirrored, so it is the multipart probe that failed. + assert_eq!(target_report["Phases"]["Put"]["Status"], "OK", "{payload}"); + assert_eq!(target_report["Phases"]["DeleteMarker"]["Status"], "SKIPPED", "{payload}"); + assert_eq!(target_report["Phases"]["Cleanup"]["Status"], "OK", "{payload}"); + + let probe_key = target + .requests() + .into_iter() + .find(|record| record.operation == FakeTargetOperation::PutObject) + .and_then(|record| record.key) + .ok_or("the probe PUT never reached the fake target")?; + assert!( + target.stored_versions(target_bucket, &probe_key).is_empty(), + "both probe versions must be cleaned up on the mismatching target" + ); + + target.shutdown().await; + Ok(()) +} + +#[tokio::test] +#[serial] +async fn test_replication_check_aborts_failed_multipart_probes() -> TestResult { + init_logging(); + + let target = FakeS3Target::start().await?; + let target_bucket = "multipart-cleanup-dst"; + target.create_bucket(target_bucket); + + 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 = "multipart-cleanup-src"; + let source_client = source_env.create_s3_client(); + source_client.create_bucket().bucket(source_bucket).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, + secure: false, + skip_tls_verify: false, + ca_cert_pem: None, + }, + ) + .await?; + put_bucket_replication(&source_env, source_bucket, &target_arn).await?; + + for failed_operation in [FakeTargetOperation::UploadPart, FakeTargetOperation::CompleteMultipartUpload] { + target.clear_faults(); + target.take_requests(); + target.inject(failed_operation, FakeTargetFault::Status(StatusCode::SERVICE_UNAVAILABLE), 16); + + let response = run_replication_check(&source_env, source_bucket).await?; + assert_eq!(response.status(), StatusCode::OK); + let payload: serde_json::Value = response.json().await?; + assert_eq!( + payload["Status"], "FAILED", + "the injected multipart failure must fail the check: {payload}" + ); + + let requests = target.requests(); + assert!( + requests.iter().any(|request| { + request.operation == failed_operation + && request.fault == Some(FakeTargetFault::Status(StatusCode::SERVICE_UNAVAILABLE)) + }), + "the check must reach the injected {failed_operation:?} failure: {requests:?}" + ); + assert!( + requests + .iter() + .any(|request| request.operation == FakeTargetOperation::AbortMultipartUpload), + "the failed {failed_operation:?} probe must be aborted: {requests:?}" + ); + assert_eq!( + target.active_multipart_upload_count(), + 0, + "the failed {failed_operation:?} probe must not leave multipart state" + ); + assert_eq!(payload["Targets"][0]["Phases"]["Cleanup"]["Status"], "OK", "{payload}"); + } + + target.clear_faults(); + target.take_requests(); + target.inject(FakeTargetOperation::CompleteMultipartUpload, FakeTargetFault::DisconnectAfterResponse, 16); + + let response = run_replication_check(&source_env, source_bucket).await?; + assert_eq!(response.status(), StatusCode::OK); + let payload: serde_json::Value = response.json().await?; + let target_report = &payload["Targets"][0]; + assert_eq!(target_report["Status"], "FAILED", "{payload}"); + assert_eq!(target_report["Phases"]["VersionFidelity"]["Status"], "FAILED", "{payload}"); + assert_eq!( + target_report["Phases"]["Cleanup"]["Status"], "OK", + "NoSuchUpload after an ambiguous complete means the multipart artifact is gone: {payload}" + ); + let requests = target.requests(); + let completed_key = requests + .iter() + .find(|request| { + request.operation == FakeTargetOperation::CompleteMultipartUpload + && request.fault == Some(FakeTargetFault::DisconnectAfterResponse) + }) + .and_then(|request| request.key.as_deref()) + .expect("the scripted complete response disconnect must be observed"); + assert!( + requests + .iter() + .any(|request| request.operation == FakeTargetOperation::AbortMultipartUpload), + "the ambiguous complete must still attempt abort: {requests:?}" + ); + assert_eq!(target.active_multipart_upload_count(), 0); + assert!( + target.stored_versions(target_bucket, completed_key).is_empty(), + "outer cleanup must remove the object committed before the response disconnect" + ); + + target.clear_faults(); + target.take_requests(); + target.inject(FakeTargetOperation::UploadPart, FakeTargetFault::Status(StatusCode::FORBIDDEN), 16); + target.inject( + FakeTargetOperation::AbortMultipartUpload, + FakeTargetFault::Status(StatusCode::SERVICE_UNAVAILABLE), + 16, + ); + + let response = run_replication_check(&source_env, source_bucket).await?; + assert_eq!(response.status(), StatusCode::OK); + let payload: serde_json::Value = response.json().await?; + let target_report = &payload["Targets"][0]; + assert_eq!(target_report["Status"], "FAILED", "{payload}"); + assert_eq!(target_report["Phases"]["VersionFidelity"]["Status"], "FAILED", "{payload}"); + assert_eq!( + target_report["Phases"]["Cleanup"]["Status"], "FAILED", + "an unremoved multipart probe must be reported as a cleanup failure: {payload}" + ); + assert_eq!( + target_report["Error"], "s3:ReplicateObject permissions missing for replication user", + "the primary multipart error must remain the target error: {payload}" + ); + assert_eq!( + target_report["Phases"]["VersionFidelity"]["Error"], "s3:ReplicateObject permissions missing for replication user", + "{payload}" + ); + assert_eq!( + target_report["Phases"]["Cleanup"]["Error"], "failed to abort multipart replication probe", + "{payload}" + ); + assert!( + target.requests().iter().any(|request| { + request.operation == FakeTargetOperation::AbortMultipartUpload + && request.fault == Some(FakeTargetFault::Status(StatusCode::SERVICE_UNAVAILABLE)) + }), + "the abort failure must be observed" + ); + assert_eq!( + target.active_multipart_upload_count(), + 1, + "the report must match the retained multipart state" + ); + + target.shutdown().await; + Ok(()) +} + +/// P1-19 (backlog#1675): the supported replication contract is targets that +/// adopt the source version id (RustFS/MinIO semantics). A target that mints +/// its own version ids silently breaks every version-addressed operation that +/// follows — version deletes and heal re-drives never match, diverging the +/// two sides. replication-check must surface this explicitly: a +/// VersionFidelity phase that compares the probe PUT's response version id +/// against the sent source version id and fails with +/// BucketRemoteTargetVersionMismatch — while still cleaning up the probe +/// object via the version id the target actually assigned. +#[tokio::test] +#[serial] +async fn test_replication_check_flags_version_minting_target() -> TestResult { + init_logging(); + + let target = FakeS3Target::start().await?; + let target_bucket = "version-fidelity-dst"; + target.create_bucket(target_bucket); + target.assign_own_version_ids(true); + + 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 = "version-fidelity-src"; + let source_client = source_env.create_s3_client(); + source_client.create_bucket().bucket(source_bucket).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, + secure: false, + skip_tls_verify: false, + ca_cert_pem: None, + }, + ) + .await?; + put_bucket_replication(&source_env, source_bucket, &target_arn).await?; + + let response = run_replication_check(&source_env, source_bucket).await?; + assert_eq!(response.status(), StatusCode::OK); + let payload: serde_json::Value = response.json().await?; + + assert_eq!( + payload["Status"], "FAILED", + "a version-minting target must fail the replication check: {payload}" + ); + let target_report = &payload["Targets"][0]; + assert_eq!(target_report["Status"], "FAILED", "target must be FAILED: {payload}"); + let fidelity = &target_report["Phases"]["VersionFidelity"]; + assert_eq!(fidelity["Status"], "FAILED", "VersionFidelity phase must fail: {payload}"); + assert_eq!( + fidelity["Code"], "BucketRemoteTargetVersionMismatch", + "the failure must carry a machine-readable code: {payload}" + ); + // The probe PUT itself succeeded (fidelity is judged from its response); + // the later mutation phases are pointless against a drifting target and + // must be skipped, but cleanup still runs. + assert_eq!(target_report["Phases"]["Put"]["Status"], "OK", "{payload}"); + assert_eq!(target_report["Phases"]["DeleteMarker"]["Status"], "SKIPPED", "{payload}"); + assert_eq!(target_report["Phases"]["Cleanup"]["Status"], "OK", "{payload}"); + + // The probe PUT must carry the source version as `?versionId=` — the + // exact shape live replication uses (P0-5), and the only shape MinIO + // consumes. The journal records the query value. + let probe_put = target + .requests() + .into_iter() + .find(|record| record.operation == FakeTargetOperation::PutObject) + .ok_or("the probe PUT never reached the fake target")?; + let probe_query_version = probe_put + .version_id + .as_deref() + .ok_or("the probe PUT must carry a versionId query")?; + assert!( + uuid::Uuid::parse_str(probe_query_version).is_ok(), + "the probe versionId query must be the source uuid, got {probe_query_version}" + ); + + // No probe residue: cleanup must address the version id the target + // actually assigned, not the source id (which never matched anything). + let probe_key = probe_put.key.ok_or("probe PUT journal record has no key")?; + assert!( + target.stored_versions(target_bucket, &probe_key).is_empty(), + "the probe object must be cleaned up on the mismatching target" + ); + + Ok(()) +} + // --- P1-21 (backlog#1675): delayed delete-marker purge failure handling --- // // The fixtures below wire a versioned source bucket to a FakeS3Target with the diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 9a63c7cd9..f7b930241 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -32,7 +32,7 @@ pub mod bucket { pub mod bucket_target_sys { pub use crate::bucket::bucket_target_sys::{ AdvancedPutOptions, BucketTargetError, BucketTargetSys, PutObjectOptions, RemoveObjectOptions, S3ClientError, - TargetClient, + TargetClient, append_version_id_query, }; } diff --git a/crates/ecstore/src/bucket/bucket_target_sys.rs b/crates/ecstore/src/bucket/bucket_target_sys.rs index 1cf1ce833..1e1e18dc0 100644 --- a/crates/ecstore/src/bucket/bucket_target_sys.rs +++ b/crates/ecstore/src/bucket/bucket_target_sys.rs @@ -1450,7 +1450,7 @@ fn resolve_put_api_version_id(source_version_id: &str) -> Option<&str> { /// member, so the query is spliced in via `map_request`, which runs at /// `modify_before_signing`: the parameter becomes part of the SigV4 canonical /// request. -fn append_version_id_query(uri: &str, version_id: &str) -> String { +pub fn append_version_id_query(uri: &str, version_id: &str) -> String { let separator = if uri.contains('?') { '&' } else { '?' }; format!("{uri}{separator}versionId={}", urlencoding::encode(version_id)) } @@ -1861,6 +1861,9 @@ impl TargetClient { } } + /// On success returns the version id the target assigned (from + /// `x-amz-version-id`), letting callers audit the version-identity + /// contract — a target that adopts the source version echoes it back. pub async fn put_object( &self, bucket: &str, @@ -1868,7 +1871,7 @@ impl TargetClient { size: i64, body: ByteStream, opts: &PutObjectOptions, - ) -> Result<(), S3ClientError> { + ) -> Result, S3ClientError> { let mut headers = opts.header(); let builder = self.client.put_object(); @@ -1903,7 +1906,7 @@ impl TargetClient { .send() .await { - Ok(_) => Ok(()), + Ok(output) => Ok(output.version_id().map(ToOwned::to_owned)), Err(e) => match e { SdkError::ServiceError(service_err) => { let err = service_err.into_err(); diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index a7a2b6a7a..c6cdbcb03 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -74,10 +74,10 @@ use rustfs_utils::http::{ use rustfs_utils::{DEFAULT_SIP_HASH_KEY, get_env_usize, sip_hash}; #[cfg(test)] use s3s::dto::ReplicationConfiguration; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::fmt::Display; -use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, LazyLock, Mutex as StdMutex}; use time::OffsetDateTime; use time::format_description::well_known::Rfc3339; use tokio::io::AsyncRead; @@ -105,6 +105,7 @@ const EVENT_RESYNC_RUNTIME_CHANNEL_FAILED: &str = "replication_resync_runtime_ch const EVENT_DELETE_MARKER_PURGE_FAILED: &str = "replication_delete_marker_purge_failed"; const EVENT_DELETE_MARKER_PURGE_MRF: &str = "replication_delete_marker_purge_mrf"; const METRIC_DELETE_MARKER_PURGE_TOTAL: &str = "rustfs_replication_delete_marker_purge_total"; +const EVENT_REPLICATION_VERSION_IDENTITY_DRIFT: &str = "replication_version_identity_drift"; const REPLICATION_TARGET_OFFLINE_ERROR_MARKERS: &[&str] = &[ "dispatch failure", "timeouterror", @@ -186,6 +187,57 @@ fn is_head_proxy_failure(err: &SdkError) -> bool { should_count_head_proxy_failure(is_not_found, code, raw_status) } +const METRIC_VERSION_IDENTITY_DRIFT_TOTAL: &str = "rustfs_replication_version_identity_drift_total"; + +/// Targets that already produced a version-identity-drift warning this +/// process lifetime, by ARN. Deduping is advisory only (the metric still +/// counts every drifting PUT), so a reconfigured target re-warning only +/// after a restart is acceptable. +static VERSION_IDENTITY_WARNED_ARNS: LazyLock>> = LazyLock::new(|| StdMutex::new(HashSet::new())); + +/// Runtime half of the P1-19 version-identity contract (the explicit probe +/// lives in replication-check's VersionFidelity phase): every replication PUT +/// response reveals whether the target adopted the source version id. A +/// target minting its own ids silently breaks version-addressed deletes and +/// heal, so surface it — once per target — instead of letting the divergence +/// accumulate unseen. +/// Pure drift judgment: the contract only applies when the source addressed a +/// real (non-nil) version uuid, and drift means the target answered with +/// anything else — including nothing at all. +fn version_identity_drifted(source_version_id: &str, assigned_version_id: Option<&str>) -> bool { + if source_version_id.is_empty() { + return false; + } + // A nil source uuid travels as the literal "null" (unversioned-source + // semantics); no identity contract applies to it. + if Uuid::parse_str(source_version_id).map(|uuid| uuid.is_nil()).unwrap_or(true) { + return false; + } + assigned_version_id != Some(source_version_id) +} + +fn audit_target_version_identity(tgt_client: &TargetClient, source_version_id: &str, assigned_version_id: Option<&str>) { + if !version_identity_drifted(source_version_id, assigned_version_id) { + return; + } + counter!(METRIC_VERSION_IDENTITY_DRIFT_TOTAL).increment(1); + let mut warned = VERSION_IDENTITY_WARNED_ARNS + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + if warned.insert(tgt_client.arn.clone()) { + warn!( + event = EVENT_REPLICATION_VERSION_IDENTITY_DRIFT, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + arn = %tgt_client.arn, + endpoint = %tgt_client.endpoint, + sent_version_id = %source_version_id, + assigned_version_id = assigned_version_id.unwrap_or(""), + "Replication target does not adopt source version ids; version-addressed replication cannot converge (run ?replication-check for details)" + ); + } +} + async fn record_proxy_request(bucket: &str, api: &str, is_err: bool) { if let Some(stats) = runtime_sources::replication_stats() { stats.inc_proxy(bucket, api, is_err).await; @@ -2872,6 +2924,13 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { let result = tgt_client .put_object(&tgt_client.bucket, &object, transfer_size, byte_stream, &put_opts) .await + .map(|assigned_version_id| { + audit_target_version_identity( + &tgt_client, + &put_opts.internal.source_version_id, + assigned_version_id.as_deref(), + ) + }) .map_err(|e| std::io::Error::other(e.to_string())); record_proxy_request(&bucket, "PutObject", result.is_err()).await; if has_tagging_replication { @@ -3279,6 +3338,13 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { let result = tgt_client .put_object(&tgt_client.bucket, &object, transfer_size, byte_stream, &put_opts) .await + .map(|assigned_version_id| { + audit_target_version_identity( + &tgt_client, + &put_opts.internal.source_version_id, + assigned_version_id.as_deref(), + ) + }) .map_err(|e| std::io::Error::other(e.to_string())); record_proxy_request(&bucket, "PutObject", result.is_err()).await; if has_tagging_replication { @@ -3470,15 +3536,27 @@ async fn replicate_object_with_multipart(ctx: MultipartR let actual_size = replication_multipart_complete_actual_size(&object_info.user_defined); - cli.complete_multipart_upload( - dst_bucket, - object, - &upload_id, - uploaded_parts, - &replication_complete_multipart_options(actual_size, object_info.etag.clone().unwrap_or_default(), object_info.mod_time), - ) - .await - .map_err(|e| std::io::Error::other(e.to_string()))?; + let completed = cli + .complete_multipart_upload( + dst_bucket, + object, + &upload_id, + uploaded_parts, + &replication_complete_multipart_options( + actual_size, + object_info.etag.clone().unwrap_or_default(), + object_info.mod_time, + ), + ) + .await + .map_err(|e| std::io::Error::other(e.to_string()))?; + + // Multipart decides the target version at initiate time and only reveals + // it on completion, so this is where the identity contract is observable + // for this path. A target can mirror PutObject version ids and still mint + // its own here, which would leave multipart deletes and heals addressing + // a version that never existed. + audit_target_version_identity(&cli, &put_opts.internal.source_version_id, completed.version_id()); Ok(()) } @@ -3525,6 +3603,27 @@ mod tests { ReplicationTargetStore::register_test_target(target).await; } + /// P1-19 runtime spot-check exemption matrix: drift only applies when the + /// source addressed a real version uuid. + #[test] + fn test_version_identity_drift_judgment() { + let source = "6fa459ea-ee8a-3ca4-894e-db77e160355e"; + for (sent, got, expected) in [ + (source, Some(source), false), + (source, Some("0e304ce5-33e9-4b8a-9b12-9e40a53e6ded"), true), + (source, None, true), + ("", None, false), + ("null", Some("anything"), false), + ("00000000-0000-0000-0000-000000000000", Some("anything"), false), + ] { + assert_eq!( + version_identity_drifted(sent, got), + expected, + "sent {sent:?} got {got:?} must judge drift = {expected}" + ); + } + } + #[test] fn resync_admission_configuration_is_bounded() { assert_eq!(ENV_REPL_RESYNC_MAX_JOBS, "RUSTFS_REPL_RESYNC_MAX_JOBS"); diff --git a/docs/operations/replication-check.md b/docs/operations/replication-check.md index 5a03d1440..c623a1300 100644 --- a/docs/operations/replication-check.md +++ b/docs/operations/replication-check.md @@ -42,6 +42,7 @@ target results remain present when another target fails. "Versioning": { "Status": "OK" }, "ObjectLock": { "Status": "OK" }, "Put": { "Status": "OK" }, + "VersionFidelity": { "Status": "OK" }, "DeleteMarker": { "Status": "OK" }, "VersionDelete": { "Status": "OK" }, "Cleanup": { @@ -58,3 +59,19 @@ Phase states are `OK`, `FAILED`, or `SKIPPED`. Errors are single-line, bounded to 512 bytes, and omit remote messages, endpoints, credentials, signatures, and authorization material. A cleanup failure is always explicit; it is never reported as a successful check. + +`VersionFidelity` pins the version-identity contract on **both** write paths: +the probe PUT carries a source version id (header plus `?versionId=` query, +the exact shape live replication uses) and the target must answer with the +same id, and a second probe repeats it through CreateMultipartUpload -> +UploadPart -> CompleteMultipartUpload, where the target fixes the version at +initiate and only reports it on completion. A target can adopt PutObject ids +and still mint its own for multipart, which would leave multipart deletes and +heals addressing a version that never existed; the failure message names the +path that drifted. Targets that +mint their own version ids break every version-addressed operation that +follows (version deletes, heal re-drives), so the phase fails with the +machine-readable extension key `"Code": "BucketRemoteTargetVersionMismatch"`, +the later mutation phases are skipped, and cleanup still removes the probe via +the version id the target actually assigned. `Code` only appears on failures +that callers are expected to branch on; Go decoders ignore the unknown key. diff --git a/rustfs/src/admin/router.rs b/rustfs/src/admin/router.rs index d7249404d..13a45fd35 100644 --- a/rustfs/src/admin/router.rs +++ b/rustfs/src/admin/router.rs @@ -17,7 +17,7 @@ use super::storage_api::bucket::metadata_sys; use super::storage_api::bucket::replication::{self, BucketReplicationResyncStatus, BucketStats, ReplicationStatusType}; use super::storage_api::bucket::target::{BucketTarget, BucketTargetType, BucketTargets}; use super::storage_api::bucket::target_sys::{ - BucketTargetSys, PutObjectOptions, RemoveObjectOptions, S3ClientError, TargetClient, + BucketTargetSys, PutObjectOptions, RemoveObjectOptions, S3ClientError, TargetClient, append_version_id_query, }; use super::storage_api::bucket::versioning_sys::BucketVersioningSys; use super::storage_api::bucket::{AdminReplicationConfigExt as _, AdminVersioningConfigExt as _}; @@ -42,6 +42,7 @@ use crate::server::{ }; use crate::storage::storage_api::lock_bucket_targets_metadata; use aws_sdk_s3::primitives::ByteStream as AwsByteStream; +use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart}; use bytes::Bytes; use futures::{Stream, StreamExt}; use http::HeaderValue; @@ -206,6 +207,9 @@ struct ReplicationResetStatusTarget { const REPLICATION_CHECK_PROBE_PREFIX: &str = ".rustfs.sys/replication-check/"; const REPLICATION_CHECK_ERROR_MAX_BYTES: usize = 512; +/// RustFS extension code (no madmin analogue): the target does not adopt the +/// source version id, breaking the version-identity replication contract. +const REPLICATION_CHECK_CODE_VERSION_MISMATCH: &str = "BucketRemoteTargetVersionMismatch"; #[derive(Debug, Clone, serde::Serialize)] struct ReplicationCheckResponse { @@ -245,6 +249,8 @@ struct ReplicationCheckPhases { object_lock: ReplicationCheckPhaseStatus, #[serde(rename = "Put")] put: ReplicationCheckPhaseStatus, + #[serde(rename = "VersionFidelity")] + version_fidelity: ReplicationCheckPhaseStatus, #[serde(rename = "DeleteMarker")] delete_marker: ReplicationCheckPhaseStatus, #[serde(rename = "VersionDelete")] @@ -259,6 +265,11 @@ struct ReplicationCheckPhaseStatus { status: &'static str, #[serde(rename = "Error", skip_serializing_if = "Option::is_none")] error: Option, + /// Machine-readable failure code (RustFS extension key; Go decoders + /// ignore unknown keys). Only set for failures that a caller is expected + /// to branch on, e.g. `BucketRemoteTargetVersionMismatch`. + #[serde(rename = "Code", skip_serializing_if = "Option::is_none")] + code: Option<&'static str>, } impl Default for ReplicationCheckPhaseStatus { @@ -266,6 +277,7 @@ impl Default for ReplicationCheckPhaseStatus { Self { status: "SKIPPED", error: None, + code: None, } } } @@ -275,6 +287,7 @@ impl ReplicationCheckPhaseStatus { Self { status: "OK", error: None, + code: None, } } @@ -282,6 +295,14 @@ impl ReplicationCheckPhaseStatus { Self { status: "FAILED", error: Some(bound_replication_check_error(error.into())), + code: None, + } + } + + fn failed_with_code(error: impl Into, code: &'static str) -> Self { + Self { + code: Some(code), + ..Self::failed(error) } } } @@ -2046,12 +2067,39 @@ fn fail_replication_check_target(result: &mut ReplicationCheckTargetStatus, erro } } +/// The probe PUT reports both sides of the version-identity contract: the +/// source version id it sent (header + `?versionId=` query, the exact shape +/// live replication uses) and the version id the target answered with. +struct ReplicationProbePutOutcome { + sent_version_id: String, + response_version_id: Option, +} + +struct ReplicationProbeMultipartError { + primary: S3ClientError, + cleanup_error: Option, +} + +impl From for ReplicationProbeMultipartError { + fn from(primary: S3ClientError) -> Self { + Self { + primary, + cleanup_error: None, + } + } +} + #[async_trait::async_trait] trait ReplicationProbeOperations { - async fn put(&mut self) -> Result, S3ClientError>; + async fn put(&mut self) -> Result; + /// Multipart decides the target version at initiate time and only reports + /// it on completion, so the identity contract has to be probed separately + /// there: a target can adopt PutObject version ids and still mint its own + /// for CreateMultipartUpload. + async fn multipart_put(&mut self) -> Result; async fn create_delete_marker(&mut self, version_id: Option<&str>) -> Result, S3ClientError>; async fn delete_version(&mut self, version_id: Option<&str>) -> Result<(), S3ClientError>; - async fn cleanup(&mut self, known_version_ids: [Option<&str>; 2]) -> Result<(), String>; + async fn cleanup(&mut self, known_version_ids: [Option<&str>; 3]) -> Result<(), String>; } struct RemoteReplicationProbeOperations<'a> { @@ -2063,10 +2111,14 @@ struct RemoteReplicationProbeOperations<'a> { #[async_trait::async_trait] impl ReplicationProbeOperations for RemoteReplicationProbeOperations<'_> { - async fn put(&mut self) -> Result, S3ClientError> { + async fn put(&mut self) -> Result { put_replication_probe_object(self.client, self.bucket, self.key, self.time).await } + async fn multipart_put(&mut self) -> Result { + multipart_put_replication_probe_object(self.client, self.bucket, self.key, self.time).await + } + async fn create_delete_marker(&mut self, version_id: Option<&str>) -> Result, S3ClientError> { delete_replication_probe_object( self.client, @@ -2090,20 +2142,49 @@ impl ReplicationProbeOperations for RemoteReplicationProbeOperations<'_> { .map(|_| ()) } - async fn cleanup(&mut self, known_version_ids: [Option<&str>; 2]) -> Result<(), String> { + async fn cleanup(&mut self, known_version_ids: [Option<&str>; 3]) -> Result<(), String> { cleanup_replication_probe(self.client, self.bucket, self.key, known_version_ids).await } } +/// `None` when the target adopted the source version id on this path. +fn version_fidelity_error(api: &str, outcome: &ReplicationProbePutOutcome) -> Option { + if outcome.response_version_id.as_deref() == Some(outcome.sent_version_id.as_str()) { + return None; + } + Some(format!( + "target assigned version id {} instead of adopting the source version id {} on {api}; \ + version-addressed replication (version deletes, heal) cannot converge on this target", + outcome.response_version_id.as_deref().unwrap_or(""), + outcome.sent_version_id, + )) +} + async fn execute_replication_probe(result: &mut ReplicationCheckTargetStatus, operations: &mut impl ReplicationProbeOperations) { let mut probe_version_id = None; + let mut multipart_probe_version_id = None; let mut delete_marker_version_id = None; let mut cleanup_required = true; + let mut multipart_cleanup_error = None; match operations.put().await { - Ok(version_id) => { - probe_version_id = version_id; + Ok(outcome) => { result.phases.put = ReplicationCheckPhaseStatus::passed(); + // P1-19 version-identity contract: replication only converges on + // targets that adopt the source version id — version-addressed + // deletes and heal re-drives never match a minted id. Judge it + // from the probe PUT's own response; on mismatch the later + // mutation phases are pointless (they address by version id), but + // cleanup still runs against whatever id the target assigned. + match version_fidelity_error("PutObject", &outcome) { + None => result.phases.version_fidelity = ReplicationCheckPhaseStatus::passed(), + Some(error) => { + result.phases.version_fidelity = + ReplicationCheckPhaseStatus::failed_with_code(&error, REPLICATION_CHECK_CODE_VERSION_MISMATCH); + fail_replication_check_target(result, error); + } + } + probe_version_id = outcome.response_version_id; } Err(err) => { let error = format_replication_check_client_error(&err, ReplicationCheckFailureContext::ReplicateObject); @@ -2115,7 +2196,29 @@ async fn execute_replication_probe(result: &mut ReplicationCheckTargetStatus, op } } - if result.phases.put.status == "OK" { + // The multipart path fixes the target version at initiate and only + // reports it on completion, so a target can adopt PutObject ids and still + // mint its own here — probe it before declaring the contract met. + if result.phases.version_fidelity.status == "OK" { + match operations.multipart_put().await { + Ok(outcome) => { + multipart_probe_version_id = outcome.response_version_id.clone(); + if let Some(error) = version_fidelity_error("CreateMultipartUpload", &outcome) { + result.phases.version_fidelity = + ReplicationCheckPhaseStatus::failed_with_code(&error, REPLICATION_CHECK_CODE_VERSION_MISMATCH); + fail_replication_check_target(result, error); + } + } + Err(err) => { + let error = format_replication_check_client_error(&err.primary, ReplicationCheckFailureContext::ReplicateObject); + result.phases.version_fidelity = ReplicationCheckPhaseStatus::failed(&error); + fail_replication_check_target(result, error); + multipart_cleanup_error = err.cleanup_error; + } + } + } + + if result.phases.put.status == "OK" && result.phases.version_fidelity.status == "OK" { match operations.create_delete_marker(probe_version_id.as_deref()).await { Ok(version_id) => { delete_marker_version_id = version_id; @@ -2138,19 +2241,31 @@ async fn execute_replication_probe(result: &mut ReplicationCheckTargetStatus, op } } - if cleanup_required { - match operations - .cleanup([probe_version_id.as_deref(), delete_marker_version_id.as_deref()]) + let cleanup_result = if cleanup_required { + operations + .cleanup([ + probe_version_id.as_deref(), + multipart_probe_version_id.as_deref(), + delete_marker_version_id.as_deref(), + ]) .await - { - Ok(()) => result.phases.cleanup = ReplicationCheckPhaseStatus::passed(), - Err(error) => { - result.phases.cleanup = ReplicationCheckPhaseStatus::failed(&error); - fail_replication_check_target(result, format!("probe cleanup failed: {error}")); - } - } } else { + Ok(()) + }; + + let mut cleanup_errors = Vec::new(); + if let Some(error) = multipart_cleanup_error { + cleanup_errors.push(error); + } + if let Err(error) = cleanup_result { + cleanup_errors.push(error); + } + if cleanup_errors.is_empty() { result.phases.cleanup = ReplicationCheckPhaseStatus::passed(); + } else { + let error = cleanup_errors.join("; "); + result.phases.cleanup = ReplicationCheckPhaseStatus::failed(&error); + fail_replication_check_target(result, format!("probe cleanup failed: {error}")); } } @@ -2225,13 +2340,7 @@ fn build_replication_probe_remove_options(now: OffsetDateTime, replication_delet } } -async fn put_replication_probe_object( - target_client: &TargetClient, - target_bucket: &str, - probe_key: &str, - now: OffsetDateTime, -) -> Result, S3ClientError> { - let options = build_replication_probe_put_options(now); +fn build_replication_probe_headers(options: &PutObjectOptions) -> HeaderMap { let mut headers = HeaderMap::new(); insert_header(&mut headers, SUFFIX_SOURCE_VERSION_ID, &options.internal.source_version_id); insert_header( @@ -2245,8 +2354,166 @@ async fn put_replication_probe_object( HeaderName::from_static("x-amz-replication-status"), HeaderValue::from_static(ReplicationStatusType::Replica.as_str()), ); + headers +} - target_client +/// Probe the identity contract on the multipart path: initiate carrying the +/// source version as `?versionId=` (where the target fixes the version), +/// upload one small part, and read the version the completion reports. +async fn multipart_put_replication_probe_object( + target_client: &TargetClient, + target_bucket: &str, + probe_key: &str, + now: OffsetDateTime, +) -> Result { + let options = build_replication_probe_put_options(now); + let sent_version_id = options.internal.source_version_id.clone(); + let headers = build_replication_probe_headers(&options); + + let initiate_headers = headers.clone(); + let initiate_version_id = sent_version_id.clone(); + let created = target_client + .client + .create_multipart_upload() + .bucket(target_bucket) + .key(probe_key) + .customize() + .map_request(move |mut req| { + for (key, value) in initiate_headers.clone() { + req.headers_mut().insert(key.expect("operation should succeed"), value); + } + let uri = append_version_id_query(req.uri(), &initiate_version_id); + req.set_uri(uri).map_err(std::io::Error::other)?; + Result::<_, std::io::Error>::Ok(req) + }) + .send() + .await + .map_err(S3ClientError::from) + .map_err(ReplicationProbeMultipartError::from)?; + let upload_id = created + .upload_id() + .ok_or_else(|| S3ClientError::new("target multipart initiate returned no upload id")) + .map_err(ReplicationProbeMultipartError::from)? + .to_string(); + + let uploaded = match target_client + .client + .upload_part() + .bucket(target_bucket) + .key(probe_key) + .upload_id(&upload_id) + .part_number(1) + .content_length(8) + .body(AwsByteStream::from_static(b"aaaaaaaa")) + .send() + .await + { + Ok(uploaded) => uploaded, + Err(error) => { + return Err(abort_failed_replication_probe_multipart( + target_client, + target_bucket, + probe_key, + &upload_id, + S3ClientError::from(error), + ) + .await); + } + }; + + let completed_part = CompletedPart::builder() + .part_number(1) + .set_e_tag(uploaded.e_tag().map(ToOwned::to_owned)) + .build(); + let complete_headers = headers.clone(); + let completed = match target_client + .client + .complete_multipart_upload() + .bucket(target_bucket) + .key(probe_key) + .upload_id(&upload_id) + .multipart_upload( + CompletedMultipartUpload::builder() + .set_parts(Some(vec![completed_part])) + .build(), + ) + .customize() + .map_request(move |mut req| { + for (key, value) in complete_headers.clone() { + req.headers_mut().insert(key.expect("operation should succeed"), value); + } + Result::<_, std::io::Error>::Ok(req) + }) + .send() + .await + { + Ok(completed) => completed, + Err(error) => { + return Err(abort_failed_replication_probe_multipart( + target_client, + target_bucket, + probe_key, + &upload_id, + S3ClientError::from(error), + ) + .await); + } + }; + + Ok(ReplicationProbePutOutcome { + sent_version_id, + response_version_id: completed.version_id().map(ToOwned::to_owned), + }) +} + +async fn abort_failed_replication_probe_multipart( + target_client: &TargetClient, + target_bucket: &str, + probe_key: &str, + upload_id: &str, + primary_error: S3ClientError, +) -> ReplicationProbeMultipartError { + match target_client + .client + .abort_multipart_upload() + .bucket(target_bucket) + .key(probe_key) + .upload_id(upload_id) + .send() + .await + { + Ok(_) => ReplicationProbeMultipartError::from(primary_error), + Err(error) => { + let abort_error = S3ClientError::from(error); + if abort_error.code.as_deref() == Some("NoSuchUpload") { + ReplicationProbeMultipartError::from(primary_error) + } else { + ReplicationProbeMultipartError { + primary: primary_error, + cleanup_error: Some("failed to abort multipart replication probe".to_string()), + } + } + } + } +} + +async fn put_replication_probe_object( + target_client: &TargetClient, + target_bucket: &str, + probe_key: &str, + now: OffsetDateTime, +) -> Result { + let options = build_replication_probe_put_options(now); + let sent_version_id = options.internal.source_version_id.clone(); + let headers = build_replication_probe_headers(&options); + + // Carry the source version as `?versionId=` exactly like a live + // replication PUT (P0-5 shape): the probe must exercise the query the + // real data path relies on, and the response tells us whether the target + // adopts the id. The probe id is always a fresh non-nil UUID, so the + // null-version mapping in the live path does not apply here. + let query_version_id = sent_version_id.clone(); + let response = target_client .client .put_object() .bucket(target_bucket) @@ -2259,12 +2526,18 @@ async fn put_replication_probe_object( for (key, value) in headers.clone() { req.headers_mut().insert(key.expect("operation should succeed"), value); } + let uri = append_version_id_query(req.uri(), &query_version_id); + req.set_uri(uri).map_err(std::io::Error::other)?; Result::<_, std::io::Error>::Ok(req) }) .send() .await - .map(|output| output.version_id().map(ToOwned::to_owned)) - .map_err(S3ClientError::from) + .map_err(S3ClientError::from)?; + + Ok(ReplicationProbePutOutcome { + sent_version_id, + response_version_id: response.version_id().map(ToOwned::to_owned), + }) } async fn delete_replication_probe_object( @@ -3431,6 +3704,12 @@ mod tests { #[derive(Default)] struct ScriptedReplicationProbe { put_error: Option<&'static str>, + /// Version id the scripted target answers with on PUT; None models a + /// mirroring target that echoes the sent source version id. + minted_version_id: Option<&'static str>, + /// Same, for the multipart leg: a target may mirror PutObject ids and + /// still mint its own at CreateMultipartUpload. + minted_multipart_version_id: Option<&'static str>, delete_marker_error: Option<&'static str>, version_delete_error: Option<&'static str>, cleanup_error: Option<&'static str>, @@ -3449,14 +3728,25 @@ mod tests { #[async_trait::async_trait] impl ReplicationProbeOperations for ScriptedReplicationProbe { - async fn put(&mut self) -> Result, S3ClientError> { + async fn put(&mut self) -> Result { self.calls.push("put"); match self.put_error { Some(code) => Err(scripted_probe_error(code)), - None => Ok(Some("object-version".to_string())), + None => Ok(ReplicationProbePutOutcome { + sent_version_id: "object-version".to_string(), + response_version_id: Some(self.minted_version_id.unwrap_or("object-version").to_string()), + }), } } + async fn multipart_put(&mut self) -> Result { + self.calls.push("multipart-put"); + Ok(ReplicationProbePutOutcome { + sent_version_id: "multipart-version".to_string(), + response_version_id: Some(self.minted_multipart_version_id.unwrap_or("multipart-version").to_string()), + }) + } + async fn create_delete_marker(&mut self, _version_id: Option<&str>) -> Result, S3ClientError> { self.calls.push("delete-marker"); match self.delete_marker_error { @@ -3473,7 +3763,7 @@ mod tests { } } - async fn cleanup(&mut self, known_version_ids: [Option<&str>; 2]) -> Result<(), String> { + async fn cleanup(&mut self, known_version_ids: [Option<&str>; 3]) -> Result<(), String> { self.calls.push("cleanup"); self.cleanup_ids = known_version_ids .into_iter() @@ -3486,6 +3776,44 @@ mod tests { } } + /// P1-19: a target that mints its own version ids must fail the + /// VersionFidelity phase with the machine-readable mismatch code, skip + /// the version-addressed mutation phases (they cannot mean anything on a + /// drifting target), and still clean up using the id the target actually + /// assigned — the source-derived id would never match. + #[tokio::test] + async fn replication_probe_flags_version_minting_target() { + let mut result = replication_check_target("arn:a", "OK", None); + let mut operations = ScriptedReplicationProbe { + minted_version_id: Some("target-minted-version"), + ..Default::default() + }; + + execute_replication_probe(&mut result, &mut operations).await; + + assert_eq!(operations.calls, ["put", "cleanup"]); + assert_eq!(result.status, "FAILED"); + assert_eq!(result.phases.put.status, "OK"); + assert_eq!(result.phases.version_fidelity.status, "FAILED"); + assert_eq!(result.phases.version_fidelity.code, Some(REPLICATION_CHECK_CODE_VERSION_MISMATCH)); + assert_eq!(result.phases.delete_marker.status, "SKIPPED"); + assert_eq!(result.phases.version_delete.status, "SKIPPED"); + assert_eq!(result.phases.cleanup.status, "OK"); + assert_eq!(operations.cleanup_ids, [Some("target-minted-version".to_string()), None, None]); + } + + #[tokio::test] + async fn replication_probe_passes_version_fidelity_for_mirroring_target() { + let mut result = replication_check_target("arn:a", "OK", None); + let mut operations = ScriptedReplicationProbe::default(); + + execute_replication_probe(&mut result, &mut operations).await; + + assert_eq!(result.status, "OK"); + assert_eq!(result.phases.version_fidelity.status, "OK"); + assert_eq!(result.phases.version_fidelity.code, None); + } + #[tokio::test] async fn replication_probe_attempts_cleanup_after_ambiguous_put_failure() { let mut result = replication_check_target("arn:a", "OK", None); @@ -3514,8 +3842,15 @@ mod tests { execute_replication_probe(&mut result, &mut operations).await; - assert_eq!(operations.calls, ["put", "delete-marker", "version-delete", "cleanup"]); - assert_eq!(operations.cleanup_ids, [Some("object-version".to_string()), None]); + assert_eq!(operations.calls, ["put", "multipart-put", "delete-marker", "version-delete", "cleanup"]); + assert_eq!( + operations.cleanup_ids, + [ + Some("object-version".to_string()), + Some("multipart-version".to_string()), + None + ] + ); assert_eq!(result.phases.delete_marker.status, "FAILED"); assert_eq!(result.phases.version_delete.status, "OK"); assert_eq!(result.phases.cleanup.status, "OK"); @@ -3532,10 +3867,14 @@ mod tests { execute_replication_probe(&mut result, &mut operations).await; - assert_eq!(operations.calls, ["put", "delete-marker", "version-delete", "cleanup"]); + assert_eq!(operations.calls, ["put", "multipart-put", "delete-marker", "version-delete", "cleanup"]); assert_eq!( operations.cleanup_ids, - [Some("object-version".to_string()), Some("marker-version".to_string())] + [ + Some("object-version".to_string()), + Some("multipart-version".to_string()), + Some("marker-version".to_string()) + ] ); assert_eq!(result.phases.version_delete.status, "FAILED"); assert_eq!(result.phases.cleanup.status, "FAILED"); diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index bf82ff20c..5dbc7f58d 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -183,6 +183,7 @@ pub(crate) mod bandwidth { } pub(crate) mod bucket_target_sys { + pub(crate) use super::ecstore_bucket::bucket_target_sys::append_version_id_query; pub(crate) type AdvancedPutOptions = super::ecstore_bucket::bucket_target_sys::AdvancedPutOptions; pub(crate) type BucketTargetError = super::ecstore_bucket::bucket_target_sys::BucketTargetError; pub(crate) type BucketTargetSys = super::ecstore_bucket::bucket_target_sys::BucketTargetSys;