diff --git a/.config/nextest.toml b/.config/nextest.toml index 3073dae55..fdb877939 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 + 49 nightly = 69 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` (49 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/config/README.md b/crates/config/README.md index 92141d4e8..0c703b57c 100644 --- a/crates/config/README.md +++ b/crates/config/README.md @@ -97,6 +97,14 @@ Current guidance: - enables minimal payload mode for GET health responses (`status`, `ready` only). - `RUSTFS_HEALTH_READINESS_CACHE_TTL_MS` - TTL for readiness cache evaluation. +- `RUSTFS_HEALTH_OBJECT_PROGRESS_ENABLE` + - withdraws readiness when bounded object read/write stages stop completing while requests remain active. + - default is `true`. +- `RUSTFS_HEALTH_OBJECT_PROGRESS_TIMEOUT_MS` + - maximum time without completion in a bounded object stage before readiness is withdrawn. + - default is `30000`; `0` uses the default. + - the effective value is at least 5 seconds longer than `RUSTFS_OBJECT_LOCK_ACQUIRE_TIMEOUT`. + - this readiness SLO is independent of disk read/write failure deadlines and may withdraw traffic before those deadlines expire. - `RUSTFS_HEALTH_COMPAT_BUSY_CHECK_ENABLE` - enables busy protection behavior for health probes. - default is `false`. diff --git a/crates/config/src/constants/health.rs b/crates/config/src/constants/health.rs index 1a83fe326..89be0c036 100644 --- a/crates/config/src/constants/health.rs +++ b/crates/config/src/constants/health.rs @@ -22,6 +22,19 @@ pub const DEFAULT_HEALTH_ENDPOINT_ENABLE: bool = true; pub const ENV_HEALTH_READINESS_CACHE_TTL_MS: &str = "RUSTFS_HEALTH_READINESS_CACHE_TTL_MS"; pub const DEFAULT_HEALTH_READINESS_CACHE_TTL_MS: u64 = 1000; +/// Enable readiness withdrawal when bounded object read/write stages stop +/// completing while requests remain active. +pub const ENV_HEALTH_OBJECT_PROGRESS_ENABLE: &str = "RUSTFS_HEALTH_OBJECT_PROGRESS_ENABLE"; +pub const DEFAULT_HEALTH_OBJECT_PROGRESS_ENABLE: bool = true; + +/// Requested time without completion in a bounded object stage before local +/// readiness is withdrawn (milliseconds). A value of `0` uses the default; +/// runtime adds a safety floor based on the object-lock acquisition timeout. +pub const ENV_HEALTH_OBJECT_PROGRESS_TIMEOUT_MS: &str = "RUSTFS_HEALTH_OBJECT_PROGRESS_TIMEOUT_MS"; +pub const DEFAULT_HEALTH_OBJECT_PROGRESS_TIMEOUT_MS: u64 = 30_000; +/// Additional time beyond the configured object-lock acquisition deadline. +pub const HEALTH_OBJECT_PROGRESS_LOCK_MARGIN_MS: u64 = 5_000; + /// Timeout for cluster health readiness collectors (milliseconds). /// This bounds expensive storage and lock quorum checks used by cluster probes. pub const ENV_HEALTH_CLUSTER_TIMEOUT_MS: &str = "RUSTFS_HEALTH_CLUSTER_TIMEOUT_MS"; 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..2734295f3 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/crates/e2e_test/src/replication_extension_test.rs @@ -629,6 +629,17 @@ async fn put_bucket_replication_with_delete_statuses( target_arn: &str, delete_marker_status: &str, version_delete_status: Option<&str>, +) -> Result<(), Box> { + put_bucket_replication_with_statuses(env, bucket, target_arn, delete_marker_status, version_delete_status, "Enabled").await +} + +async fn put_bucket_replication_with_statuses( + env: &RustFSTestEnvironment, + bucket: &str, + target_arn: &str, + delete_marker_status: &str, + version_delete_status: Option<&str>, + existing_object_status: &str, ) -> Result<(), Box> { let delete_replication = version_delete_status .map(|status| format!("{status}")) @@ -645,7 +656,7 @@ async fn put_bucket_replication_with_delete_statuses( {delete_replication} - Enabled + {existing_object_status} {target_arn} @@ -2595,6 +2606,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 +7503,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 @@ -7818,5 +8160,239 @@ async fn test_delayed_delete_marker_purge_exhaustion_persists_to_mrf_and_replays ); target.shutdown().await; + + Ok(()) +} + +// --- P1-20 (backlog#1675): scanner existing-object compensation matrix --- +// +// Every case below inverts the order used by the rest of this file: objects +// are written FIRST and the replication rule arrives afterwards, so the only +// channel that can move the pre-existing objects is the data scanner's +// existing-object resync pass. Negative cells ("never compensated") are +// contracts and are asserted over multiple scanner cycles, always next to a +// replicated control key that proves the scanner and the live path are +// running — an absent key on a dead scanner proves nothing. + +/// Envs + buckets only: versioning, the remote target, and the rule variant +/// are wired by each test (the null-version case must PUT before the source +/// bucket becomes versioned). The source runs with FAST_SCANNER_ENV so +/// existing keys are rescanned within seconds instead of 16 dir cycles. +async fn build_scanner_compensation_pair( + source_bucket: &str, + target_bucket: &str, +) -> Result<(RustFSTestEnvironment, RustFSTestEnvironment), Box> { + let mut source_env = RustFSTestEnvironment::new().await?; + let mut source_process_env = replication_fast_env(); + source_process_env.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV); + source_process_env.extend_from_slice(FAST_SCANNER_ENV); + source_env.start_rustfs_server_with_env(vec![], &source_process_env).await?; + + let mut target_env = RustFSTestEnvironment::new().await?; + target_env.start_rustfs_server_without_cleanup(vec![]).await?; + + source_env + .create_s3_client() + .create_bucket() + .bucket(source_bucket) + .send() + .await?; + target_env + .create_s3_client() + .create_bucket() + .bucket(target_bucket) + .send() + .await?; + + Ok((source_env, target_env)) +} + +/// P1-20: objects that already exist when a rule with +/// ExistingObjectReplication=Enabled arrives are compensated by the scanner's +/// existing-object resync pass, whatever wrote them — plain PUT, CopyObject, +/// or Snowball auto-extract. The pinned exception is a null-version object +/// (written before the bucket became versioned): the scanner heal gate skips +/// nil-version objects entirely (`scanner_folder.rs` heal_replication), so it +/// must NEVER be compensated. +#[tokio::test] +#[serial] +async fn test_scanner_compensates_existing_objects_across_write_paths() -> TestResult { + init_logging(); + let source_bucket = "scanner-comp-src"; + let target_bucket = "scanner-comp-dst"; + let (source_env, target_env) = build_scanner_compensation_pair(source_bucket, target_bucket).await?; + let source_client = source_env.create_s3_client(); + let target_client = target_env.create_s3_client(); + + // Null-version cell: PUT before versioning; the object keeps the nil + // version id forever. + let null_key = "pre-versioning-null.txt"; + source_client + .put_object() + .bucket(source_bucket) + .key(null_key) + .body(ByteStream::from_static(b"null version payload")) + .send() + .await?; + + enable_bucket_versioning(&source_env, source_bucket).await?; + enable_bucket_versioning(&target_env, target_bucket).await?; + + // Pre-existing objects from three write paths, all before any replication + // config exists (their replication status stays Empty). + let plain_key = "existing-plain.txt"; + let plain_payload = "existing plain payload"; + source_client + .put_object() + .bucket(source_bucket) + .key(plain_key) + .body(ByteStream::from_static(plain_payload.as_bytes())) + .send() + .await?; + + let copy_key = "existing-copy.txt"; + source_client + .copy_object() + .bucket(source_bucket) + .key(copy_key) + .copy_source(format!("{source_bucket}/{plain_key}")) + .send() + .await?; + + let member_key = "snowball/existing-member.txt"; + let member_payload: &[u8] = b"existing snowball member payload"; + let mut builder = tokio_tar::Builder::new(std::io::Cursor::new(Vec::new())); + let mut header = tokio_tar::Header::new_gnu(); + header.set_size(member_payload.len() as u64); + header.set_mode(0o644); + header.set_cksum(); + builder + .append_data(&mut header, member_key, std::io::Cursor::new(member_payload)) + .await?; + let archive = builder.into_inner().await?.into_inner(); + source_client + .put_object() + .bucket(source_bucket) + .key("existing-members.tar") + .metadata("Snowball-Auto-Extract", "true") + .body(ByteStream::from(archive)) + .send() + .await?; + // The extracted member must exist locally before the rule arrives, or it + // would replicate through the live path instead of the scanner. + let deadline = tokio::time::Instant::now() + Duration::from_secs(10); + loop { + if source_client + .head_object() + .bucket(source_bucket) + .key(member_key) + .send() + .await + .is_ok() + { + break; + } + if tokio::time::Instant::now() >= deadline { + return Err("snowball member was never extracted on the source".into()); + } + sleep(Duration::from_millis(200)).await; + } + + // Only now wire the remote target and the Enabled rule. + let target_arn = set_replication_target(&source_env, source_bucket, &target_env, target_bucket).await?; + put_bucket_replication(&source_env, source_bucket, &target_arn).await?; + + // Control key written after the rule replicates through the live path. + let control_key = "control-live.txt"; + let control_payload = "control live payload"; + source_client + .put_object() + .bucket(source_bucket) + .key(control_key) + .body(ByteStream::from_static(control_payload.as_bytes())) + .send() + .await?; + wait_for_replicated_object(&target_client, target_bucket, control_key, control_payload).await?; + + // Scanner compensation for each pre-existing write path. + wait_for_replicated_object(&target_client, target_bucket, plain_key, plain_payload).await?; + wait_for_replicated_object(&target_client, target_bucket, copy_key, plain_payload).await?; + wait_for_replicated_object(&target_client, target_bucket, member_key, std::str::from_utf8(member_payload)?).await?; + + // Null-version contract: with every sibling compensated (scanner proven + // live), the nil-version object must stay absent across further cycles. + assert_replication_key_absent(&target_client, target_bucket, null_key, Duration::from_secs(6)).await?; + + Ok(()) +} + +/// P1-20: ExistingObjectReplication=Disabled is a contract, not a delay — the +/// scanner must NEVER compensate objects that predate the rule, while objects +/// written after the rule replicate normally (the setting only gates the +/// existing-object resync path). +#[tokio::test] +#[serial] +async fn test_scanner_never_compensates_when_existing_object_replication_disabled() -> TestResult { + init_logging(); + let source_bucket = "scanner-disabled-src"; + let target_bucket = "scanner-disabled-dst"; + let (source_env, mut target_env) = build_scanner_compensation_pair(source_bucket, target_bucket).await?; + let source_client = source_env.create_s3_client(); + let target_client = target_env.create_s3_client(); + + enable_bucket_versioning(&source_env, source_bucket).await?; + enable_bucket_versioning(&target_env, target_bucket).await?; + + let existing_key = "existing-disabled.txt"; + source_client + .put_object() + .bucket(source_bucket) + .key(existing_key) + .body(ByteStream::from_static(b"existing disabled payload")) + .send() + .await?; + + let target_arn = set_replication_target(&source_env, source_bucket, &target_env, target_bucket).await?; + put_bucket_replication_with_statuses(&source_env, source_bucket, &target_arn, "Enabled", None, "Disabled").await?; + + // The live path is unaffected by the Disabled existing-object setting. + let control_key = "control-live.txt"; + let control_payload = "control live payload"; + source_client + .put_object() + .bucket(source_bucket) + .key(control_key) + .body(ByteStream::from_static(control_payload.as_bytes())) + .send() + .await?; + wait_for_replicated_object(&target_client, target_bucket, control_key, control_payload).await?; + + // Scanner-only witness. A live-path control key alone would let this test + // pass while the existing-object scanner is disabled or wedged, so make + // the scanner itself observable: an object whose replication FAILED while + // the target was down can only be re-driven by the data scanner's + // replication heal pass (see FAST_SCANNER_ENV), and that pass is NOT + // gated by ExistingObjectReplication. The witness lives in the same + // bucket and prefix as the pre-existing key, so a heal pass that reached + // it necessarily walked the pre-existing key in the same scan. + let witness_key = "scanner-witness.txt"; + let witness_payload = "scanner witness payload"; + target_env.stop_server(); + source_client + .put_object() + .bucket(source_bucket) + .key(witness_key) + .body(ByteStream::from_static(witness_payload.as_bytes())) + .send() + .await?; + wait_for_source_replication_status(&source_client, source_bucket, witness_key, "FAILED", false).await?; + target_env.restart_server_preserving_data(vec![], &[]).await?; + let target_client = target_env.create_s3_client(); + wait_for_replicated_object(&target_client, target_bucket, witness_key, witness_payload).await?; + + // The scanner demonstrably swept this bucket; the pre-existing key must + // still be absent, and stay absent over further cycles. + assert_replication_key_absent(&target_client, target_bucket, existing_key, Duration::from_secs(6)).await?; + Ok(()) } diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 9a63c7cd9..d376cb0b7 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, }; } @@ -281,7 +281,7 @@ pub mod config { pub mod com { pub use crate::config::com::{ COMMA_SEPARATED_LISTS, CONFIG_PREFIX, ENV_CONFIG_RECOVER_ON_CORRUPTION, STORAGE_CLASS_SUB_SYS, - ServerConfigCorruptError, ServerConfigSaveResult, ServerConfigSnapshot, delete_config, + ServerConfigCorruptError, ServerConfigSaveResult, ServerConfigSnapshot, delete_config, delete_config_no_lock, is_server_config_corrupt_error, lookup_configs, read_config, read_config_no_lock, read_config_with_metadata, read_config_without_migrate, read_config_without_migrate_no_lock, read_existing_server_config_no_lock, read_server_config_snapshot, save_config, save_config_no_lock, save_config_with_opts, save_server_config, 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/crates/ecstore/src/cluster/rpc/client.rs b/crates/ecstore/src/cluster/rpc/client.rs index 59082b659..0e1395fff 100644 --- a/crates/ecstore/src/cluster/rpc/client.rs +++ b/crates/ecstore/src/cluster/rpc/client.rs @@ -15,12 +15,12 @@ #[cfg(test)] use crate::cluster::rpc::http_auth::RPC_REPLAY_SCOPE_VERSION_HEADER; use crate::cluster::rpc::http_auth::{ - RPC_AUTH_VERSION_HEADER, RPC_AUTH_VERSION_V2, RPC_BOOT_EPOCH_CHALLENGE_HEADER, RPC_BOOT_EPOCH_HEADER, - RPC_BOOT_EPOCH_PROOF_HEADER, RPC_CONTENT_SHA256_HEADER, TIMESTAMP_HEADER, -}; -use crate::cluster::rpc::{ - gen_tonic_replay_scope_headers, gen_tonic_signature_headers, normalize_tonic_rpc_audience, verify_tonic_boot_epoch_response, + AuthenticatedPeerReplayCapabilities, RPC_AUTH_VERSION_HEADER, RPC_AUTH_VERSION_V2, RPC_BOOT_EPOCH_CHALLENGE_HEADER, + RPC_CONTENT_SHA256_HEADER, RPC_REPLAY_CACHE_CAPABILITY_HEADER, RPC_REPLAY_CACHE_CAPABILITY_PROOF_HEADER, + RollingMutationBodyDigest, TIMESTAMP_HEADER, internode_rpc_body_digest_strict, + verify_tonic_peer_replay_capabilities_response, }; +use crate::cluster::rpc::{gen_tonic_replay_scope_headers, gen_tonic_signature_headers, normalize_tonic_rpc_audience}; #[cfg(test)] use crate::cluster::rpc::{tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers}; use crate::disk::error::{DiskError, Error as DiskErrorType, RpcStatusError}; @@ -233,7 +233,22 @@ pub struct ReplayScopeChannel { /// The channel type used by internode clients after v2 authentication and replay-scope handling. pub type AuthenticatedChannel = ReplayScopeChannel; -static PEER_BOOT_EPOCHS: LazyLock>> = LazyLock::new(|| Mutex::new(HashMap::new())); +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum PeerReplayCapability { + Capable { boot_epoch: Uuid }, + Revoked, +} + +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +struct PeerReplayState { + boot_epoch: Option, + cache_capability: Option, +} + +#[derive(Clone, Copy, Debug)] +struct PeerReplayStateSnapshot(PeerReplayState); + +static PEER_REPLAY_STATES: LazyLock>> = LazyLock::new(|| Mutex::new(HashMap::new())); impl ReplayScopeChannel { fn new(inner: S, audience: Option) -> Self { @@ -241,13 +256,67 @@ impl ReplayScopeChannel { } } -fn cached_peer_boot_epoch(audience: &str) -> Option { - PEER_BOOT_EPOCHS.lock().ok().and_then(|epochs| epochs.get(audience).copied()) +fn peer_replay_state(audience: &str) -> PeerReplayState { + PEER_REPLAY_STATES + .lock() + .ok() + .and_then(|states| states.get(audience).copied()) + .unwrap_or_default() } -fn remember_peer_boot_epoch(audience: String, epoch: Uuid) { - if let Ok(mut epochs) = PEER_BOOT_EPOCHS.lock() { - epochs.insert(audience, epoch); +fn apply_peer_replay_response( + audience: String, + sent_state: PeerReplayState, + response: std::io::Result, +) { + if let Ok(mut states) = PEER_REPLAY_STATES.lock() { + let current_state = states.get(&audience).copied().unwrap_or_default(); + let mut next_state = current_state; + if let Ok(response) = &response + && sent_state.boot_epoch == current_state.boot_epoch + { + next_state.boot_epoch = Some(response.boot_epoch); + } + + if sent_state.boot_epoch == current_state.boot_epoch { + let response_capability = response + .as_ref() + .ok() + .filter(|response| response.dynamic_replay_cache) + .map(|response| response.boot_epoch); + match (sent_state.cache_capability, current_state.cache_capability, response_capability) { + (None, None, Some(boot_epoch)) + | (Some(PeerReplayCapability::Revoked), Some(PeerReplayCapability::Revoked), Some(boot_epoch)) => { + next_state.cache_capability = Some(PeerReplayCapability::Capable { boot_epoch }); + } + ( + Some(PeerReplayCapability::Capable { + boot_epoch: sent_boot_epoch, + }), + Some(PeerReplayCapability::Capable { + boot_epoch: current_boot_epoch, + }), + Some(response_boot_epoch), + ) if sent_boot_epoch == current_boot_epoch => { + next_state.cache_capability = Some(PeerReplayCapability::Capable { + boot_epoch: response_boot_epoch, + }); + } + ( + Some(PeerReplayCapability::Capable { + boot_epoch: sent_boot_epoch, + }), + Some(PeerReplayCapability::Capable { + boot_epoch: current_boot_epoch, + }), + None, + ) if sent_boot_epoch == current_boot_epoch => { + next_state.cache_capability = Some(PeerReplayCapability::Revoked); + } + _ => {} + } + } + states.insert(audience, next_state); } } @@ -276,6 +345,11 @@ where == Some(RPC_AUTH_VERSION_V2) }); let challenge = authenticated.then(Uuid::new_v4); + let sent_state = request + .extensions() + .get::() + .map(|snapshot| snapshot.0) + .unwrap_or_default(); if let (Some(audience), Some(challenge)) = (self.audience.as_deref(), challenge) { // The challenge is independently HMAC-authenticated by the response proof. It is not // part of v2 so old peers ignore it, while a new peer can safely advertise its epoch. @@ -284,7 +358,7 @@ where challenge.to_string().parse().expect("UUID must be a valid header value"), ); if let (Some(boot_epoch), Some(timestamp), Some(content_sha256)) = ( - cached_peer_boot_epoch(audience), + sent_state.boot_epoch, request.headers().get(TIMESTAMP_HEADER).and_then(|value| value.to_str().ok()), request .headers() @@ -303,16 +377,21 @@ where Box::pin(async move { let response = future.await?; if let (Some(audience), Some(challenge)) = (audience, challenge) { - match verify_tonic_boot_epoch_response(&audience, challenge, response.headers()) { - Ok(epoch) => remember_peer_boot_epoch(audience, epoch), - Err(error) - if response.headers().contains_key(RPC_BOOT_EPOCH_HEADER) - || response.headers().contains_key(RPC_BOOT_EPOCH_PROOF_HEADER) => - { - debug!(error = %error, "peer boot epoch response proof was rejected") - } - Err(_) => {} + let response_state = verify_tonic_peer_replay_capabilities_response(&audience, challenge, response.headers()); + if let Err(error) = &response_state + && (response.headers().contains_key(RPC_REPLAY_CACHE_CAPABILITY_HEADER) + || response.headers().contains_key(RPC_REPLAY_CACHE_CAPABILITY_PROOF_HEADER)) + { + debug!( + event = "internode_rpc_capability_proof_rejected", + component = "ecstore", + subsystem = "rpc_client", + result = "rejected", + error = %error, + "internode RPC capability proof rejected" + ) } + apply_peer_replay_response(audience, sent_state, response_state); } Ok(response) }) @@ -321,6 +400,7 @@ where pub struct TonicSignatureInterceptor { audience: Option, + body_digest_strict: bool, } impl tonic::service::Interceptor for TonicSignatureInterceptor { @@ -337,9 +417,31 @@ impl tonic::service::Interceptor for TonicSignatureInterceptor { .metadata() .get(RPC_CONTENT_SHA256_HEADER) .and_then(|value| value.to_str().ok()); + // RUSTFS_COMPAT_TODO(disk-mutation-body-digest): use cache-free v2 for peers without an authenticated boot epoch. Remove after every supported peer advertises the authenticated dynamic replay-cache capability and body-digest strict mode is the default. + // beta.11 verifies v2 body digests but stores their nonces in a fixed-size cache. + let rolling_mutation = req.extensions().get::().is_some(); + let peer_state = PEER_REPLAY_STATES + .lock() + .map_err(|_| tonic::Status::unauthenticated("RPC peer capability state unavailable"))? + .get(audience) + .copied() + .unwrap_or_default(); + let content_sha256 = if content_sha256.is_some() { + if peer_state.cache_capability == Some(PeerReplayCapability::Revoked) { + return Err(tonic::Status::unauthenticated("RPC peer replay capability changed")); + } + if rolling_mutation && !self.body_digest_strict && peer_state.boot_epoch.is_none() { + None + } else { + content_sha256 + } + } else { + content_sha256 + }; let headers = gen_tonic_signature_headers(audience, method.service(), method.method(), content_sha256) .map_err(|_| tonic::Status::unauthenticated("No valid auth token"))?; req.metadata_mut().as_mut().extend(headers); + req.extensions_mut().insert(PeerReplayStateSnapshot(peer_state)); inject_trace_context_into_metadata(req.metadata_mut()); inject_request_id_into_metadata(req.metadata_mut()); Ok(req) @@ -347,7 +449,10 @@ impl tonic::service::Interceptor for TonicSignatureInterceptor { } pub fn gen_tonic_signature_interceptor() -> TonicSignatureInterceptor { - TonicSignatureInterceptor { audience: None } + TonicSignatureInterceptor { + audience: None, + body_digest_strict: internode_rpc_body_digest_strict(), + } } pub struct NoOpInterceptor; @@ -409,6 +514,7 @@ mod tests { #[derive(Clone)] struct EpochProofService { audience: String, + include_capability: bool, seen_headers: std::sync::Arc>>, } @@ -430,29 +536,97 @@ mod tests { .expect("client challenge must be syntactically valid") .expect("authenticated client request must carry a boot epoch challenge"); let mut response = HttpResponse::new(()); - response.headers_mut().extend( - tonic_boot_epoch_response_headers(&self.audience, challenge) - .expect("test server must be able to sign an epoch proof"), - ); + let mut headers = tonic_boot_epoch_response_headers(&self.audience, challenge) + .expect("test server must be able to sign an epoch proof"); + if !self.include_capability { + headers.remove(RPC_REPLAY_CACHE_CAPABILITY_HEADER); + headers.remove(RPC_REPLAY_CACHE_CAPABILITY_PROOF_HEADER); + } + response.headers_mut().extend(headers); std::future::ready(Ok(response)) } } + #[derive(Clone)] + struct MissingProofService; + + impl Service> for MissingProofService { + type Response = HttpResponse<()>; + type Error = std::convert::Infallible; + type Future = std::future::Ready>; + + fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn call(&mut self, _request: HttpRequest<()>) -> Self::Future { + std::future::ready(Ok(HttpResponse::new(()))) + } + } + fn ensure_test_rpc_secret() { runtime_sources::ensure_test_rpc_secret(); } fn test_request() -> tonic::Request<()> { + test_request_for("Ping") + } + + fn test_request_for(method: &'static str) -> tonic::Request<()> { let mut request = tonic::Request::new(()); request .extensions_mut() - .insert(tonic::GrpcMethod::new("node_service.NodeService", "Ping")); + .insert(tonic::GrpcMethod::new("node_service.NodeService", method)); request } fn test_interceptor() -> TonicSignatureInterceptor { + test_interceptor_for("node-a:9000", false) + } + + fn test_interceptor_for(audience: &str, body_digest_strict: bool) -> TonicSignatureInterceptor { TonicSignatureInterceptor { - audience: Some("node-a:9000".to_string()), + audience: Some(audience.to_string()), + body_digest_strict, + } + } + + fn clear_peer_capability(audience: &str) { + PEER_REPLAY_STATES + .lock() + .expect("peer capability cache lock must not be poisoned") + .remove(audience); + } + + fn rolling_mutation_request(method: &'static str) -> tonic::Request<()> { + let mut request = tonic::Request::new(rustfs_protos::proto_gen::node_service::GenerallyLockRequest { + args: "canonical mutation request".to_string(), + }); + request + .extensions_mut() + .insert(tonic::GrpcMethod::new("node_service.NodeService", method)); + crate::cluster::rpc::set_tonic_rolling_mutation_body_digest(&mut request).expect("test mutation digest must be attached"); + request.map(|_| ()) + } + + fn replay_scope_request(audience: &str, method: &'static str) -> HttpRequest<()> { + let mut request = HttpRequest::builder() + .uri(format!("/node_service.NodeService/{method}")) + .body(()) + .expect("test RPC request must build"); + request.headers_mut().extend( + gen_tonic_signature_headers(audience, "node_service.NodeService", method, None).expect("v2 test headers must mint"), + ); + request + .extensions_mut() + .insert(PeerReplayStateSnapshot(peer_replay_state(audience))); + request + } + + fn authenticated_peer_response(boot_epoch: Uuid, dynamic_replay_cache: bool) -> AuthenticatedPeerReplayCapabilities { + AuthenticatedPeerReplayCapabilities { + boot_epoch, + dynamic_replay_cache, } } @@ -567,6 +741,431 @@ mod tests { ); } + #[test] + fn unknown_peer_mutations_use_cache_free_unsigned_v2() { + ensure_test_rpc_secret(); + let audience = "legacy-body-digest-client-test:9000"; + clear_peer_capability(audience); + let mut interceptor = test_interceptor_for(audience, false); + for method in ["Lock", "WriteAll"] { + let request = interceptor + .call(rolling_mutation_request(method)) + .expect("interceptor call should succeed"); + + assert_eq!( + request + .metadata() + .get("x-rustfs-content-sha256") + .and_then(|value| value.to_str().ok()), + Some("UNSIGNED-PAYLOAD") + ); + assert_eq!( + request + .metadata() + .get("x-rustfs-rpc-nonce") + .and_then(|value| value.to_str().ok()), + Some("unsigned") + ); + assert!( + crate::cluster::rpc::verify_tonic_rpc_signature( + audience, + &format!("/node_service.NodeService/{method}"), + request.metadata().as_ref(), + ) + .is_ok(), + "the cache-free request must retain valid audience- and method-bound v2 authentication" + ); + } + } + + #[test] + fn unknown_peer_exact_body_contract_remains_body_bound() { + ensure_test_rpc_secret(); + let audience = "exact-body-contract-client-test:9000"; + clear_peer_capability(audience); + let mut interceptor = test_interceptor_for(audience, false); + let mut request = test_request_for("ScannerActivity"); + crate::cluster::rpc::set_tonic_canonical_body_digest(&mut request, b"exact scanner activity body") + .expect("test exact body digest must be attached"); + let expected_digest = request + .metadata() + .get("x-rustfs-content-sha256") + .and_then(|value| value.to_str().ok()) + .expect("test request must carry its digest") + .to_string(); + + let request = interceptor.call(request).expect("interceptor call should succeed"); + + assert_eq!( + request + .metadata() + .get("x-rustfs-content-sha256") + .and_then(|value| value.to_str().ok()), + Some(expected_digest.as_str()) + ); + assert!( + crate::cluster::rpc::verify_tonic_rpc_signature( + audience, + "/node_service.NodeService/ScannerActivity", + request.metadata().as_ref(), + ) + .is_ok() + ); + } + + #[test] + fn unknown_peer_iam_mutation_helper_remains_body_bound() { + ensure_test_rpc_secret(); + let audience = "exact-iam-mutation-client-test:9000"; + clear_peer_capability(audience); + let mut interceptor = test_interceptor_for(audience, false); + let mut request = tonic::Request::new(rustfs_protos::proto_gen::node_service::DeleteUserRequest { + access_key: "target-access-key".to_string(), + }); + request + .extensions_mut() + .insert(tonic::GrpcMethod::new("node_service.NodeService", "DeleteUser")); + crate::cluster::rpc::set_tonic_mutation_body_digest(&mut request).expect("test IAM mutation digest must be attached"); + let request = request.map(|_| ()); + let expected_digest = request + .metadata() + .get("x-rustfs-content-sha256") + .and_then(|value| value.to_str().ok()) + .expect("test IAM mutation must carry its digest") + .to_string(); + + let request = interceptor.call(request).expect("interceptor call should succeed"); + + assert_eq!( + request + .metadata() + .get("x-rustfs-content-sha256") + .and_then(|value| value.to_str().ok()), + Some(expected_digest.as_str()) + ); + assert!( + crate::cluster::rpc::verify_tonic_rpc_signature( + audience, + "/node_service.NodeService/DeleteUser", + request.metadata().as_ref(), + ) + .is_ok(), + "IAM mutations must remain body-bound before capability discovery" + ); + } + + #[test] + fn authenticated_replay_cache_capability_enables_body_binding() { + ensure_test_rpc_secret(); + let audience = "body-digest-capable-client-test:9000"; + clear_peer_capability(audience); + let seen_headers = std::sync::Arc::new(Mutex::new(Vec::new())); + let service = EpochProofService { + audience: audience.to_string(), + include_capability: true, + seen_headers, + }; + let mut channel = ReplayScopeChannel::new(service, Some(audience.to_string())); + futures::executor::block_on(channel.call(replay_scope_request(audience, "Ping"))) + .expect("authenticated capability probe must complete"); + + let mut interceptor = test_interceptor_for(audience, false); + let request = rolling_mutation_request("Lock"); + let expected_digest = request + .metadata() + .get("x-rustfs-content-sha256") + .and_then(|value| value.to_str().ok()) + .expect("test mutation must carry its digest") + .to_string(); + + let request = interceptor.call(request).expect("interceptor call should succeed"); + + assert_eq!( + request + .metadata() + .get("x-rustfs-content-sha256") + .and_then(|value| value.to_str().ok()), + Some(expected_digest.as_str()) + ); + let nonce = request + .metadata() + .get("x-rustfs-rpc-nonce") + .and_then(|value| value.to_str().ok()) + .and_then(|value| Uuid::parse_str(value).ok()) + .expect("capable peer body-bound mutation must carry a UUID nonce"); + assert!(!nonce.is_nil()); + assert!( + crate::cluster::rpc::verify_tonic_rpc_signature( + audience, + "/node_service.NodeService/Lock", + request.metadata().as_ref(), + ) + .is_ok(), + "the body-bound request must retain valid audience- and method-bound v2 authentication" + ); + clear_peer_capability(audience); + } + + #[test] + fn invalid_capability_proof_does_not_enable_body_binding() { + ensure_test_rpc_secret(); + let audience = "invalid-capability-client-test:9000"; + clear_peer_capability(audience); + let service = EpochProofService { + audience: "wrong-capability-audience:9000".to_string(), + include_capability: true, + seen_headers: std::sync::Arc::new(Mutex::new(Vec::new())), + }; + let mut channel = ReplayScopeChannel::new(service, Some(audience.to_string())); + futures::executor::block_on(channel.call(replay_scope_request(audience, "Ping"))) + .expect("invalid capability response must still complete"); + + let mut interceptor = test_interceptor_for(audience, false); + let request = interceptor + .call(rolling_mutation_request("Lock")) + .expect("legacy-compatible mutation must still be signed"); + + assert_eq!( + request + .metadata() + .get("x-rustfs-content-sha256") + .and_then(|value| value.to_str().ok()), + Some("UNSIGNED-PAYLOAD") + ); + } + + #[test] + fn legacy_boot_proof_keeps_mutations_body_bound_and_enables_non_ping_v3() { + ensure_test_rpc_secret(); + let audience = "legacy-boot-proof-client-test:9000"; + clear_peer_capability(audience); + let seen_headers = std::sync::Arc::new(Mutex::new(Vec::new())); + let service = EpochProofService { + audience: audience.to_string(), + include_capability: false, + seen_headers: seen_headers.clone(), + }; + let mut channel = ReplayScopeChannel::new(service, Some(audience.to_string())); + futures::executor::block_on(channel.call(replay_scope_request(audience, "Ping"))) + .expect("legacy boot proof response must complete"); + + let state = peer_replay_state(audience); + assert!(state.boot_epoch.is_some(), "authenticated legacy proof must enable replay-scoped v3"); + assert_eq!(state.cache_capability, None); + + let mut interceptor = test_interceptor_for(audience, false); + let request = rolling_mutation_request("Lock"); + let expected_digest = request + .metadata() + .get("x-rustfs-content-sha256") + .and_then(|value| value.to_str().ok()) + .expect("test mutation must carry its digest") + .to_string(); + let request = interceptor.call(request).expect("legacy-compatible mutation must be signed"); + assert_eq!( + request + .metadata() + .get("x-rustfs-content-sha256") + .and_then(|value| value.to_str().ok()), + Some(expected_digest.as_str()) + ); + let (metadata, extensions, body) = request.into_parts(); + let mut request = HttpRequest::new(body); + *request.uri_mut() = "/node_service.NodeService/Lock".parse().expect("test RPC URI must parse"); + *request.headers_mut() = metadata.into_headers(); + *request.extensions_mut() = extensions; + futures::executor::block_on(channel.call(request)).expect("legacy strict-compatible lock request must complete"); + + let headers = seen_headers.lock().expect("test header capture lock must not be poisoned"); + assert!( + headers[1].contains_key(RPC_REPLAY_SCOPE_VERSION_HEADER), + "authenticated legacy boot proof must enable v3 on a non-Ping request" + ); + } + + #[test] + fn reordered_capability_responses_cannot_undo_newer_state() { + let audience = "reordered-capability-client-test:9000"; + let epoch_one = Uuid::new_v4(); + let epoch_two = Uuid::new_v4(); + clear_peer_capability(audience); + + let unknown = PeerReplayState::default(); + apply_peer_replay_response(audience.to_string(), unknown, Ok(authenticated_peer_response(epoch_one, true))); + apply_peer_replay_response(audience.to_string(), unknown, Err(std::io::Error::other("delayed legacy response"))); + let epoch_one_state = PeerReplayState { + boot_epoch: Some(epoch_one), + cache_capability: Some(PeerReplayCapability::Capable { boot_epoch: epoch_one }), + }; + assert_eq!(peer_replay_state(audience), epoch_one_state); + + apply_peer_replay_response(audience.to_string(), epoch_one_state, Err(std::io::Error::other("rollback response"))); + apply_peer_replay_response(audience.to_string(), epoch_one_state, Ok(authenticated_peer_response(epoch_one, true))); + assert_eq!( + peer_replay_state(audience), + PeerReplayState { + boot_epoch: Some(epoch_one), + cache_capability: Some(PeerReplayCapability::Revoked), + } + ); + + let revoked = peer_replay_state(audience); + apply_peer_replay_response(audience.to_string(), revoked, Ok(authenticated_peer_response(epoch_two, true))); + apply_peer_replay_response(audience.to_string(), epoch_one_state, Ok(authenticated_peer_response(epoch_one, true))); + assert_eq!( + peer_replay_state(audience), + PeerReplayState { + boot_epoch: Some(epoch_two), + cache_capability: Some(PeerReplayCapability::Capable { boot_epoch: epoch_two }), + } + ); + clear_peer_capability(audience); + } + + #[test] + fn stale_capability_response_cannot_cross_a_new_boot_epoch() { + let audience = "cross-epoch-capability-client-test:9000"; + let epoch_one = Uuid::new_v4(); + let epoch_two = Uuid::new_v4(); + let epoch_three = Uuid::new_v4(); + clear_peer_capability(audience); + + let revoked_epoch_one = PeerReplayState { + boot_epoch: Some(epoch_one), + cache_capability: Some(PeerReplayCapability::Revoked), + }; + PEER_REPLAY_STATES + .lock() + .expect("peer replay state lock must not be poisoned") + .insert(audience.to_string(), revoked_epoch_one); + + apply_peer_replay_response( + audience.to_string(), + revoked_epoch_one, + Ok(authenticated_peer_response(epoch_three, false)), + ); + apply_peer_replay_response(audience.to_string(), revoked_epoch_one, Ok(authenticated_peer_response(epoch_two, true))); + + assert_eq!( + peer_replay_state(audience), + PeerReplayState { + boot_epoch: Some(epoch_three), + cache_capability: Some(PeerReplayCapability::Revoked), + }, + "a stale dynamic-cache proof must not cross a newer authenticated boot epoch" + ); + clear_peer_capability(audience); + } + + #[test] + fn interceptor_snapshot_prevents_delayed_legacy_response_from_revoking_capability() { + ensure_test_rpc_secret(); + let audience = "capability-snapshot-client-test:9000"; + clear_peer_capability(audience); + let boot_epoch = Uuid::new_v4(); + let mut interceptor = test_interceptor_for(audience, false); + let request = interceptor + .call(rolling_mutation_request("Lock")) + .expect("legacy-compatible request must pass the interceptor"); + assert_eq!( + request + .extensions() + .get::() + .map(|snapshot| snapshot.0), + Some(PeerReplayState::default()), + "interceptor must preserve its unknown-state admission snapshot" + ); + + let capable_state = PeerReplayState { + boot_epoch: Some(boot_epoch), + cache_capability: Some(PeerReplayCapability::Capable { boot_epoch }), + }; + PEER_REPLAY_STATES + .lock() + .expect("peer capability cache lock must not be poisoned") + .insert(audience.to_string(), capable_state); + let (metadata, extensions, body) = request.into_parts(); + let mut request = HttpRequest::new(body); + *request.uri_mut() = "/node_service.NodeService/Lock".parse().expect("test RPC URI must parse"); + *request.headers_mut() = metadata.into_headers(); + *request.extensions_mut() = extensions; + let mut channel = ReplayScopeChannel::new(MissingProofService, Some(audience.to_string())); + + futures::executor::block_on(channel.call(request)).expect("in-flight request response must complete"); + + assert_eq!(peer_replay_state(audience), capable_state); + clear_peer_capability(audience); + } + + #[test] + fn strict_mode_keeps_unknown_peer_mutations_body_bound() { + ensure_test_rpc_secret(); + let audience = "strict-body-digest-client-test:9000"; + clear_peer_capability(audience); + let mut interceptor = test_interceptor_for(audience, true); + let request = rolling_mutation_request("WriteAll"); + let expected_digest = request + .metadata() + .get("x-rustfs-content-sha256") + .and_then(|value| value.to_str().ok()) + .expect("test mutation must carry its digest") + .to_string(); + + let request = interceptor.call(request).expect("interceptor call should succeed"); + + assert_eq!( + request + .metadata() + .get("x-rustfs-content-sha256") + .and_then(|value| value.to_str().ok()), + Some(expected_digest.as_str()) + ); + let nonce = request + .metadata() + .get("x-rustfs-rpc-nonce") + .and_then(|value| value.to_str().ok()) + .and_then(|value| Uuid::parse_str(value).ok()) + .expect("strict body-bound mutation must carry a UUID nonce"); + assert!(!nonce.is_nil()); + assert!( + crate::cluster::rpc::verify_tonic_rpc_signature( + audience, + "/node_service.NodeService/WriteAll", + request.metadata().as_ref(), + ) + .is_ok() + ); + } + + #[test] + fn missing_capability_after_pin_fails_closed() { + ensure_test_rpc_secret(); + let audience = "revoked-capability-client-test:9000"; + let boot_epoch = Uuid::new_v4(); + PEER_REPLAY_STATES + .lock() + .expect("peer capability cache lock must not be poisoned") + .insert( + audience.to_string(), + PeerReplayState { + boot_epoch: Some(boot_epoch), + cache_capability: Some(PeerReplayCapability::Capable { boot_epoch }), + }, + ); + let mut channel = ReplayScopeChannel::new(MissingProofService, Some(audience.to_string())); + futures::executor::block_on(channel.call(replay_scope_request(audience, "Ping"))) + .expect("legacy response must complete before capability rejection"); + + let mut interceptor = test_interceptor_for(audience, false); + let error = interceptor + .call(rolling_mutation_request("Lock")) + .expect_err("a peer that loses its pinned capability must fail closed"); + + assert_eq!(error.code(), tonic::Code::Unauthenticated); + assert_eq!(error.message(), "RPC peer replay capability changed"); + clear_peer_capability(audience); + } + #[test] fn test_signature_interceptor_binds_audience_from_peer_uri() { let interceptor = TonicInterceptor::Signature(gen_tonic_signature_interceptor()) @@ -583,27 +1182,15 @@ mod tests { fn replay_scope_channel_uses_epoch_proof_before_sending_v3() { ensure_test_rpc_secret(); let audience = "replay-scope-client-test:9000"; - PEER_BOOT_EPOCHS - .lock() - .expect("peer epoch cache lock must not be poisoned") - .remove(audience); + clear_peer_capability(audience); let seen_headers = std::sync::Arc::new(Mutex::new(Vec::new())); let service = EpochProofService { audience: audience.to_string(), + include_capability: true, seen_headers: seen_headers.clone(), }; let mut channel = ReplayScopeChannel::new(service, Some(audience.to_string())); - let make_request = || { - let mut request = HttpRequest::builder() - .uri("/node_service.NodeService/Ping") - .body(()) - .expect("test RPC request must build"); - request.headers_mut().extend( - gen_tonic_signature_headers(audience, "node_service.NodeService", "Ping", None) - .expect("v2 test headers must mint"), - ); - request - }; + let make_request = || replay_scope_request(audience, "Ping"); futures::executor::block_on(channel.call(make_request())).expect("first request must complete"); futures::executor::block_on(channel.call(make_request())).expect("second request must complete"); @@ -619,10 +1206,7 @@ mod tests { headers[1].contains_key(RPC_REPLAY_SCOPE_VERSION_HEADER), "the second request must carry the replay-scoped v3 signature" ); - PEER_BOOT_EPOCHS - .lock() - .expect("peer epoch cache lock must not be poisoned") - .remove(audience); + clear_peer_capability(audience); } #[test] diff --git a/crates/ecstore/src/cluster/rpc/http_auth.rs b/crates/ecstore/src/cluster/rpc/http_auth.rs index 48c3ce696..bf05522ed 100644 --- a/crates/ecstore/src/cluster/rpc/http_auth.rs +++ b/crates/ecstore/src/cluster/rpc/http_auth.rs @@ -73,10 +73,14 @@ pub const RPC_REPLAY_SCOPE_NONCE_HEADER: &str = "x-rustfs-rpc-replay-nonce"; pub const RPC_BOOT_EPOCH_HEADER: &str = "x-rustfs-rpc-boot-epoch"; pub const RPC_BOOT_EPOCH_CHALLENGE_HEADER: &str = "x-rustfs-rpc-boot-epoch-challenge"; pub const RPC_BOOT_EPOCH_PROOF_HEADER: &str = "x-rustfs-rpc-boot-epoch-proof"; +pub(crate) const RPC_REPLAY_CACHE_CAPABILITY_HEADER: &str = "x-rustfs-rpc-replay-cache-capability"; +pub(crate) const RPC_REPLAY_CACHE_CAPABILITY_PROOF_HEADER: &str = "x-rustfs-rpc-replay-cache-capability-proof"; const RPC_REPLAY_SCOPE_VERSION_V3: &str = "3"; const RPC_RESPONSE_PROOF_DOMAIN: &[u8] = b"rustfs-rpc-response-proof-v1\0"; const RPC_REPLAY_SCOPE_DOMAIN: &[u8] = b"rustfs-rpc-replay-scope-v3\0"; const RPC_BOOT_EPOCH_PROOF_DOMAIN: &[u8] = b"rustfs-rpc-boot-epoch-proof-v1\0"; +const RPC_REPLAY_CACHE_CAPABILITY_PROOF_DOMAIN: &[u8] = b"rustfs-rpc-replay-cache-capability-proof-v1\0"; +const RPC_REPLAY_CACHE_CAPABILITY_V1: &str = "dynamic-replay-cache-v1"; const HTTP_PUT_FILE_AUTH_DOMAIN: &[u8] = b"rustfs-http-put-file-auth-v1\0"; const HTTP_PUT_FILE_CAPABILITY_AUTH_DOMAIN: &[u8] = b"rustfs-http-put-file-capability-v1\0"; const UNSIGNED_PAYLOAD: &str = "UNSIGNED-PAYLOAD"; @@ -85,8 +89,9 @@ const SIGNATURE_VALID_DURATION: i64 = 300; // 5 minutes const REPLAY_CACHE_RETENTION: Duration = Duration::from_secs(601); const REPLAY_CACHE_RETENTION_SECS: usize = 601; const REPLAY_CACHE_ENTRY_BYTES_ESTIMATE: u64 = 128; -const REPLAY_CACHE_AUTO_MEMORY_PERCENT: u64 = 8; -const REPLAY_CACHE_AUTO_RPC_RPS_PER_CPU: usize = 2048; +// Keep 16 CPU / 32 GiB field nodes at the 32M cap without requiring an env override. +const REPLAY_CACHE_AUTO_MEMORY_PERCENT: u64 = 13; +const REPLAY_CACHE_AUTO_RPC_RPS_PER_CPU: usize = 4096; const REPLAY_CACHE_AUTO_MAX_CAPACITY: usize = 33_554_432; const NS_SCANNER_CAPABILITY_AUTH_DOMAIN: &[u8] = b"rustfs-ns-scanner-capability-v3"; pub const TONIC_RPC_PREFIX: &str = "/node_service.NodeService"; @@ -102,6 +107,10 @@ static INTERNODE_RPC_BODY_DIGEST_STRICT: LazyLock = LazyLock::new(|| { rustfs_config::DEFAULT_INTERNODE_RPC_BODY_DIGEST_STRICT, ) }); + +pub(crate) fn internode_rpc_body_digest_strict() -> bool { + *INTERNODE_RPC_BODY_DIGEST_STRICT +} static INTERNODE_RPC_REPLAY_SCOPE_STRICT: LazyLock = LazyLock::new(|| { get_env_bool( rustfs_config::ENV_INTERNODE_RPC_REPLAY_SCOPE_STRICT, @@ -789,6 +798,50 @@ fn verify_boot_epoch_proof(secret: &str, audience: &str, challenge: Uuid, boot_e .map_err(|_| std::io::Error::new(std::io::ErrorKind::PermissionDenied, "Invalid RPC boot epoch proof")) } +fn update_replay_cache_capability_proof(mac: &mut HmacSha256, audience: &str, challenge: Uuid, boot_epoch: Uuid) { + mac.update(RPC_REPLAY_CACHE_CAPABILITY_PROOF_DOMAIN); + for part in [ + audience.as_bytes(), + b"|", + challenge.as_bytes(), + b"|", + boot_epoch.as_bytes(), + b"|", + RPC_REPLAY_CACHE_CAPABILITY_V1.as_bytes(), + ] { + mac.update(part); + } +} + +fn generate_replay_cache_capability_proof( + secret: &str, + audience: &str, + challenge: Uuid, + boot_epoch: Uuid, +) -> std::io::Result { + let mut mac = + ::new_from_slice(secret.as_bytes()).map_err(|_| std::io::Error::other("Invalid RPC HMAC key"))?; + update_replay_cache_capability_proof(&mut mac, audience, challenge, boot_epoch); + Ok(general_purpose::STANDARD.encode(mac.finalize().into_bytes())) +} + +fn verify_replay_cache_capability_proof( + secret: &str, + audience: &str, + challenge: Uuid, + boot_epoch: Uuid, + proof: &str, +) -> std::io::Result<()> { + let proof = general_purpose::STANDARD + .decode(proof) + .map_err(|_| std::io::Error::other("Invalid RPC replay cache capability proof"))?; + let mut mac = + ::new_from_slice(secret.as_bytes()).map_err(|_| std::io::Error::other("Invalid RPC HMAC key"))?; + update_replay_cache_capability_proof(&mut mac, audience, challenge, boot_epoch); + mac.verify_slice(&proof) + .map_err(|_| std::io::Error::new(std::io::ErrorKind::PermissionDenied, "Invalid RPC replay cache capability proof")) +} + fn non_nil_uuid(value: &str, name: &str) -> std::io::Result { let value = Uuid::parse_str(value).map_err(|_| std::io::Error::other(format!("Invalid {name}")))?; (!value.is_nil()) @@ -871,15 +924,34 @@ pub fn tonic_boot_epoch_challenge(headers: &HeaderMap) -> std::io::Result