From 2dea4a9acffd3a6d39d856e42e95dfbc858087ca Mon Sep 17 00:00:00 2001 From: GatewayJ <835269233@qq.com> Date: Wed, 29 Jul 2026 23:50:59 +0800 Subject: [PATCH] fix(s3): correlate server-owned request IDs (#5433) * fix(s3): correlate server-owned request IDs * fix(server): preserve trace context and Swift routing --------- Co-authored-by: houseme --- .../e2e_test/src/notification_webhook_test.rs | 110 +++- crates/protocols/src/swift/router.rs | 70 ++- rustfs/src/app/object_usecase.rs | 25 +- rustfs/src/app/storage_api.rs | 4 +- rustfs/src/server/http.rs | 100 +-- rustfs/src/server/hybrid.rs | 19 +- rustfs/src/server/layer.rs | 587 ++++++++++++++++-- rustfs/src/server/rate_limit.rs | 45 +- rustfs/src/storage/helper.rs | 121 ++-- rustfs/src/storage/request_context.rs | 130 +++- rustfs/src/storage/storage_api.rs | 6 +- rustfs/src/storage_api.rs | 6 +- 12 files changed, 989 insertions(+), 234 deletions(-) diff --git a/crates/e2e_test/src/notification_webhook_test.rs b/crates/e2e_test/src/notification_webhook_test.rs index 07f4b4f3c..86c546e82 100644 --- a/crates/e2e_test/src/notification_webhook_test.rs +++ b/crates/e2e_test/src/notification_webhook_test.rs @@ -21,18 +21,22 @@ //! function, never as an S3 event sink. //! //! Coverage: -//! * PUT / multipart-complete / DELETE each deliver one event with the correct +//! * PUT / multipart-complete / DeleteObject / DeleteObjects each deliver one event with the correct //! eventName, bucket, key, versionId and eTag. //! * prefix/suffix filters drop non-matching keys (rule-engine gate). //! * an event queued while the target endpoint is unreachable is redelivered //! from the on-disk store once the endpoint recovers (store-and-forward). +//! * responseElements and the S3 response use the canonical request ID while +//! requestParameters preserve a conflicting client-supplied value. use crate::common::{RustFSTestEnvironment, init_logging}; use aws_sdk_s3::Client; +use aws_sdk_s3::operation::RequestId; use aws_sdk_s3::primitives::ByteStream; use aws_sdk_s3::types::{ - BucketVersioningStatus, CompletedMultipartUpload, CompletedPart, Event, FilterRule, FilterRuleName, - NotificationConfiguration, NotificationConfigurationFilter, QueueConfiguration, S3KeyFilter, VersioningConfiguration, + BucketVersioningStatus, CompletedMultipartUpload, CompletedPart, Delete, Event, FilterRule, FilterRuleName, + NotificationConfiguration, NotificationConfigurationFilter, ObjectIdentifier, QueueConfiguration, S3KeyFilter, + VersioningConfiguration, }; use http::header::{CONTENT_TYPE, HOST}; use local_ip_address::local_ip; @@ -40,10 +44,12 @@ use reqwest::StatusCode; use rustfs_signer::constants::UNSIGNED_PAYLOAD; use rustfs_signer::sign_v4; use rustfs_utils::egress::ENV_OUTBOUND_ALLOW_ORIGINS; +use rustfs_utils::http::headers::{AMZ_REQUEST_ID, REQUEST_ID_HEADER}; use s3s::Body; use serde_json::Value; use serial_test::serial; use std::error::Error; +use std::io::Cursor; use std::path::Path; use std::sync::{ Arc, Once, @@ -63,6 +69,8 @@ type BoxError = Box; /// for target lookup (see `process_queue_configurations`), so the region is /// nominal, but it must be present for `ARN::parse` to succeed. const NOTIFY_REGION: &str = "us-east-1"; +const CLIENT_REQUEST_ID: &str = "client-supplied-request-id"; +const CLIENT_AMZ_REQUEST_ID: &str = "client-supplied-amz-request-id"; /// Webhook targets are registered as `TargetID { id: , name: "webhook" }`, /// so the ARN a notification rule references is @@ -579,6 +587,36 @@ fn trimmed_etag(value: Option<&str>) -> Option { value.map(|e| e.trim_matches('"').to_string()) } +fn assert_conflicting_request_id_correlation(record: &Value, server_request_id: &str) { + assert_eq!( + record["requestParameters"][REQUEST_ID_HEADER].as_str(), + Some(CLIENT_REQUEST_ID), + "notification request parameters should retain the actual client header: {record}" + ); + assert_eq!( + record["requestParameters"][AMZ_REQUEST_ID].as_str(), + Some(CLIENT_AMZ_REQUEST_ID), + "notification request parameters should retain the actual client header: {record}" + ); + assert_eq!( + record["responseElements"][AMZ_REQUEST_ID].as_str(), + Some(server_request_id), + "notification response elements should use the canonical request ID: {record}" + ); +} + +fn assert_generated_request_id_correlation(record: &Value, request_id: &str) { + assert!( + record["requestParameters"][AMZ_REQUEST_ID].is_null(), + "notification request parameters must not invent a client request header: {record}" + ); + assert_eq!( + record["responseElements"][AMZ_REQUEST_ID].as_str(), + Some(request_id), + "notification response elements should match the S3 response request ID: {record}" + ); +} + // --------------------------------------------------------------------------- // Tests // --------------------------------------------------------------------------- @@ -671,8 +709,17 @@ async fn test_webhook_event_delivery_and_filtering() -> TestResult { .bucket(bucket) .key(put_key) .body(ByteStream::from_static(b"peri-1 put body")) + .customize() + .mutate_request(|request| { + request.headers_mut().insert(REQUEST_ID_HEADER, CLIENT_REQUEST_ID); + request.headers_mut().insert(AMZ_REQUEST_ID, CLIENT_AMZ_REQUEST_ID); + }) .send() .await?; + let put_request_id = put.request_id().ok_or("PUT response missing request ID")?.to_owned(); + assert!(uuid::Uuid::parse_str(&put_request_id).is_ok()); + assert_ne!(put_request_id, CLIENT_REQUEST_ID); + assert_ne!(put_request_id, CLIENT_AMZ_REQUEST_ID); let put_version = put .version_id() .ok_or("PUT response missing versionId (versioning not enabled?)")?; @@ -692,6 +739,7 @@ async fn test_webhook_event_delivery_and_filtering() -> TestResult { "record eventName: {record}" ); assert_eq!(record["s3"]["bucket"]["name"].as_str(), Some(bucket), "bucket in event: {record}"); + assert_conflicting_request_id_correlation(record, &put_request_id); assert_eq!(object["versionId"].as_str(), Some(put_version), "versionId in event: {object}"); assert_eq!( trimmed_etag(object["eTag"].as_str()), @@ -744,6 +792,35 @@ async fn test_webhook_event_delivery_and_filtering() -> TestResult { "multipart eTag in event: {mp_record}" ); + // --- Snowball extract: direct notification path keeps response correlation + let snowball_key = "uploads/snowball.dat"; + let snowball_body = b"snowball notification body"; + let mut archive_builder = tokio_tar::Builder::new(Cursor::new(Vec::new())); + let mut archive_header = tokio_tar::Header::new_gnu(); + archive_header.set_size(u64::try_from(snowball_body.len()).expect("snowball fixture length should fit in u64")); + archive_header.set_mode(0o644); + archive_header.set_cksum(); + archive_builder + .append_data(&mut archive_header, snowball_key, Cursor::new(snowball_body)) + .await?; + let archive = archive_builder.into_inner().await?.into_inner(); + let snowball = client + .put_object() + .bucket(bucket) + .key("snowball-fixture.tar") + .body(ByteStream::from(archive)) + .customize() + .mutate_request(|request| { + request.headers_mut().insert("x-amz-meta-snowball-auto-extract", "true"); + }) + .send() + .await?; + let snowball_request_id = snowball.request_id().ok_or("Snowball response missing request ID")?; + + let snowball_event = wait_for_event(&mut rx, snowball_key, "s3:ObjectCreated:", Duration::from_secs(20)).await?; + let snowball_record = &snowball_event["Records"][0]; + assert_generated_request_id_correlation(snowball_record, snowball_request_id); + // --- Filter: non-matching prefix and suffix must never be delivered ------ let wrong_prefix = "logs/report.dat"; // right suffix, wrong prefix let wrong_suffix = "uploads/report.txt"; // right prefix, wrong suffix @@ -772,7 +849,32 @@ async fn test_webhook_event_delivery_and_filtering() -> TestResult { "wrong-suffix key {wrong_suffix} bypassed the filter; saw {seen:?}" ); - // --- DELETE on a versioned bucket: ObjectRemoved:* with delete-marker version + // --- DeleteObjects: direct notification path keeps response correlation -- + let delete_many_key = "uploads/delete-many.dat"; + client + .put_object() + .bucket(bucket) + .key(delete_many_key) + .body(ByteStream::from_static(b"delete objects notification body")) + .send() + .await?; + wait_for_event(&mut rx, delete_many_key, "s3:ObjectCreated:", Duration::from_secs(20)).await?; + + let delete_many = client + .delete_objects() + .bucket(bucket) + .delete( + Delete::builder() + .objects(ObjectIdentifier::builder().key(delete_many_key).build()?) + .build()?, + ) + .send() + .await?; + let delete_many_request_id = delete_many.request_id().ok_or("DeleteObjects response missing request ID")?; + let delete_many_event = wait_for_event(&mut rx, delete_many_key, "s3:ObjectRemoved:", Duration::from_secs(20)).await?; + assert_generated_request_id_correlation(&delete_many_event["Records"][0], delete_many_request_id); + + // --- DeleteObject on a versioned bucket: delete-marker version ---------- let delete = client.delete_object().bucket(bucket).key(put_key).send().await?; let removed = wait_for_event(&mut rx, put_key, "s3:ObjectRemoved:", Duration::from_secs(20)).await?; let removed_record = &removed["Records"][0]; diff --git a/crates/protocols/src/swift/router.rs b/crates/protocols/src/swift/router.rs index 6f3f4b3c6..ca2740e72 100644 --- a/crates/protocols/src/swift/router.rs +++ b/crates/protocols/src/swift/router.rs @@ -84,20 +84,19 @@ impl SwiftRouter { Self { enabled, url_prefix } } + /// Return whether a URI matches the Swift route shape without allocating + /// decoded route components. + pub fn matches(&self, uri: &Uri) -> bool { + let Some(path) = self.route_path(uri) else { + return false; + }; + let mut segments = path.trim_start_matches('/').split('/'); + segments.next() == Some("v1") && segments.next().is_some_and(Self::is_valid_account) + } + /// Parse a URI and return a SwiftRoute if it matches Swift URL pattern pub fn route(&self, uri: &Uri, method: Method) -> Option { - if !self.enabled { - return None; - } - - let path = uri.path(); - - // Strip optional prefix - let path = if let Some(prefix) = &self.url_prefix { - path.strip_prefix(&format!("/{}/", prefix))? - } else { - path - }; + let path = self.route_path(uri)?; // Split path into segments - preserve empty segments to maintain object key fidelity // Swift allows trailing slashes and consecutive slashes in object names (e.g., "dir/" or "a//b") @@ -175,6 +174,17 @@ impl SwiftRouter { fn is_valid_account(account: &str) -> bool { ACCOUNT_PATTERN.is_match(account) } + + fn route_path<'a>(&self, uri: &'a Uri) -> Option<&'a str> { + if !self.enabled { + return None; + } + let path = uri.path(); + let Some(prefix) = &self.url_prefix else { + return Some(path); + }; + path.strip_prefix('/')?.strip_prefix(prefix)?.strip_prefix('/') + } } #[cfg(test)] @@ -281,6 +291,42 @@ mod tests { assert_eq!(route, None); } + #[test] + fn test_matches_agrees_with_route_without_decoding_components() { + let router = SwiftRouter::new(true, None); + for path in [ + "/v1/AUTH_project", + "/v1/AUTH_project/", + "/v1/AUTH_project/container", + "/v1/AUTH_project/container/a%20long/object", + "//v1/AUTH_project/container/object", + "/v1/not-a-swift-account/object", + "/v1/AUTH_/object", + "/bucket/object", + ] { + let uri = path.parse().expect("Swift route test URI"); + assert_eq!( + router.matches(&uri), + router.route(&uri, Method::GET).is_some(), + "classification must match full routing for {path}" + ); + } + + let prefixed_router = SwiftRouter::new(true, Some("swift".to_string())); + for path in [ + "/swift/v1/AUTH_project/container", + "/swiftish/v1/AUTH_project/container", + "/v1/AUTH_project/container", + ] { + let uri = path.parse().expect("prefixed Swift route test URI"); + assert_eq!( + prefixed_router.matches(&uri), + prefixed_router.route(&uri, Method::GET).is_some(), + "prefixed classification must match full routing for {path}" + ); + } + } + #[test] fn test_project_id_extraction() { let route = SwiftRoute::Account { diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 1e789a20f..8d46feadd 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -72,7 +72,7 @@ use super::storage_api::object_usecase::error::{ is_err_version_not_found, }; use super::storage_api::object_usecase::head_prefix::{head_prefix_not_found_message, probe_prefix_has_children}; -use super::storage_api::object_usecase::helper::{OperationHelper, spawn_background_with_context}; +use super::storage_api::object_usecase::helper::{OperationHelper, build_event_resp_elements, spawn_background_with_context}; use super::storage_api::object_usecase::io::{DynReader, HashReader, WritePlan, compression_metadata_value, wrap_reader}; #[cfg(test)] use super::storage_api::object_usecase::object_cache::GetObjectBodySource; @@ -130,9 +130,7 @@ use rustfs_object_capacity::capacity_manager::get_capacity_manager; use rustfs_policy::policy::action::{Action, S3Action}; use rustfs_s3_ops::{S3Operation, delete_event_name_for_marker, put_event_name_for_post_object}; use rustfs_s3select_api::object_store::bytes_stream; -use rustfs_targets::{ - EventName, extract_params_header, extract_resp_elements, get_request_host, get_request_port, get_request_user_agent, -}; +use rustfs_targets::{EventName, get_request_host, get_request_port, get_request_user_agent}; use rustfs_utils::CompressionAlgorithm; use rustfs_utils::http::{ AMZ_BUCKET_REPLICATION_STATUS, AMZ_CHECKSUM_MODE, AMZ_CHECKSUM_TYPE, AMZ_WEBSITE_REDIRECT_LOCATION, CONTENT_TYPE, @@ -6540,6 +6538,7 @@ impl DefaultObjectUsecase { } let helper = OperationHelper::new(&req, EventName::ObjectRemovedDelete, S3Operation::DeleteObjects).suppress_event(); + let request_context = helper.request_context_or_from_request(&req); let (bucket, delete) = { let bucket = req.input.bucket.clone(); let delete = req.input.delete.clone(); @@ -6948,10 +6947,12 @@ impl DefaultObjectUsecase { let req_headers = req.headers.clone(); let notify = current_notify_interface_for_context(self.context.as_deref()); - let request_context = req.extensions.get::().cloned(); + let req_params = rustfs_targets::extract_params_header(&req_headers); + let resp_elements = + build_event_resp_elements(&S3Response::new(DeleteObjectsOutput::default()), &request_context.request_id); let deleted_any = delete_results.iter().any(|result| result.delete_object.is_some()); let notify_bucket = bucket.clone(); - spawn_background_with_context(request_context, async move { + spawn_background_with_context(Some(request_context), async move { let _activity_guard = DeleteTailActivityGuard::new(DeleteTailStage::Notify); for res in delete_results { if let Some(dobj) = res.delete_object { @@ -6966,8 +6967,8 @@ impl DefaultObjectUsecase { }), ) .version_id(dobj.version_id.map(|v| v.to_string()).unwrap_or_default()) - .req_params(extract_params_header(&req_headers)) - .resp_elements(extract_resp_elements(&S3Response::new(DeleteObjectsOutput::default()))) + .req_params(req_params.clone()) + .resp_elements(resp_elements.clone()) .host(get_request_host(&req_headers)) .user_agent(get_request_user_agent(&req_headers)) .build(); @@ -7849,6 +7850,7 @@ impl DefaultObjectUsecase { #[instrument(level = "debug", skip(self, req))] pub async fn execute_put_object_extract(&self, req: S3Request) -> S3Result> { let helper = OperationHelper::new(&req, EventName::ObjectCreatedPut, S3Operation::PutObject).suppress_event(); + let request_context = helper.request_context_or_from_request(&req); let auth_method = req.method.clone(); let auth_uri = req.uri.clone(); let auth_headers = req.headers.clone(); @@ -8025,7 +8027,7 @@ impl DefaultObjectUsecase { }; let notify = current_notify_interface_for_context(self.context.as_deref()); - let req_params = extract_params_header(&req.headers); + let req_params = rustfs_targets::extract_params_header(&req.headers); let host = get_request_host(&req.headers); let port = get_request_port(&req.headers); let user_agent = get_request_user_agent(&req.headers); @@ -8254,7 +8256,7 @@ impl DefaultObjectUsecase { bucket_name: bucket.clone(), object: convert_ecstore_object_info(obj_info.clone()), req_params: req_params.clone(), - resp_elements: extract_resp_elements(&S3Response::new(output.clone())), + resp_elements: build_event_resp_elements(&S3Response::new(output.clone()), &request_context.request_id), version_id: version_id.clone(), host: host.clone(), port, @@ -8262,8 +8264,7 @@ impl DefaultObjectUsecase { }; let notify = notify.clone(); - let request_context = req.extensions.get::().cloned(); - spawn_background_with_context(request_context, async move { + spawn_background_with_context(Some(request_context.clone()), async move { notify.notify(event_args).await; }); } diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index b97477b9a..4b9f6daa7 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -853,7 +853,9 @@ pub(crate) mod head_prefix { } pub(crate) mod helper { - pub(crate) use crate::storage::storage_api::helper_consumer::{OperationHelper, spawn_background_with_context}; + pub(crate) use crate::storage::storage_api::helper_consumer::{ + OperationHelper, build_event_resp_elements, spawn_background_with_context, + }; } pub(crate) mod object_utils { diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index c7c49bbdf..fc8e7d712 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -23,9 +23,9 @@ use crate::server::{ hybrid::hybrid, layer::{ BodylessStatusFixLayer, ConditionalCorsLayer, DoubleSlashListBucketsCompatLayer, EmptyBodyContentLengthCompatLayer, - HeadRequestBodyFixLayer, IcebergRestErrorCompatLayer, ObjectAttributesEtagFixLayer, PublicHealthEndpointLayer, - RedirectLayer, RequestContextLayer, RequestLoggingLayer, S3ErrorMessageCompatLayer, StsQueryApiCompatLayer, - VirtualHostStyleHintLayer, redact_sensitive_uri_query, + ExternalRequestContextLayer, HeadRequestBodyFixLayer, IcebergRestErrorCompatLayer, ObjectAttributesEtagFixLayer, + PublicHealthEndpointLayer, RedirectLayer, RequestContextLayer, RequestLoggingLayer, S3ErrorMessageCompatLayer, + StsQueryApiCompatLayer, VirtualHostStyleHintLayer, redact_sensitive_uri_query, }, rate_limit::{RateLimitLayer, api_rate_limit_layer_from_env}, tls_material::{ @@ -34,7 +34,6 @@ use crate::server::{ }, }; use crate::storage_api::server::http as storage; -use crate::storage_api::server::http::request_context::{RequestContext, extract_request_id_from_headers}; use crate::storage_api::server::http::rpc::InternodeRpcService; use crate::storage_api::server::http::tonic_service::make_server; use crate::storage_api::server::http::{ @@ -1135,47 +1134,6 @@ struct PathDispatchService { internode: B, } -#[derive(Clone, Default)] -struct InternodeRequestContextLiteLayer; - -impl tower::Layer for InternodeRequestContextLiteLayer { - type Service = InternodeRequestContextLiteService; - - fn layer(&self, inner: S) -> Self::Service { - InternodeRequestContextLiteService { inner } - } -} - -#[derive(Clone)] -struct InternodeRequestContextLiteService { - inner: S, -} - -impl Service> for InternodeRequestContextLiteService -where - S: Service> + Clone, -{ - type Response = S::Response; - type Error = S::Error; - type Future = S::Future; - - fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { - self.inner.poll_ready(cx) - } - - fn call(&mut self, mut req: HttpRequest) -> Self::Future { - let request_id = extract_request_id_from_headers(req.headers()); - req.extensions_mut().insert(RequestContext { - x_amz_request_id: request_id.clone(), - request_id, - trace_id: None, - span_id: None, - start_time: std::time::Instant::now(), - }); - self.inner.call(req) - } -} - impl PathDispatchService { fn new(external: A, internode: B) -> Self { Self { external, internode } @@ -1388,34 +1346,28 @@ fn process_connection( // 1. AddExtensionLayer — per-connection peer address // 2. AddExtensionLayer — per-connection raw socket addr (TrustedProxy) // 3. TrustedProxyLayer — conditional, parses X-Forwarded-For - // 4. SetRequestIdLayer — generates X-Request-ID - // 5. RequestContextLayer — creates RequestContext in extensions - // 6. StsQueryApiCompatLayer — route-scoped STS envelopes, including outer short-circuit errors - // 7. EmptyBodyContentLengthCompatLayer — adds Content-Length: 0 for known empty-body API routes - // 8. CatchPanicLayer — panic → 500 - // 9. RateLimitLayer — conditional (external stack only), per-client 429 throttling - // 10. ReadinessGateLayer — blocks until ready - // 11. KeystoneAuthLayer — X-Auth-Token validation - // 12. TraceLayer — request span creation + metrics - // 13. RequestLoggingLayer — single completion event per request - // 14. PropagateRequestIdLayer — X-Request-ID → response - // 15. CompressionLayer — response compression (whitelist, path-aware) - // 16. PathCategoryInjectionLayer — injects path category for compression predicate - // 17. S3ErrorMessageCompatLayer — missing S3 error message compatibility - // 18. IcebergRestErrorCompatLayer — Iceberg REST JSON error compatibility - // 19. ObjectAttributesEtagFixLayer — ETag fix for GetObjectAttributes - // 20. ConditionalCorsLayer — S3 API CORS - // 21. RedirectLayer — console redirect (conditional) - // 22. BodylessStatusFixLayer — clears body for 1xx/204/205/304 responses - // 23. HeadRequestBodyFixLayer — strips actual body bytes from HEAD responses - // 24. PublicHealthEndpointLayer — handles public health before s3s host parsing - // 25. VirtualHostStyleHintLayer — actionable error for unroutable virtual-hosted-style (conditional) - // 26. DoubleSlashListBucketsCompatLayer — rewrites `GET //` to `GET /` for ListBuckets (MinIO browser compat) + // 4. ExternalRequestContextLayer — S3 canonical ID / control-plane propagated ID + // 5. StsQueryApiCompatLayer — route-scoped STS envelopes, including outer short-circuit errors + // 6. EmptyBodyContentLengthCompatLayer — adds Content-Length: 0 for known empty-body API routes + // 7. CatchPanicLayer — panic → 500 + // 8. RateLimitLayer — conditional (external stack only), per-client 429 throttling + // 9. ReadinessGateLayer — blocks until ready + // 10. KeystoneAuthLayer — X-Auth-Token validation + // 11. TraceLayer — request span creation + metrics + // 12. RequestLoggingLayer — single completion event per request + // 13. CompressionLayer — response compression (whitelist, path-aware) + // 14. PathCategoryInjectionLayer — injects path category for compression predicate + // 15. S3ErrorMessageCompatLayer — missing S3 error message compatibility + // 16. IcebergRestErrorCompatLayer — Iceberg REST JSON error compatibility + // 17. ObjectAttributesEtagFixLayer — ETag fix for GetObjectAttributes + // 18. ConditionalCorsLayer — S3 API CORS + // 19. RedirectLayer — console redirect (conditional) + // 20. BodylessStatusFixLayer — clears body for 1xx/204/205/304 responses + // 21. HeadRequestBodyFixLayer — strips actual body bytes from HEAD responses + // 22. PublicHealthEndpointLayer — handles public health before s3s host parsing + // 23. VirtualHostStyleHintLayer — actionable error for unroutable virtual-hosted-style (conditional) + // 24. DoubleSlashListBucketsCompatLayer — rewrites `GET //` to `GET /` for ListBuckets (MinIO browser compat) // ───────────────────────────────────────────────────────────── - // Batch 1 intentionally keeps the external and internode stacks behaviorally - // identical while giving each path family a named construction boundary. - // Later batches will trim internode-only middleware without risking drift in - // the public HTTP stack. let build_external_stack = |service| { ServiceBuilder::new() // NOTE: Both extension types are intentionally inserted to maintain compatibility: @@ -1430,8 +1382,7 @@ fn process_connection( // This should be placed before TraceLayer so that logs reflect the real client IP // Pre-computed in ConnectionContext to avoid per-connection is_enabled() check. .option_layer(trusted_proxy_layer.clone()) - .layer(SetRequestIdLayer::x_request_id(MakeRequestUuid)) - .layer(InternodeRequestContextLiteLayer) + .layer(ExternalRequestContextLayer::new(is_console)) .layer(StsQueryApiCompatLayer) .layer(EmptyBodyContentLengthCompatLayer) .layer(CatchPanicLayer::new()) @@ -1584,7 +1535,6 @@ fn process_connection( }), ) .layer(RequestLoggingLayer) - .layer(PropagateRequestIdLayer::x_request_id()) .layer(CompressionLayer::new().compress_when(PathAwareHttpCompressionPredicate::new(compression_config.clone()))) .layer(PathCategoryInjectionLayer) .layer(S3ErrorMessageCompatLayer) diff --git a/rustfs/src/server/hybrid.rs b/rustfs/src/server/hybrid.rs index fe614c4c8..c20d47215 100644 --- a/rustfs/src/server/hybrid.rs +++ b/rustfs/src/server/hybrid.rs @@ -28,6 +28,13 @@ pub(crate) fn hybrid(make_rest: MakeRest, grpc: Grpc) -> HybridS HybridService { rest: make_rest, grpc } } +pub(crate) fn is_grpc_request(req: &Request) -> bool { + matches!( + (req.version(), req.headers().get(hyper::header::CONTENT_TYPE)), + (hyper::Version::HTTP_2, Some(value)) if value.as_bytes().starts_with(b"application/grpc") + ) +} + /// The service that can serve both gRPC and REST HTTP Requests #[derive(Clone)] pub struct HybridService { @@ -63,14 +70,14 @@ where /// and if the Content-Type is "application/grpc"; otherwise, the request is served /// as a REST request fn call(&mut self, req: Request) -> Self::Future { - match (req.version(), req.headers().get(hyper::header::CONTENT_TYPE)) { - (hyper::Version::HTTP_2, Some(hv)) if hv.as_bytes().starts_with(b"application/grpc") => HybridFuture::Grpc { + if is_grpc_request(&req) { + HybridFuture::Grpc { grpc_future: self.grpc.call(req), - }, - - _ => HybridFuture::Rest { + } + } else { + HybridFuture::Rest { rest_future: self.rest.call(req), - }, + } } } } diff --git a/rustfs/src/server/layer.rs b/rustfs/src/server/layer.rs index aab384932..41eabeb82 100644 --- a/rustfs/src/server/layer.rs +++ b/rustfs/src/server/layer.rs @@ -17,27 +17,28 @@ use crate::admin::console::is_console_path; use crate::error::ApiError; use crate::server::RemoteAddr; use crate::server::cors; -use crate::server::hybrid::HybridBody; +use crate::server::hybrid::{HybridBody, is_grpc_request}; use crate::server::{ ADMIN_PREFIX, CONSOLE_PREFIX, HEALTH_COMPAT_LIVE_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, HealthProbe, MINIO_ADMIN_PREFIX, MINIO_ADMIN_V3_PREFIX, MINIO_HEALTH_CLUSTER_PATH, MINIO_HEALTH_CLUSTER_READ_PATH, MINIO_HEALTH_LIVE_PATH, - MINIO_HEALTH_READY_PATH, RPC_PREFIX, RUSTFS_ADMIN_PREFIX, active_http_requests, build_health_response_parts, - collect_probe_readiness, has_path_prefix, is_admin_path, is_table_catalog_path, + MINIO_HEALTH_READY_PATH, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH, RPC_PREFIX, RUSTFS_ADMIN_PREFIX, active_http_requests, + build_health_response_parts, collect_probe_readiness, has_path_prefix, is_admin_path, is_table_catalog_path, }; use crate::storage_api::server::layer::apply_cors_headers; -use crate::storage_api::server::layer::request_context::{ - RequestContext, extract_request_id_from_headers, extract_trace_context_ids_from_headers, spawn_traced, -}; +use crate::storage_api::server::layer::request_context::{RequestContext, extract_request_id_from_headers, spawn_traced}; use bytes::{Bytes, BytesMut}; use futures::future::Either; use http::{HeaderMap, HeaderValue, Method, Request as HttpRequest, Response, StatusCode, Uri}; use http_body::Body; use http_body_util::{BodyExt, Full}; use hyper::body::Incoming; +use pin_project_lite::pin_project; use quick_xml::events::Event; +#[cfg(feature = "swift")] +use rustfs_protocols::swift::SwiftRouter; use rustfs_trusted_proxies::ClientInfo; use rustfs_utils::get_env_opt_str; -use rustfs_utils::http::headers::AMZ_REQUEST_ID; +use rustfs_utils::http::headers::{AMZ_REQUEST_ID, REQUEST_ID_HEADER}; use s3s::S3ErrorCode; use serde::{Deserialize, Serialize}; use std::borrow::Cow; @@ -62,6 +63,8 @@ const HTTP_REQUEST_INFLIGHT_WARN_THRESHOLD: Duration = Duration::from_secs(5); const STS_RESPONSE_METADATA_TAG: &str = "ResponseMetadata"; const STS_REQUEST_ID_TAG: &str = "RequestId"; const STS_SUCCESS_RESPONSE_TAGS: [&str; 2] = ["AssumeRoleResponse", "AssumeRoleWithWebIdentityResponse"]; +#[cfg(feature = "swift")] +const SWIFT_API_PATH_PREFIX: &str = "/v1/"; pub(crate) fn redact_sensitive_uri_query(uri: &http::Uri) -> String { let path = uri.path(); @@ -117,9 +120,6 @@ fn is_object_zip_download_path(path: &str) -> bool { /// /// This layer must be placed after `SetRequestIdLayer` in the middleware stack, /// as it reads the `x-request-id` header that `SetRequestIdLayer` generates. -/// -/// Additionally, it preserves any upstream `x-amz-request-id` in the separate -/// `RequestContext.x_amz_request_id` field without mutating signed request headers. #[derive(Clone, Default)] pub struct RequestContextLayer; @@ -150,35 +150,163 @@ where } fn call(&mut self, mut req: HttpRequest) -> Self::Future { - let request_id = extract_request_id_from_headers(req.headers()); - - let (trace_id, span_id) = extract_trace_context_ids_from_headers(req.headers()) - .map(|(trace_id, span_id)| (Some(trace_id), Some(span_id))) - .unwrap_or((None, None)); - - // Preserve the upstream x-amz-request-id if present as the S3 compatibility alias; - // otherwise mirror the canonical internal request_id. - let x_amz_request_id = req - .headers() - .get(AMZ_REQUEST_ID) - .and_then(|v| v.to_str().ok()) - .map(String::from) - .unwrap_or_else(|| request_id.clone()); - - let ctx = RequestContext { - request_id, - x_amz_request_id, - trace_id, - span_id, - start_time: Instant::now(), - }; - - req.extensions_mut().insert(ctx); + let request_context = RequestContext::from_headers(req.headers()); + req.extensions_mut().insert(request_context); self.inner.call(req) } } +fn uses_server_owned_s3_request_id(req: &HttpRequest, console_redirect_enabled: bool) -> bool { + if is_grpc_request(req) + || req.uri().path().starts_with(RPC_PREFIX) + || is_sts_query_request(req.method(), req.uri(), req.headers()) + || (console_redirect_enabled && is_console_redirect_request(req)) + { + return false; + } + + #[cfg(feature = "swift")] + if req.uri().path().starts_with(SWIFT_API_PATH_PREFIX) && SwiftRouter::new(true, None).matches(req.uri()) { + return false; + } + + let path = req.uri().path(); + let method = req.method(); + let is_admin_health_request = + (method == Method::GET || method == Method::HEAD) && matches!(path, HEALTH_PREFIX | HEALTH_READY_PATH); + let is_profile_request = method == Method::GET && matches!(path, PROFILE_CPU_PATH | PROFILE_MEMORY_PATH); + let is_public_health_alias_request = (method == Method::GET || method == Method::HEAD) + && matches!( + path, + HEALTH_COMPAT_LIVE_PATH + | MINIO_HEALTH_LIVE_PATH + | MINIO_HEALTH_READY_PATH + | MINIO_HEALTH_CLUSTER_PATH + | MINIO_HEALTH_CLUSTER_READ_PATH + ) + && is_public_health_endpoint_request(method, path); + + !(is_admin_path(path) + || is_console_path(path) + || is_admin_health_request + || is_profile_request + || is_public_health_alias_request) +} + +/// Creates a server-owned context and response ID for S3 requests while +/// preserving the existing request-ID propagation contract for non-S3 routes. +#[derive(Clone, Default)] +pub struct ExternalRequestContextLayer { + console_redirect_enabled: bool, +} + +impl ExternalRequestContextLayer { + pub(crate) fn new(console_redirect_enabled: bool) -> Self { + Self { + console_redirect_enabled, + } + } +} + +impl Layer for ExternalRequestContextLayer { + type Service = ExternalRequestContextService; + + fn layer(&self, inner: S) -> Self::Service { + ExternalRequestContextService { + inner, + console_redirect_enabled: self.console_redirect_enabled, + } + } +} + +#[derive(Clone)] +pub struct ExternalRequestContextService { + inner: S, + console_redirect_enabled: bool, +} + +impl Service> for ExternalRequestContextService +where + S: Service, Response = Response>, +{ + type Response = Response; + type Error = S::Error; + type Future = ExternalRequestContextFuture; + + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + self.inner.poll_ready(cx) + } + + fn call(&mut self, mut req: HttpRequest) -> Self::Future { + let is_s3 = uses_server_owned_s3_request_id(&req, self.console_redirect_enabled); + let has_request_id = req.headers().contains_key(REQUEST_ID_HEADER); + let request_context = if is_s3 { + let request_context = RequestContext::from_external_headers(req.headers()); + if !has_request_id && let Ok(request_id) = HeaderValue::from_str(&request_context.request_id) { + req.headers_mut().insert(REQUEST_ID_HEADER, request_id); + } + request_context + } else { + if !has_request_id { + let request_id = uuid::Uuid::new_v4().to_string(); + if let Ok(request_id) = HeaderValue::from_str(&request_id) { + req.headers_mut().insert(REQUEST_ID_HEADER, request_id); + } + } + RequestContext::from_headers_without_trace_context(req.headers()) + }; + let request_id = if is_s3 { + HeaderValue::from_str(&request_context.request_id).ok() + } else { + req.headers().get(REQUEST_ID_HEADER).cloned() + }; + req.extensions_mut().insert(request_context); + + ExternalRequestContextFuture { + inner: self.inner.call(req), + request_id, + is_s3, + } + } +} + +pin_project! { + pub struct ExternalRequestContextFuture { + #[pin] + inner: F, + request_id: Option, + is_s3: bool, + } +} + +impl Future for ExternalRequestContextFuture +where + F: Future, E>>, +{ + type Output = Result, E>; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let this = self.project(); + let mut response = match this.inner.poll(cx) { + Poll::Ready(Ok(response)) => response, + Poll::Ready(Err(error)) => return Poll::Ready(Err(error)), + Poll::Pending => return Poll::Pending, + }; + + if let Some(request_id) = this.request_id.take() { + if *this.is_s3 { + response.headers_mut().insert(REQUEST_ID_HEADER, request_id.clone()); + response.headers_mut().insert(AMZ_REQUEST_ID, request_id); + } else if !response.headers().contains_key(REQUEST_ID_HEADER) { + response.headers_mut().insert(REQUEST_ID_HEADER, request_id); + } + } + + Poll::Ready(Ok(response)) + } +} + #[derive(Clone, Default)] pub struct RequestLoggingLayer; @@ -395,6 +523,18 @@ pub struct RedirectService { inner: S, } +fn is_console_redirect_request(req: &HttpRequest) -> bool { + let path = req.uri().path().trim_end_matches('/'); + req.method() == http::Method::GET + && !req.headers().contains_key(http::header::AUTHORIZATION) + && req + .headers() + .get(http::header::USER_AGENT) + .and_then(|value| value.to_str().ok()) + .is_some_and(|user_agent| user_agent.contains("Mozilla")) + && (path.is_empty() || path == "/rustfs" || path == "/index.html") +} + impl Service> for RedirectService where S: Service, Response = Response>> + Clone + Send + 'static, @@ -412,20 +552,8 @@ where } fn call(&mut self, req: HttpRequest) -> Self::Future { - // Check if this is a GET request without Authorization header and User-Agent contains Mozilla - // and the path is either "/" or "/index.html" let path = req.uri().path().trim_end_matches('/'); - let should_redirect = req.method() == http::Method::GET - && !req.headers().contains_key(http::header::AUTHORIZATION) - && req - .headers() - .get(http::header::USER_AGENT) - .and_then(|v| v.to_str().ok()) - .map(|ua| ua.contains("Mozilla")) - .unwrap_or(false) - && (path.is_empty() || path == "/rustfs" || path == "/index.html"); - - if should_redirect { + if is_console_redirect_request(&req) { debug!("Redirecting browser request from {} to console", path); // Create redirect response @@ -1733,7 +1861,7 @@ impl ConditionalCorsLayer { // Expose common headers response_headers.insert( cors::response::ACCESS_CONTROL_EXPOSE_HEADERS, - HeaderValue::from_static("x-request-id, content-type, content-length, etag"), + HeaderValue::from_static("x-request-id, x-amz-request-id, content-type, content-length, etag"), ); // Credentials are only safe for origins matched from an explicit allow-list. @@ -2046,12 +2174,25 @@ mod tests { #[derive(Clone, Default)] struct HeaderCaptureService { headers: Arc>>, + request_context: Arc>>, + response_request_id: Option, } impl HeaderCaptureService { + fn with_response_request_id(request_id: &'static str) -> Self { + Self { + response_request_id: Some(HeaderValue::from_static(request_id)), + ..Self::default() + } + } + fn headers(&self) -> Arc>> { Arc::clone(&self.headers) } + + fn request_context(&self) -> Arc>> { + Arc::clone(&self.request_context) + } } impl Service> for HeaderCaptureService { @@ -2065,10 +2206,350 @@ mod tests { fn call(&mut self, req: Request) -> Self::Future { *self.headers.lock().expect("capture headers") = Some(req.headers().clone()); - ready(Ok(Response::new(Full::from(Bytes::new())))) + *self.request_context.lock().expect("capture request context") = req.extensions().get::().cloned(); + let mut response = Response::new(Full::from(Bytes::new())); + if let Some(request_id) = self.response_request_id.clone() { + response.headers_mut().insert(REQUEST_ID_HEADER, request_id); + } + ready(Ok(response)) } } + async fn assert_non_s3_request_id_contract(mut request: Request<()>, route: &str) { + let capture = HeaderCaptureService::default(); + let captured_context = capture.request_context(); + let mut service = ExternalRequestContextLayer::default().layer(capture); + request + .headers_mut() + .insert(REQUEST_ID_HEADER, HeaderValue::from_static("client-request-id")); + request + .headers_mut() + .insert(AMZ_REQUEST_ID, HeaderValue::from_static("client-amz-request-id")); + + let response = service.call(request).await.expect("non-S3 response"); + + assert_eq!( + response + .headers() + .get(REQUEST_ID_HEADER) + .and_then(|value| value.to_str().ok()), + Some("client-request-id"), + "non-S3 x-request-id contract changed for {route}" + ); + assert!( + !response.headers().contains_key(AMZ_REQUEST_ID), + "non-S3 response unexpectedly gained x-amz-request-id for {route}" + ); + let context = captured_context + .lock() + .expect("captured request context") + .clone() + .expect("non-S3 request context"); + assert_eq!(context.request_id, "client-request-id", "non-S3 context changed for {route}"); + assert_eq!(context.x_amz_request_id, "client-request-id"); + assert!(context.trace_id.is_none()); + assert!(context.span_id.is_none()); + } + + #[tokio::test] + async fn external_request_context_rejects_client_request_id_as_canonical() { + global::set_text_map_propagator(TraceContextPropagator::new()); + let capture = HeaderCaptureService::default(); + let captured_headers = capture.headers(); + let captured_context = capture.request_context(); + let mut service = ExternalRequestContextLayer::default().layer(capture); + let mut request = Request::builder().uri("/bucket/object").body(()).expect("build S3 request"); + request + .headers_mut() + .insert(REQUEST_ID_HEADER, HeaderValue::from_static("client-supplied-request-id")); + request + .headers_mut() + .insert(AMZ_REQUEST_ID, HeaderValue::from_static("client-supplied-amz-request-id")); + request.headers_mut().insert( + "traceparent", + HeaderValue::from_static("00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"), + ); + + let response = service.call(request).await.expect("response"); + let request_id = response + .headers() + .get(REQUEST_ID_HEADER) + .and_then(|value| value.to_str().ok()) + .expect("canonical request ID"); + + assert!(uuid::Uuid::parse_str(request_id).is_ok()); + assert_ne!(request_id, "client-supplied-request-id"); + assert_eq!( + response.headers().get(AMZ_REQUEST_ID).and_then(|value| value.to_str().ok()), + Some(request_id) + ); + let headers = captured_headers.lock().expect("captured headers"); + let headers = headers.as_ref().expect("inner request headers"); + assert_eq!(headers.get(REQUEST_ID_HEADER).expect("client x-request-id"), "client-supplied-request-id"); + assert_eq!( + headers.get(AMZ_REQUEST_ID).expect("client x-amz-request-id"), + "client-supplied-amz-request-id" + ); + assert_eq!( + headers.get("traceparent").expect("client traceparent"), + "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01" + ); + let context = captured_context + .lock() + .expect("captured request context") + .clone() + .expect("S3 request context"); + assert_eq!(context.request_id, request_id); + assert_eq!(context.trace_id.as_deref(), Some("4bf92f3577b34da6a3ce929d0e0e4736")); + assert_eq!(context.span_id.as_deref(), Some("00f067aa0ba902b7")); + } + + #[tokio::test] + async fn external_request_context_replaces_empty_response_id() { + let capture = HeaderCaptureService::default(); + let captured_headers = capture.headers(); + let mut service = ExternalRequestContextLayer::default().layer(capture); + let mut request = Request::builder().uri("/bucket/object").body(()).expect("build S3 request"); + request.headers_mut().insert(REQUEST_ID_HEADER, HeaderValue::from_static("")); + request.headers_mut().insert(AMZ_REQUEST_ID, HeaderValue::from_static(" ")); + + let response = service.call(request).await.expect("response"); + let request_id = response + .headers() + .get(REQUEST_ID_HEADER) + .and_then(|value| value.to_str().ok()) + .expect("canonical request ID"); + + assert!(uuid::Uuid::parse_str(request_id).is_ok()); + assert_eq!( + response.headers().get(AMZ_REQUEST_ID).and_then(|value| value.to_str().ok()), + Some(request_id) + ); + let headers = captured_headers.lock().expect("captured headers"); + let headers = headers.as_ref().expect("inner request headers"); + assert_eq!(headers.get(REQUEST_ID_HEADER).expect("empty client x-request-id"), ""); + assert_eq!(headers.get(AMZ_REQUEST_ID).expect("blank client x-amz-request-id"), " "); + } + + #[tokio::test] + async fn external_request_context_inserts_generated_id_when_header_is_absent() { + let capture = HeaderCaptureService::default(); + let captured_headers = capture.headers(); + let mut service = ExternalRequestContextLayer::default().layer(capture); + let request = Request::builder().uri("/bucket/object").body(()).expect("build S3 request"); + + let response = service.call(request).await.expect("response"); + let response_request_id = response + .headers() + .get(REQUEST_ID_HEADER) + .and_then(|value| value.to_str().ok()) + .expect("canonical request ID"); + let headers = captured_headers.lock().expect("captured headers"); + let headers = headers.as_ref().expect("inner request headers"); + + assert_eq!( + headers.get(REQUEST_ID_HEADER).and_then(|value| value.to_str().ok()), + Some(response_request_id) + ); + } + + #[tokio::test] + async fn non_s3_request_id_contract_preserves_client_correlation() { + for path in [ + "/rustfs/admin/v3/info", + "/minio/admin/v3/info", + "/rustfs/console/", + HEALTH_PREFIX, + "/iceberg/v1/config", + "/rustfs/rpc/v1/read-file", + "/rustfs/rpcx", + ] { + let request = Request::builder().uri(path).body(()).expect("build non-S3 request"); + assert_non_s3_request_id_contract(request, path).await; + } + } + + #[test] + fn console_redirect_request_id_contract_follows_redirect_enablement() { + for path in ["/", "/rustfs", "/index.html"] { + let request = Request::builder() + .method(Method::GET) + .uri(path) + .header(http::header::USER_AGENT, "Mozilla/5.0") + .body(()) + .expect("build console redirect request"); + + assert!(!uses_server_owned_s3_request_id(&request, true)); + assert!(uses_server_owned_s3_request_id(&request, false)); + } + } + + #[test] + fn method_scoped_control_routes_preserve_request_id_contract() { + for path in [HEALTH_READY_PATH, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH] { + let control_request = Request::builder() + .method(Method::GET) + .uri(path) + .body(()) + .expect("build control-plane request"); + assert!(!uses_server_owned_s3_request_id(&control_request, false)); + + let s3_request = Request::builder() + .method(Method::POST) + .uri(path) + .body(()) + .expect("build S3 request"); + assert!(uses_server_owned_s3_request_id(&s3_request, false)); + } + } + + #[test] + #[serial] + fn public_health_alias_request_id_contract_follows_enablement() { + let request = Request::builder() + .method(Method::GET) + .uri(MINIO_HEALTH_LIVE_PATH) + .body(()) + .expect("build health request"); + + with_var(rustfs_config::ENV_HEALTH_ENDPOINT_ENABLE, Some("true"), || { + assert!(!uses_server_owned_s3_request_id(&request, false)); + }); + with_var(rustfs_config::ENV_HEALTH_ENDPOINT_ENABLE, Some("false"), || { + assert!(uses_server_owned_s3_request_id(&request, false)); + }); + } + + #[tokio::test] + async fn grpc_request_id_contract_preserves_client_correlation() { + for path in [ + "/node_service.NodeService/GetMetrics", + "/node_service.HealControlService/HealControl", + "/node_service.TierMutationControlService/PrepareTierMutation", + ] { + let request = Request::builder() + .version(http::Version::HTTP_2) + .uri(path) + .header(http::header::CONTENT_TYPE, "application/grpc") + .body(()) + .expect("build gRPC request"); + assert_non_s3_request_id_contract(request, path).await; + } + } + + #[tokio::test] + async fn sts_query_request_id_contract_preserves_client_correlation() { + let request = Request::builder() + .method(Method::POST) + .uri("/") + .header(http::header::CONTENT_TYPE, "application/x-www-form-urlencoded") + .body(()) + .expect("build STS Query request"); + assert_non_s3_request_id_contract(request, "STS Query").await; + } + + #[cfg(feature = "swift")] + #[tokio::test] + async fn swift_request_id_contract_preserves_client_correlation() { + let path = "/v1/AUTH_project/container/object"; + let request = Request::builder().uri(path).body(()).expect("build Swift request"); + assert_non_s3_request_id_contract(request, path).await; + } + + #[cfg(feature = "swift")] + #[test] + fn swift_request_id_classification_preserves_v1_prefix_boundary() { + let swift_request = Request::builder() + .uri("/v1/AUTH_project/container/object") + .body(()) + .expect("build Swift request"); + assert!(!uses_server_owned_s3_request_id(&swift_request, false)); + + let double_slash_request = Request::builder() + .uri("//v1/AUTH_project/container/object") + .body(()) + .expect("build double-slash request"); + assert!(uses_server_owned_s3_request_id(&double_slash_request, false)); + } + + #[cfg(feature = "swift")] + #[tokio::test] + async fn non_swift_v1_path_keeps_s3_request_id_contract() { + let capture = HeaderCaptureService::default(); + let mut service = ExternalRequestContextLayer::default().layer(capture); + let mut request = Request::builder() + .uri("/v1/not-a-swift-account/object") + .body(()) + .expect("build S3 request"); + request + .headers_mut() + .insert(REQUEST_ID_HEADER, HeaderValue::from_static("client-request-id")); + + let response = service.call(request).await.expect("S3 response"); + let response_request_id = response + .headers() + .get(REQUEST_ID_HEADER) + .and_then(|value| value.to_str().ok()) + .expect("S3 response request ID"); + + assert!(uuid::Uuid::parse_str(response_request_id).is_ok()); + assert_eq!( + response.headers().get(AMZ_REQUEST_ID).and_then(|value| value.to_str().ok()), + Some(response_request_id) + ); + } + + #[tokio::test] + async fn non_s3_response_preserves_handler_request_id() { + let capture = HeaderCaptureService::with_response_request_id("handler-request-id"); + let mut service = ExternalRequestContextLayer::default().layer(capture); + let mut request = Request::builder() + .uri("/rustfs/admin/v3/info") + .body(()) + .expect("build admin request"); + request + .headers_mut() + .insert(REQUEST_ID_HEADER, HeaderValue::from_static("client-request-id")); + + let response = service.call(request).await.expect("admin response"); + + assert_eq!( + response + .headers() + .get(REQUEST_ID_HEADER) + .and_then(|value| value.to_str().ok()), + Some("handler-request-id") + ); + assert!(!response.headers().contains_key(AMZ_REQUEST_ID)); + } + + #[tokio::test] + async fn non_s3_request_without_id_generates_only_x_request_id() { + let capture = HeaderCaptureService::default(); + let captured_context = capture.request_context(); + let mut service = ExternalRequestContextLayer::default().layer(capture); + let request = Request::builder() + .uri("/rustfs/admin/v3/info") + .body(()) + .expect("build admin request"); + + let response = service.call(request).await.expect("admin response"); + let response_request_id = response + .headers() + .get(REQUEST_ID_HEADER) + .and_then(|value| value.to_str().ok()) + .expect("generated admin request ID"); + + assert!(uuid::Uuid::parse_str(response_request_id).is_ok()); + assert!(!response.headers().contains_key(AMZ_REQUEST_ID)); + let context = captured_context + .lock() + .expect("captured request context") + .clone() + .expect("admin request context"); + assert_eq!(context.request_id, response_request_id); + } + #[derive(Clone, Default)] struct CountingHybridService { calls: Arc, @@ -3571,6 +4052,12 @@ mod tests { "https://allowed.com" ); assert_eq!(resp_headers.get(cors::response::ACCESS_CONTROL_ALLOW_CREDENTIALS).unwrap(), "true"); + let exposed = resp_headers + .get(cors::response::ACCESS_CONTROL_EXPOSE_HEADERS) + .and_then(|value| value.to_str().ok()) + .expect("exposed response headers"); + assert!(exposed.split(',').any(|header| header.trim() == "x-request-id")); + assert!(exposed.split(',').any(|header| header.trim() == "x-amz-request-id")); } #[test] @@ -3619,7 +4106,7 @@ mod tests { } #[test] - fn request_context_layer_preserves_upstream_s3_request_id() { + fn request_context_layer_does_not_mutate_upstream_s3_request_id() { let mut service = RequestContextLayer.layer(CaptureService); let request = Request::builder() .uri("/bucket/object") diff --git a/rustfs/src/server/rate_limit.rs b/rustfs/src/server/rate_limit.rs index bdd11e9f2..16d8b92e6 100644 --- a/rustfs/src/server/rate_limit.rs +++ b/rustfs/src/server/rate_limit.rs @@ -57,6 +57,7 @@ use crate::server::{ MINIO_HEALTH_CLUSTER_READ_PATH, MINIO_HEALTH_LIVE_PATH, MINIO_HEALTH_READY_PATH, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH, RPC_PREFIX, RemoteAddr, TONIC_PREFIX, has_path_prefix, is_admin_path, is_table_catalog_path, }; +use crate::storage_api::server::layer::request_context::RequestContext; use bytes::Bytes; use futures::future::{Either, Ready, ready}; use http::{HeaderMap, HeaderValue, Request, Response, StatusCode}; @@ -408,15 +409,10 @@ fn u64_header(value: u64) -> HeaderValue { type BoxError = Box; type BoxBody = http_body_util::combinators::UnsyncBoxBody; -/// Build the S3-style `429` rejection. The request id (already generated by -/// `SetRequestIdLayer`, which sits outside this layer) is echoed manually -/// because the rejection short-circuits below `PropagateRequestIdLayer`. -fn s3_too_many_requests_response(request_id: Option<&HeaderValue>, limit_rpm: u32, throttle: &ThrottleInfo) -> Response { - // The header may be client-supplied (SetRequestIdLayer only fills it when - // absent), so gate the XML interpolation on a UUID-safe charset instead of - // reflecting arbitrary bytes into the body. +/// Build the S3-style `429` rejection. The server-owned request ID is echoed +/// manually because the rejection short-circuits the inner response stack. +fn s3_too_many_requests_response(request_id: Option<&str>, limit_rpm: u32, throttle: &ThrottleInfo) -> Response { let request_id_xml = request_id - .and_then(|value| value.to_str().ok()) .filter(|id| !id.is_empty() && id.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'-')) .map(|id| format!("{id}")) .unwrap_or_default(); @@ -436,8 +432,8 @@ fn s3_too_many_requests_response(request_id: Option<&HeaderValue>, limit_rpm: u3 .headers_mut() .insert(http::header::CONTENT_TYPE, HeaderValue::from_static("application/xml")); apply_throttle_headers(response.headers_mut(), limit_rpm, throttle); - if let Some(id) = request_id { - response.headers_mut().insert(X_REQUEST_ID, id.clone()); + if let Some(id) = request_id.and_then(|id| HeaderValue::from_str(id).ok()) { + response.headers_mut().insert(X_REQUEST_ID, id); } response } @@ -559,7 +555,9 @@ fn rejected_response( "Request rejected by API rate limit" ); Either::Right(ready(Ok(s3_too_many_requests_response( - req.headers().get(X_REQUEST_ID), + req.extensions() + .get::() + .map(|context| context.request_id.as_str()), limit_rpm, throttle, )))) @@ -871,7 +869,15 @@ mod tests { } let mut req = request_from(ip(1), "/bucket/object"); - req.headers_mut().insert(X_REQUEST_ID, HeaderValue::from_static("req-123")); + req.headers_mut() + .insert(X_REQUEST_ID, HeaderValue::from_static("client-request-id")); + req.extensions_mut().insert(RequestContext { + request_id: "req-123".to_string(), + x_amz_request_id: "req-123".to_string(), + trace_id: None, + span_id: None, + start_time: Instant::now(), + }); let resp = service.call(req).await.expect("ok"); assert_eq!(resp.status(), StatusCode::TOO_MANY_REQUESTS); assert_eq!(resp.headers().get(http::header::RETRY_AFTER).and_then(|v| v.to_str().ok()), Some("1")); @@ -891,20 +897,31 @@ mod tests { } #[tokio::test] - async fn hostile_request_id_is_not_reflected_into_the_429_body() { + async fn hostile_request_id_does_not_override_the_429_request_id() { let mut service = service_with_quota(60, 1); let _ = service.call(request_from(ip(3), "/bucket/object")).await.expect("ok"); let mut req = request_from(ip(3), "/bucket/object"); req.headers_mut() .insert(X_REQUEST_ID, HeaderValue::from_static("evil")); + req.extensions_mut().insert(RequestContext { + request_id: "server-request-id".to_string(), + x_amz_request_id: "server-request-id".to_string(), + trace_id: None, + span_id: None, + start_time: Instant::now(), + }); let resp = service.call(req).await.expect("ok"); assert_eq!(resp.status(), StatusCode::TOO_MANY_REQUESTS); + assert_eq!( + resp.headers().get(X_REQUEST_ID).and_then(|value| value.to_str().ok()), + Some("server-request-id") + ); let body = resp.into_body().collect().await.expect("body").to_bytes(); let body = String::from_utf8_lossy(&body); assert!(!body.contains("evil"), "client-controlled request id must not be reflected: {body}"); - assert!(!body.contains(""), "malformed id must be omitted entirely: {body}"); + assert!(body.contains("server-request-id"), "body: {body}"); } #[tokio::test] diff --git a/rustfs/src/storage/helper.rs b/rustfs/src/storage/helper.rs index 768471e21..e2d01b13c 100644 --- a/rustfs/src/storage/helper.rs +++ b/rustfs/src/storage/helper.rs @@ -14,7 +14,7 @@ use crate::server::{convert_ecstore_object_info, is_audit_module_enabled, is_notify_module_enabled}; use crate::storage::access::{ReqInfo, request_context_from_req}; -use crate::storage::request_context::{RequestContext, extract_request_id_from_headers}; +use crate::storage::request_context::RequestContext; use crate::storage::storage_api::runtime_sources_consumer::runtime_sources; use hashbrown::HashMap; use http::StatusCode; @@ -73,6 +73,13 @@ where } } +/// Builds S3-compatible notification response elements with the canonical request ID. +pub(crate) fn build_event_resp_elements(response: &S3Response, request_id: &str) -> HashMap { + let mut resp_elements = extract_resp_elements(response); + resp_elements.insert(AMZ_REQUEST_ID.to_string(), request_id.to_string()); + resp_elements +} + /// A unified helper structure for building and distributing audit logs and event notifications via RAII mode at the end of an S3 operation scope. pub enum OperationHelper { Disabled, @@ -86,7 +93,7 @@ pub struct EnabledOperationHelper { api_builder: ApiDetailsBuilder, event_builder: Option, start_time: std::time::Instant, - request_context: Option, + request_context: RequestContext, } impl OperationHelper { @@ -157,16 +164,13 @@ impl OperationHelper { api_builder = api_builder.object(&object_key); } // Audit builder - // Resolve canonical request context and request_id in a single pass: - // RequestContext.request_id > extract_request_id_from_headers() > generated fallback id + // Resolve the canonical request context once for both output chains. let request_context = request_context_from_req(req); if request_context.is_none() { counter!("rustfs_log_chain_orphan_total", "component" => "operation_helper").increment(1); } - let request_id = request_context - .as_ref() - .map(|ctx| ctx.request_id.clone()) - .unwrap_or_else(|| extract_request_id_from_headers(&req.headers)); + let request_context = request_context.unwrap_or_else(|| RequestContext::from_external_headers(&req.headers)); + let request_id = request_context.request_id.clone(); let audit_builder = if audit_enabled { Some( @@ -189,12 +193,6 @@ impl OperationHelper { }; let mut req_params = extract_params_header(&req.headers); - // Inject x-amz-request-id from RequestContext into req_params for event correlation - if let Some(ref ctx) = request_context { - req_params - .entry(AMZ_REQUEST_ID.to_string()) - .or_insert_with(|| ctx.x_amz_request_id.clone()); - } if let Some(principal_id) = req_info .and_then(|info| info.cred.as_ref()) .map(|cred| cred.access_key.clone()) @@ -228,10 +226,7 @@ impl OperationHelper { audit_builder, api_builder, event_builder, - start_time: request_context - .as_ref() - .map(|ctx| ctx.start_time) - .unwrap_or_else(std::time::Instant::now), + start_time: request_context.start_time, request_context, })) } @@ -331,14 +326,12 @@ impl OperationHelper { } // Inject OpenTelemetry trace context into audit tags for distributed tracing correlation - if let Some(ref ctx) = state.request_context - && (ctx.trace_id.is_some() || ctx.span_id.is_some()) - { + if state.request_context.trace_id.is_some() || state.request_context.span_id.is_some() { let mut tags = HashMap::new(); - if let Some(ref tid) = ctx.trace_id { + if let Some(ref tid) = state.request_context.trace_id { tags.insert("traceId".to_string(), Value::String(tid.clone())); } - if let Some(ref sid) = ctx.span_id { + if let Some(ref sid) = state.request_context.span_id { tags.insert("spanId".to_string(), Value::String(sid.clone())); } final_builder = final_builder.tags(tags); @@ -352,7 +345,7 @@ impl OperationHelper { if state.notify_enabled && let (Some(builder), Ok(res)) = (state.event_builder.take(), result) { - state.event_builder = Some(builder.resp_elements(extract_resp_elements(res))); + state.event_builder = Some(builder.resp_elements(build_event_resp_elements(res, &state.request_context.request_id))); } self @@ -365,6 +358,17 @@ impl OperationHelper { } self } + + /// Returns the operation's canonical context, reading the ingress extension + /// when the disabled fast path holds no request state. + pub(crate) fn request_context_or_from_request(&self, req: &S3Request) -> RequestContext { + match self { + Self::Enabled(state) => state.request_context.clone(), + Self::Disabled => { + request_context_from_req(req).unwrap_or_else(|| RequestContext::from_external_headers(&req.headers)) + } + } + } } fn should_build_notification_event(notify_module_enabled: bool) -> bool { @@ -381,7 +385,7 @@ impl Drop for OperationHelper { if state.audit_enabled && let Some(builder) = state.audit_builder.take() { - let ctx = state.request_context.clone(); + let ctx = Some(state.request_context.clone()); spawn_background_with_context(ctx, async move { AuditLogger::log(builder.build()).await; }); @@ -395,7 +399,7 @@ impl Drop for OperationHelper { let event_args = builder.build(); // Avoid generating notifications for copy requests if !event_args.is_replication_request() { - let ctx = state.request_context.clone(); + let ctx = Some(state.request_context.clone()); spawn_background_with_context(ctx, async move { runtime_sources::current_notify_interface().notify(event_args).await; }); @@ -416,8 +420,9 @@ mod tests { use rustfs_credentials::Credentials; use rustfs_s3_ops::S3Operation; use rustfs_s3_types::EventName; - use s3s::S3Request; - use s3s::dto::DeleteObjectTaggingInput; + use rustfs_utils::http::headers::{AMZ_REQUEST_ID, REQUEST_ID_HEADER}; + use s3s::dto::{DeleteObjectTaggingInput, DeleteObjectTaggingOutput}; + use s3s::{S3Request, S3Response}; use std::sync::{Arc, Mutex}; use temp_env::with_vars; @@ -554,11 +559,14 @@ mod tests { let mut req = build_request(input, Method::DELETE, Uri::from_static("/test-bucket/test-key")); req.headers.insert("host", HeaderValue::from_static("example.com")); req.headers.insert("user-agent", HeaderValue::from_static("rustfs-test")); + req.headers + .insert(REQUEST_ID_HEADER, HeaderValue::from_static("ingress-canonical-uuid")); + req.headers + .insert("x-amz-request-id", HeaderValue::from_static("client-supplied-request-id")); - // Insert RequestContext (set by ingress layer) with a specific request_id req.extensions.insert(RequestContext { request_id: "ingress-canonical-uuid".to_string(), - x_amz_request_id: "ingress-canonical-uuid".to_string(), + x_amz_request_id: "client-supplied-request-id".to_string(), trace_id: None, span_id: None, start_time: std::time::Instant::now(), @@ -570,20 +578,32 @@ mod tests { ..Default::default() }); - let helper = OperationHelper::new(&req, EventName::ObjectAccessedGet, S3Operation::GetObject); + let result = Ok(S3Response::new(DeleteObjectTaggingOutput::default())); + let mut helper = + OperationHelper::new(&req, EventName::ObjectAccessedGet, S3Operation::GetObject).complete(&result); // Verify the helper stored the RequestContext - let OperationHelper::Enabled(state) = &helper else { + let OperationHelper::Enabled(state) = &mut helper else { panic!("helper should be enabled when notify/audit switches are on"); }; - assert!(state.request_context.is_some()); - assert_eq!(state.request_context.as_ref().unwrap().request_id, "ingress-canonical-uuid"); + assert_eq!(state.request_context.request_id, "ingress-canonical-uuid"); + let audit_entry = state.audit_builder.take().expect("audit builder should exist").build(); + assert_eq!(audit_entry.request_id.as_deref(), Some("ingress-canonical-uuid")); + let event_args = state.event_builder.clone().expect("event builder should exist").build(); + assert_eq!( + event_args.req_params.get(AMZ_REQUEST_ID).map(String::as_str), + Some("client-supplied-request-id") + ); + assert_eq!( + event_args.resp_elements.get(AMZ_REQUEST_ID).map(String::as_str), + Some("ingress-canonical-uuid") + ); }, ); } #[test] - fn operation_helper_no_request_context_when_absent() { + fn operation_helper_reuses_generated_context_when_headers_absent() { with_vars( [ (rustfs_config::ENV_NOTIFY_ENABLE, Some("true")), @@ -601,8 +621,6 @@ mod tests { let mut req = build_request(input, Method::DELETE, Uri::from_static("/test-bucket/test-key")); req.headers.insert("host", HeaderValue::from_static("example.com")); req.headers.insert("user-agent", HeaderValue::from_static("rustfs-test")); - req.headers - .insert("x-amz-request-id", HeaderValue::from_static("amz-header-uuid")); // No RequestContext inserted req.extensions.insert(ReqInfo { @@ -612,12 +630,21 @@ mod tests { }); let helper = OperationHelper::new(&req, EventName::ObjectAccessedGet, S3Operation::GetObject); - - // Verify the helper has no RequestContext + let request_context = helper.request_context_or_from_request(&req); + let request_id = request_context.request_id; + assert!(uuid::Uuid::parse_str(&request_id).is_ok()); + let result = Ok(S3Response::new(DeleteObjectTaggingOutput::default())); + let helper = helper.complete(&result); let OperationHelper::Enabled(state) = &helper else { panic!("helper should be enabled when notify/audit switches are on"); }; - assert!(state.request_context.is_none()); + assert_eq!(state.request_context.request_id, request_id); + let event_args = state.event_builder.clone().expect("event builder should exist").build(); + assert!(!event_args.req_params.contains_key(AMZ_REQUEST_ID)); + assert_eq!( + event_args.resp_elements.get(AMZ_REQUEST_ID).map(String::as_str), + Some(request_id.as_str()) + ); }, ); } @@ -638,10 +665,20 @@ mod tests { .key("test-key".to_string()) .build() .unwrap(); - let req = build_request(input, Method::DELETE, Uri::from_static("/test-bucket/test-key")); + let mut req = build_request(input, Method::DELETE, Uri::from_static("/test-bucket/test-key")); + req.headers + .insert(REQUEST_ID_HEADER, HeaderValue::from_static("client-request-id")); + req.extensions.insert(RequestContext { + request_id: "server-request-id".to_string(), + x_amz_request_id: "server-request-id".to_string(), + trace_id: None, + span_id: None, + start_time: std::time::Instant::now(), + }); let helper = OperationHelper::new(&req, EventName::ObjectAccessedGet, S3Operation::GetObject); - assert!(matches!(helper, OperationHelper::Disabled)); + assert!(matches!(&helper, OperationHelper::Disabled)); + assert_eq!(helper.request_context_or_from_request(&req).request_id, "server-request-id"); }, ); } diff --git a/rustfs/src/storage/request_context.rs b/rustfs/src/storage/request_context.rs index e096a332a..6ef750c8f 100644 --- a/rustfs/src/storage/request_context.rs +++ b/rustfs/src/storage/request_context.rs @@ -17,11 +17,10 @@ //! # Architecture //! //! ```text -//! HTTP Ingress (SetRequestIdLayer) -//! → generates x-request-id UUID -//! → RequestContextLayer creates RequestContext +//! External S3 HTTP ingress +//! → generates a server-owned request ID without mutating signed headers +//! → ExternalRequestContextLayer creates RequestContext //! → stores in request.extensions() -//! → stores the S3-compatible request-id alias without changing signed headers //! Auth (FS::check) //! → copies RequestContext into ReqInfo.request_context //! Storage (FS methods) @@ -40,10 +39,11 @@ //! # Frozen Rules (T00 Guardrails) //! //! ## request-id contract -//! - Canonical wire header: `x-request-id` (set by `SetRequestIdLayer`) +//! - External S3 response headers: `x-request-id` and `x-amz-request-id` +//! - Non-S3 response header: propagated `x-request-id` //! - Compatibility wire header: `x-amz-request-id` //! - Canonical internal field: `RequestContext.request_id` -//! - S3 compatibility internal alias field: `RequestContext.x_amz_request_id` +//! - Client-provided request ID headers are never canonical on external S3 requests //! - Internal modules MUST NOT generate a second request id under the field name `request_id` //! except for orphan/non-ingress fallback paths where no canonical request-id exists. //! - Internal identifiers for sub-operations should use `operation_id` or `subtask_id` @@ -72,10 +72,14 @@ use tracing_opentelemetry::OpenTelemetrySpanExt; /// Created exactly once at HTTP ingress. Cloned by value; never mutated after creation. #[derive(Clone, Debug)] pub struct RequestContext { - /// Canonical request ID (from `x-request-id` header, set by `SetRequestIdLayer`). + /// Canonical request ID: server-owned for external S3 requests and + /// propagated for non-S3 and trusted internal requests. pub request_id: String, - /// S3-compatible request ID alias (preserves upstream `x-amz-request-id` if present, - /// otherwise equals `request_id`). + /// Compatibility-only alias that preserves an incoming + /// `x-amz-request-id`, or mirrors [`Self::request_id`] when absent. + /// + /// Internal correlation must use [`Self::request_id`]. This field remains + /// for compatibility with existing in-crate consumers. pub x_amz_request_id: String, /// OpenTelemetry trace ID (if present from upstream propagation). pub trace_id: Option, @@ -86,6 +90,54 @@ pub struct RequestContext { } impl RequestContext { + /// Create a context for a trusted internal request that may propagate its + /// canonical ID through headers. + pub(crate) fn from_headers(headers: &HeaderMap) -> Self { + Self::new( + extract_request_id_from_headers(headers), + headers, + extract_trace_context_ids_from_headers(headers), + ) + } + + /// Create a context from propagated request headers without copying trace + /// state into the request context. + pub(crate) fn from_headers_without_trace_context(headers: &HeaderMap) -> Self { + let request_id = extract_request_id_from_headers(headers); + Self { + x_amz_request_id: request_id.clone(), + request_id, + trace_id: None, + span_id: None, + start_time: Instant::now(), + } + } + + /// Create an external request context with a server-owned ID while keeping + /// client headers unchanged for signature verification. + pub(crate) fn from_external_headers(headers: &HeaderMap) -> Self { + Self::new(uuid::Uuid::new_v4().to_string(), headers, extract_trace_context_ids_from_headers(headers)) + } + + fn new(request_id: String, headers: &HeaderMap, trace_context: Option<(String, String)>) -> Self { + let x_amz_request_id = headers + .get(AMZ_REQUEST_ID) + .and_then(|value| value.to_str().ok()) + .map(String::from) + .unwrap_or_else(|| request_id.clone()); + let (trace_id, span_id) = trace_context + .map(|(trace_id, span_id)| (Some(trace_id), Some(span_id))) + .unwrap_or((None, None)); + + Self { + request_id, + x_amz_request_id, + trace_id, + span_id, + start_time: Instant::now(), + } + } + /// Create a fallback `RequestContext` for paths that bypass HTTP ingress. /// Generates a canonical internal `request_id` in `trace-{trace_id}` or `req-{uuid}` format. pub fn fallback() -> Self { @@ -206,7 +258,7 @@ where #[allow(unused_imports)] mod tests { use super::{RequestContext, extract_request_id_from_headers, extract_trace_context_ids_from_headers}; - use http::HeaderMap; + use http::{HeaderMap, HeaderValue}; use opentelemetry::global; use opentelemetry::trace::{SpanContext, TraceContextExt, TraceFlags, TraceId, TraceState, TracerProvider as _}; use opentelemetry_sdk::propagation::TraceContextPropagator; @@ -252,6 +304,64 @@ mod tests { assert!(ctx.span_id.is_none()); } + #[test] + fn test_request_context_from_headers_prioritizes_canonical_request_id() { + let mut headers = HeaderMap::new(); + headers.insert("x-request-id", HeaderValue::from_static("canonical-request-id")); + headers.insert("x-amz-request-id", HeaderValue::from_static("client-request-id")); + + let ctx = RequestContext::from_headers(&headers); + + assert_eq!(ctx.request_id, "canonical-request-id"); + assert_eq!(ctx.x_amz_request_id, "client-request-id"); + } + + #[test] + fn test_request_context_from_headers_preserves_empty_amz_alias() { + let mut headers = HeaderMap::new(); + headers.insert("x-request-id", HeaderValue::from_static("canonical-request-id")); + headers.insert("x-amz-request-id", HeaderValue::from_static("")); + + let ctx = RequestContext::from_headers(&headers); + + assert_eq!(ctx.request_id, "canonical-request-id"); + assert_eq!(ctx.x_amz_request_id, ""); + } + + #[test] + fn test_propagated_request_context_mirrors_canonical_request_id() { + let mut headers = HeaderMap::new(); + headers.insert("x-request-id", HeaderValue::from_static("canonical-request-id")); + headers.insert("x-amz-request-id", HeaderValue::from_static("untrusted-amz-request-id")); + + let ctx = RequestContext::from_headers_without_trace_context(&headers); + + assert_eq!(ctx.request_id, "canonical-request-id"); + assert_eq!(ctx.x_amz_request_id, "canonical-request-id"); + assert!(ctx.trace_id.is_none()); + assert!(ctx.span_id.is_none()); + } + + #[test] + fn test_external_request_context_owns_id_and_preserves_trace_context() { + global::set_text_map_propagator(TraceContextPropagator::new()); + let mut headers = HeaderMap::new(); + headers.insert("x-request-id", HeaderValue::from_static("client-request-id")); + headers.insert("x-amz-request-id", HeaderValue::from_static("client-amz-request-id")); + headers.insert( + "traceparent", + HeaderValue::from_static("00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"), + ); + + let ctx = RequestContext::from_external_headers(&headers); + + assert_ne!(ctx.request_id, "client-request-id"); + assert!(uuid::Uuid::parse_str(&ctx.request_id).is_ok()); + assert_eq!(ctx.x_amz_request_id, "client-amz-request-id"); + assert_eq!(ctx.trace_id.as_deref(), Some("4bf92f3577b34da6a3ce929d0e0e4736")); + assert_eq!(ctx.span_id.as_deref(), Some("00f067aa0ba902b7")); + } + #[test] fn test_request_context_fallback_uses_trace_prefix_when_span_context_valid() { let trace_id = "70f5f77e2f0a4f24be343b59f8b66f8f"; diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 9c891d9f6..296636264 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -174,7 +174,7 @@ pub(crate) mod head_prefix_consumer { } pub(crate) mod helper_consumer { - pub(crate) use super::super::helper::{OperationHelper, spawn_background_with_context}; + pub(crate) use super::super::helper::{OperationHelper, build_event_resp_elements, spawn_background_with_context}; pub(crate) type StorageObjectInfo = super::StorageObjectInfo; } @@ -203,9 +203,7 @@ pub(crate) mod options_consumer { } pub(crate) mod request_context_consumer { - pub(crate) use super::super::request_context::{ - RequestContext, extract_request_id_from_headers, extract_trace_context_ids_from_headers, spawn_traced, - }; + pub(crate) use super::super::request_context::{RequestContext, extract_request_id_from_headers, spawn_traced}; } pub(crate) mod rpc_consumer { diff --git a/rustfs/src/storage_api.rs b/rustfs/src/storage_api.rs index 13c4e638a..b10ead5f2 100644 --- a/rustfs/src/storage_api.rs +++ b/rustfs/src/storage_api.rs @@ -124,9 +124,7 @@ pub(crate) mod server { } pub(crate) mod request_context { - pub(crate) use crate::storage::storage_api::request_context_consumer::{ - RequestContext, extract_request_id_from_headers, - }; + pub(crate) use crate::storage::storage_api::request_context_consumer::RequestContext; } pub(crate) mod rpc { @@ -149,7 +147,7 @@ pub(crate) mod server { pub(crate) mod request_context { pub(crate) use crate::storage::storage_api::request_context_consumer::{ - RequestContext, extract_request_id_from_headers, extract_trace_context_ids_from_headers, spawn_traced, + RequestContext, extract_request_id_from_headers, spawn_traced, }; } }