test(ecstore): fix rename_all WARN flake from callsite-interest poisoning

rename_all_missing_source_still_warns and rename_all_real_failure_still_warns
assert that `warn_reliable_rename_failure` emitted its WARN, but that is a
single production callsite shared with tests that call rename_all *without*
installing a subscriber — rename_all_missing_source_returns_file_not_found,
two tests above, is one of them.

tracing caches each callsite's Interest process-globally and the first thread
to reach a callsite fixes that value; while at most one dispatcher is
registered, tracing-core derives it from the registering thread's own
subscriber, and registration is once-only. When the subscriber-less sibling
wins, the callsite is cached as Interest::never() and the WARN never fires,
so the assertion sees empty output:

    ordinary missing-source failures must keep the WARN, got:

Reproduced at 3/25 with `disk::os::tests::rename_all_missing_source` (both
tests), against 0/20 for the victim alone. Fixed by pinning callsite interest
inside warn_capture(), so every current and future user of that helper is
covered rather than just the two tests that happen to fail today.

pin_callsite_interest_for_test() moves from cluster::rpc::background_monitor
to a new crate-level test_tracing module: it is domain-neutral and now has
consumers in two unrelated subsystems, and disk::os should not have to reach
into a cluster::rpc test helper.

Verified: repro filter 0/30 (was 3/25); disk::os:: 0/12; cluster::rpc:: 0/12
and its poisoner pair 0/15, confirming the moved helper still holds.

Follow-up to #5438. Closes the last item in #5439.

Cargo.lock

@@ -9212,6 +9212,7 @@ dependencies = [
  "tonic",
  "tower",
  "tracing",
+ "tracing-core",
  "tracing-opentelemetry",
  "tracing-subscriber",
  "url",

Cargo.toml

@@ -303,6 +303,7 @@ test-case = "3.3.1"
 thiserror = "2.0.19"
 tracing = { version = "0.1.44" }
 tracing-appender = "0.2.5"
+tracing-core = "0.1.36"
 tracing-error = "0.2.1"
 tracing-opentelemetry = { version = "0.33" }
 tracing-subscriber = { version = "0.3.23" }

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<dyn Error + Send + Sync>;
 /// 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>, name: "webhook" }`,
 /// so the ARN a notification rule references is
@@ -579,6 +587,36 @@ fn trimmed_etag(value: Option<&str>) -> Option<String> {
     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];

crates/ecstore/Cargo.toml

@@ -158,6 +158,9 @@ tokio = { workspace = true, features = ["rt-multi-thread", "macros", "test-util"
 criterion = { workspace = true, features = ["html_reports"] }
 temp-env = { workspace = true, features = ["async_closure"] }
 tracing-subscriber = { workspace = true, features = ["json", "env-filter", "time"] }
+# Only for `pin_callsite_interest_for_test`, which registers a `NoSubscriber`
+# dispatcher to keep tracing's process-global callsite-interest cache honest.
+tracing-core = { workspace = true }
 serial_test = { workspace = true }
 opentelemetry_sdk = { workspace = true, features = ["rt-tokio"] }
 proptest = "1"

crates/ecstore/src/client/provider_versions.rs

@@ -172,7 +172,7 @@ impl ProviderVersionCapabilities {
     }
 }
 
-fn validate_remote_version_id(version_id: &str) -> Result<(), Error> {
+pub(crate) fn validate_remote_version_id(version_id: &str) -> Result<(), Error> {
     if version_id.is_empty() {
         return Err(Error::new(
             ErrorKind::InvalidData,

crates/ecstore/src/cluster/rpc/peer_rest_client.rs

@@ -2741,6 +2741,11 @@ mod tests {
                 .with_span_list(true),
         );
         let _guard = tracing::subscriber::set_default(subscriber);
+        // The `recovery-monitor` callsite is shared with the production
+        // `mark_offline_and_spawn_recovery` path that sibling tests exercise from
+        // subscriber-less threads; without this the span can be cached as
+        // `Interest::never()` and silently degrade to `Span::none()`.
+        let _callsite_pin = crate::test_tracing::pin_callsite_interest_for_test();
 
         let client = test_peer_client();
         let span = tracing::info_span!("request-span", request_id = "req-peer-rest");

crates/ecstore/src/cluster/rpc/remote_disk.rs

@@ -3005,6 +3005,7 @@ mod tests {
     use crate::cluster::rpc::internode_data_transport::{InternodeDataTransportCapabilities, TcpHttpInternodeDataTransport};
     use crate::runtime::sources as runtime_sources;
     use serde_json::Value;
+    use serial_test::serial;
     use std::io::{self as std_io, Write};
     use std::pin::Pin;
     use std::sync::{Arc, Mutex, Mutex as StdMutex, Once};
@@ -3018,6 +3019,20 @@ mod tests {
 
     static INIT: Once = Once::new();
 
+    // `#[serial(internode_metrics)]` marks every test that observes
+    // `global_internode_metrics()`. Those counters are a process-wide singleton:
+    // some of these tests snapshot a counter, run one decode, and assert on the
+    // delta, while others deliberately record decode errors or call
+    // `reset_internode_metrics_for_test()`. Run concurrently in one process they
+    // corrupt each other's deltas — a sibling's error bumps the "no decode error"
+    // assertion off zero, and a sibling's reset can drive an `after > before`
+    // assertion backwards.
+    //
+    // The marker only takes effect under the `cargo test` fallback; nextest
+    // already isolates each test in its own process, so every test there gets its
+    // own copy of the counters (see `docs/testing/README.md`). Any new test that
+    // reads or mutates the global internode metrics belongs in this group.
+
     #[test]
     fn snapshot_lease_response_requires_current_protocol_and_valid_token() {
         let token = SnapshotLeaseToken::new();
@@ -3306,6 +3321,7 @@ mod tests {
     }
 
     #[test]
+    #[serial(internode_metrics)]
     fn read_multiple_response_decode_prefers_msgpack_payloads() {
         let endpoint = sample_remote_endpoint();
         let msgpack_resp = sample_read_multiple_resp("msgpack", b"binary");
@@ -3328,6 +3344,7 @@ mod tests {
     }
 
     #[test]
+    #[serial(internode_metrics)]
     fn read_multiple_response_decode_falls_back_to_json_payloads() {
         let endpoint = sample_remote_endpoint();
         let json_resp = sample_read_multiple_resp("json", b"fallback");
@@ -3349,6 +3366,7 @@ mod tests {
     }
 
     #[test]
+    #[serial(internode_metrics)]
     fn rename_data_response_accepts_legacy_json_without_decode_error() {
         crate::cluster::rpc::runtime_sources::reset_internode_metrics_for_test();
         let response = RenameDataResp {
@@ -3500,6 +3518,7 @@ mod tests {
     }
 
     #[test]
+    #[serial(internode_metrics)]
     fn read_multiple_response_decode_reports_corrupt_msgpack_item() {
         let endpoint = sample_remote_endpoint();
         let response = ReadMultipleResponse {
@@ -3525,6 +3544,7 @@ mod tests {
     }
 
     #[test]
+    #[serial(internode_metrics)]
     fn read_multiple_response_decode_reports_corrupt_json_item() {
         let endpoint = sample_remote_endpoint();
         let response = ReadMultipleResponse {
@@ -3565,6 +3585,7 @@ mod tests {
     }
 
     #[test]
+    #[serial(internode_metrics)]
     fn batch_read_version_response_decode_prefers_msgpack_payloads() {
         let endpoint = sample_remote_endpoint();
         let msgpack_resp = sample_batch_read_version_resp(7, "msgpack-object", true);
@@ -3585,6 +3606,7 @@ mod tests {
     }
 
     #[test]
+    #[serial(internode_metrics)]
     fn batch_read_version_response_rejects_invalid_success_metadata() {
         let endpoint = sample_remote_endpoint();
         let mut response_item = sample_batch_read_version_resp(0, "invalid-object", true);
@@ -3604,6 +3626,7 @@ mod tests {
     }
 
     #[test]
+    #[serial(internode_metrics)]
     fn batch_read_version_response_decode_reports_corrupt_msgpack_item() {
         let endpoint = sample_remote_endpoint();
         let response = BatchReadVersionResponse {
@@ -3630,6 +3653,7 @@ mod tests {
     }
 
     #[test]
+    #[serial(internode_metrics)]
     fn batch_read_version_response_decode_reports_corrupt_json_item() {
         let endpoint = sample_remote_endpoint();
         let response = BatchReadVersionResponse {
@@ -4419,6 +4443,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial(internode_metrics)]
     async fn test_remote_disk_create_file_retries_once_on_retryable_open_write_error() {
         let transport = RetryingOpenWriteInternodeDataTransport::with_steps(vec![
             OpenWriteTestStep::Error(DiskError::from(rustfs_rio::new_test_internode_http_io_error(
@@ -4457,6 +4482,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial(internode_metrics)]
     async fn test_remote_disk_read_file_stream_retries_once_on_retryable_open_read_error() {
         // A transient reset-by-peer on a shard read during the read-after-write window must be
         // absorbed by one re-dial rather than eroding read quorum (issue #2761).
@@ -5306,6 +5332,11 @@ mod tests {
                 .with_span_list(true),
         );
         let _guard = tracing::subscriber::set_default(subscriber);
+        // The `recovery-monitor` span and the monitor's own log events are
+        // production callsites that sibling tests exercise from subscriber-less
+        // threads; without this they can be cached as `Interest::never()` and go
+        // silently missing here.
+        let _callsite_pin = crate::test_tracing::pin_callsite_interest_for_test();
 
         let endpoint = Endpoint {
             url: url::Url::parse("http://127.0.0.1:59996/data").expect("endpoint URL should parse"),
@@ -5360,6 +5391,11 @@ mod tests {
                 .with_span_list(true),
         );
         let _guard = tracing::subscriber::set_default(subscriber);
+        // The `recovery-monitor` span and the monitor's own log events are
+        // production callsites that sibling tests exercise from subscriber-less
+        // threads; without this they can be cached as `Interest::never()` and go
+        // silently missing here.
+        let _callsite_pin = crate::test_tracing::pin_callsite_interest_for_test();
 
         let addr = "http://127.0.0.1:59997".to_string();
         let endpoint = Endpoint {

crates/ecstore/src/disk/os.rs

@@ -771,9 +771,26 @@ mod tests {
         }
     }
 
+    /// Holds a `warn_capture()` capture alive: the thread-local subscriber, plus
+    /// the pin that keeps tracing's process-global callsite-interest cache from
+    /// being decided by some other test's thread.
+    struct WarnCaptureGuard {
+        _subscriber: tracing::subscriber::DefaultGuard,
+        _callsite_pin: tracing::Dispatch,
+    }
+
     /// Capture WARN-level output on the current thread; tokio tests here run on
     /// the current-thread runtime, so the guard covers the whole test body.
-    fn warn_capture() -> (CapturedLogs, tracing::subscriber::DefaultGuard) {
+    ///
+    /// The callsite pin matters because `warn_reliable_rename_failure` is a
+    /// single production callsite shared with tests that call `rename_all`
+    /// *without* installing a subscriber — `rename_all_missing_source_returns_file_not_found`
+    /// is one. Whichever thread reaches it first fixes its `Interest`
+    /// process-wide, so without the pin that sibling can cache
+    /// `Interest::never()` and the WARN never fires here at all, leaving the
+    /// "must keep the WARN" assertions staring at empty output. See
+    /// [`crate::test_tracing::pin_callsite_interest_for_test`].
+    fn warn_capture() -> (CapturedLogs, WarnCaptureGuard) {
         let logs = CapturedLogs::default();
         let subscriber = tracing_subscriber::fmt()
             .with_max_level(tracing::Level::WARN)
@@ -781,7 +798,10 @@ mod tests {
             .with_ansi(false)
             .without_time()
             .finish();
-        let guard = tracing::subscriber::set_default(subscriber);
+        let guard = WarnCaptureGuard {
+            _subscriber: tracing::subscriber::set_default(subscriber),
+            _callsite_pin: crate::test_tracing::pin_callsite_interest_for_test(),
+        };
         (logs, guard)
     }
 

crates/ecstore/src/lib.rs

@@ -93,3 +93,6 @@ pub(crate) mod ecstore_validation_blackbox;
 
 #[cfg(test)]
 pub(crate) mod test_metrics;
+
+#[cfg(test)]
+pub(crate) mod test_tracing;

crates/ecstore/src/services/tier/warm_backend_s3.rs

@@ -29,6 +29,7 @@ use crate::client::{
     api_remove::{RemoveObjectOptions, RemoveObjectResult},
     api_s3_datatypes::ListVersionsResult,
     credentials::{Credentials, SignatureType, Static, Value},
+    provider_versions::validate_remote_version_id,
     transition_api::{BucketLookupType, Options, TransitionClient, TransitionCore},
     transition_api::{ReadCloser, ReaderImpl},
 };
@@ -189,44 +190,102 @@ impl WarmBackendS3 {
         remote_bucket_versioning_from_status(config.status.as_ref().map(|status| status.as_str()))
     }
 
-    async fn list_transition_candidate_versions(&self, object: &str) -> Result<ListVersionsResult, std::io::Error> {
+    async fn probe_transition_candidate_versions(
+        &self,
+        object: &str,
+        bucket_versioning: RemoteBucketVersioning,
+    ) -> Result<TransitionCandidateProbe, std::io::Error> {
+        let remote_object = self.get_dest(object);
         let mut opts = ListObjectsOptions::default();
-        opts.set("prefix", &self.get_dest(object));
-        opts.set("max-keys", "2");
-        self.client.list_object_versions_query(&self.bucket, &opts, "", "", "").await
+        opts.set("prefix", &remote_object);
+        opts.set("max-keys", "1000");
+
+        let mut key_marker = String::new();
+        let mut version_id_marker = String::new();
+        let mut candidates = TransitionCandidateVersions::default();
+        loop {
+            let versions = self
+                .client
+                .list_object_versions_query(&self.bucket, &opts, &key_marker, &version_id_marker, "")
+                .await?;
+            candidates.extend(&remote_object, &versions);
+            if candidates.is_ambiguous() {
+                return Ok(TransitionCandidateProbe::Ambiguous);
+            }
+            if !versions.is_truncated {
+                return classify_transition_candidates(candidates, bucket_versioning);
+            }
+
+            advance_version_markers(&mut key_marker, &mut version_id_marker, &versions)?;
+        }
     }
 }
 
-fn classify_transition_candidate_versions(
-    remote_object: &str,
+fn classify_transition_candidates(
+    candidates: TransitionCandidateVersions,
     bucket_versioning: RemoteBucketVersioning,
+) -> Result<TransitionCandidateProbe, std::io::Error> {
+    let probe = candidates.classify(bucket_versioning);
+    if let TransitionCandidateProbe::VersionedPresent(version_id) = &probe {
+        validate_remote_version_id(version_id)?;
+    }
+    Ok(probe)
+}
+
+fn advance_version_markers(
+    key_marker: &mut String,
+    version_id_marker: &mut String,
     versions: &ListVersionsResult,
-) -> TransitionCandidateProbe {
-    if versions.is_truncated {
-        return TransitionCandidateProbe::Ambiguous;
+) -> Result<(), std::io::Error> {
+    let next_markers = (&versions.next_key_marker, &versions.next_version_id_marker);
+    if next_markers == (&*key_marker, &*version_id_marker) {
+        return Err(std::io::Error::new(
+            std::io::ErrorKind::InvalidData,
+            "ListObjectVersions pagination markers did not advance",
+        ));
     }
+    key_marker.clone_from(&versions.next_key_marker);
+    version_id_marker.clone_from(&versions.next_version_id_marker);
+    Ok(())
+}
 
-    if versions.delete_markers.iter().any(|marker| marker.key == remote_object) {
-        return TransitionCandidateProbe::Ambiguous;
-    }
+#[derive(Default)]
+struct TransitionCandidateVersions {
+    version_id: Option<String>,
+    ambiguous: bool,
+}
 
-    let mut exact_versions = versions.versions.iter().filter(|version| version.key == remote_object);
-    let Some(version) = exact_versions.next() else {
-        return TransitionCandidateProbe::Missing;
-    };
-    if exact_versions.next().is_some() {
-        return TransitionCandidateProbe::Ambiguous;
-    }
-
-    match bucket_versioning {
-        RemoteBucketVersioning::Disabled => TransitionCandidateProbe::UnversionedPresent,
-        RemoteBucketVersioning::Suspended if version.version_id == "null" => {
-            TransitionCandidateProbe::VersionedPresent(version.version_id.clone())
+impl TransitionCandidateVersions {
+    fn extend(&mut self, remote_object: &str, versions: &ListVersionsResult) {
+        for version in versions.versions.iter().filter(|version| version.key == remote_object) {
+            if self.version_id.is_some() {
+                self.ambiguous = true;
+                return;
+            }
+            self.version_id = Some(version.version_id.clone());
         }
-        RemoteBucketVersioning::Suspended | RemoteBucketVersioning::Enabled if !version.version_id.is_empty() => {
-            TransitionCandidateProbe::VersionedPresent(version.version_id.clone())
+    }
+
+    fn is_ambiguous(&self) -> bool {
+        self.ambiguous
+    }
+
+    fn classify(self, bucket_versioning: RemoteBucketVersioning) -> TransitionCandidateProbe {
+        if self.ambiguous {
+            return TransitionCandidateProbe::Ambiguous;
+        }
+        let Some(version_id) = self.version_id else {
+            return TransitionCandidateProbe::Missing;
+        };
+
+        match bucket_versioning {
+            RemoteBucketVersioning::Disabled => TransitionCandidateProbe::UnversionedPresent,
+            RemoteBucketVersioning::Suspended if version_id == "null" => TransitionCandidateProbe::VersionedPresent(version_id),
+            RemoteBucketVersioning::Suspended | RemoteBucketVersioning::Enabled if !version_id.is_empty() => {
+                TransitionCandidateProbe::VersionedPresent(version_id)
+            }
+            RemoteBucketVersioning::Suspended | RemoteBucketVersioning::Enabled => TransitionCandidateProbe::Ambiguous,
         }
-        RemoteBucketVersioning::Suspended | RemoteBucketVersioning::Enabled => TransitionCandidateProbe::Ambiguous,
     }
 }
 
@@ -275,74 +334,109 @@ mod tests {
         }
     }
 
+    fn classify_pages(bucket_versioning: RemoteBucketVersioning, pages: &[ListVersionsResult]) -> TransitionCandidateProbe {
+        let mut candidates = TransitionCandidateVersions::default();
+        for page in pages {
+            candidates.extend("archive/object", page);
+        }
+        candidates.classify(bucket_versioning)
+    }
+
     #[test]
     fn transition_candidate_probe_classifier_is_fail_closed() {
         assert_eq!(
-            classify_transition_candidate_versions(
-                "archive/object",
-                RemoteBucketVersioning::Disabled,
-                &list_versions(&[], &[], false),
-            ),
+            classify_pages(RemoteBucketVersioning::Disabled, &[list_versions(&[], &[], false)],),
             TransitionCandidateProbe::Missing
         );
         assert_eq!(
-            classify_transition_candidate_versions(
-                "archive/object",
-                RemoteBucketVersioning::Disabled,
-                &list_versions(&[("archive/object", "")], &[], false),
-            ),
+            classify_pages(RemoteBucketVersioning::Disabled, &[list_versions(&[("archive/object", "")], &[], false)],),
             TransitionCandidateProbe::UnversionedPresent
         );
         assert_eq!(
-            classify_transition_candidate_versions(
-                "archive/object",
+            classify_pages(
                 RemoteBucketVersioning::Enabled,
-                &list_versions(&[("archive/object", "version-a")], &[], false),
+                &[list_versions(&[("archive/object", "version-a")], &[], false)],
             ),
             TransitionCandidateProbe::VersionedPresent("version-a".to_string())
         );
         assert_eq!(
-            classify_transition_candidate_versions(
-                "archive/object",
+            classify_pages(
                 RemoteBucketVersioning::Suspended,
-                &list_versions(&[("archive/object", "null")], &[], false),
+                &[list_versions(&[("archive/object", "null")], &[], false)],
             ),
             TransitionCandidateProbe::VersionedPresent("null".to_string())
         );
         assert_eq!(
-            classify_transition_candidate_versions(
-                "archive/object",
-                RemoteBucketVersioning::Enabled,
-                &list_versions(&[("archive/object", "")], &[], false),
-            ),
+            classify_pages(RemoteBucketVersioning::Enabled, &[list_versions(&[("archive/object", "")], &[], false)],),
             TransitionCandidateProbe::Ambiguous
         );
         assert_eq!(
-            classify_transition_candidate_versions(
-                "archive/object",
+            classify_pages(
                 RemoteBucketVersioning::Enabled,
-                &list_versions(&[("archive/object", "version-a"), ("archive/object", "version-b")], &[], false),
+                &[list_versions(
+                    &[("archive/object", "version-a"), ("archive/object", "version-b")],
+                    &[],
+                    false,
+                )],
             ),
             TransitionCandidateProbe::Ambiguous
         );
+    }
+
+    #[test]
+    fn transition_candidate_probe_reconciles_all_pages_and_ignores_delete_markers() {
+        assert_eq!(
+            classify_pages(
+                RemoteBucketVersioning::Enabled,
+                &[
+                    list_versions(&[], &[("archive/object", "marker-a")], true),
+                    list_versions(&[("archive/object", "version-a"), ("archive/object-adjacent", "unrelated"),], &[], false,),
+                ],
+            ),
+            TransitionCandidateProbe::VersionedPresent("version-a".to_string())
+        );
         assert_eq!(
-            classify_transition_candidate_versions(
-                "archive/object",
+            classify_pages(
                 RemoteBucketVersioning::Enabled,
-                &list_versions(&[("archive/object", "version-a")], &[("archive/object", "marker-a")], false),
-            ),
-            TransitionCandidateProbe::Ambiguous
-        );
-        assert_eq!(
-            classify_transition_candidate_versions(
-                "archive/object",
-                RemoteBucketVersioning::Enabled,
-                &list_versions(&[("archive/object", "version-a")], &[], true),
+                &[
+                    list_versions(&[("archive/object", "version-a")], &[], true),
+                    list_versions(&[("archive/object", "version-b")], &[], false),
+                ],
             ),
             TransitionCandidateProbe::Ambiguous
         );
     }
 
+    #[test]
+    fn transition_candidate_pagination_advances_both_markers() {
+        let mut key_marker = "old-key".to_string();
+        let mut version_id_marker = "old-version".to_string();
+        let page = ListVersionsResult {
+            next_key_marker: "next-key".to_string(),
+            next_version_id_marker: "next-version".to_string(),
+            ..Default::default()
+        };
+
+        advance_version_markers(&mut key_marker, &mut version_id_marker, &page)
+            .expect("new ListObjectVersions markers should advance pagination");
+        assert_eq!(key_marker, "next-key");
+        assert_eq!(version_id_marker, "next-version");
+
+        let err = advance_version_markers(&mut key_marker, &mut version_id_marker, &page)
+            .expect_err("repeated ListObjectVersions markers must fail closed");
+        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
+    }
+
+    #[test]
+    fn transition_candidate_probe_rejects_untrusted_version_ids() {
+        let mut candidates = TransitionCandidateVersions::default();
+        candidates.extend("archive/object", &list_versions(&[("archive/object", "version\ninjection")], &[], false));
+
+        let err = classify_transition_candidates(candidates, RemoteBucketVersioning::Enabled)
+            .expect_err("control characters in listed version IDs must fail closed");
+        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
+    }
+
     #[test]
     fn remote_bucket_versioning_status_parser_fails_closed() {
         assert_eq!(
@@ -397,12 +491,7 @@ impl WarmBackend for WarmBackendS3 {
 
     async fn probe_transition_candidate(&self, object: &str) -> Result<TransitionCandidateProbe, std::io::Error> {
         let bucket_versioning = self.remote_bucket_versioning().await?;
-        let versions = self.list_transition_candidate_versions(object).await?;
-        Ok(classify_transition_candidate_versions(
-            &self.get_dest(object),
-            bucket_versioning,
-            &versions,
-        ))
+        self.probe_transition_candidate_versions(object, bucket_versioning).await
     }
 
     async fn in_use(&self) -> Result<bool, std::io::Error> {

crates/ecstore/src/test_tracing.rs

@@ -0,0 +1,43 @@
+//! Test-only guard against tracing's process-global callsite-interest cache.
+//!
+//! Any test that installs its own subscriber and then asserts on span or event
+//! context is asserting on process-global state. See
+//! [`pin_callsite_interest_for_test`] for what goes wrong and why holding its
+//! guard fixes it.
+
+/// Stop a sibling test thread from poisoning tracing's process-global callsite
+/// interest cache while this test asserts on span or event context.
+///
+/// `tracing` caches every callsite's `Interest` in **process-global** state, and
+/// the first thread to reach a callsite fixes that value. While at most one
+/// dispatcher is registered, tracing-core takes a fast path
+/// (`Dispatchers::rebuilder` -> `Rebuilder::JustOne`) that derives a
+/// newly-registered callsite's interest from whichever subscriber is current
+/// *on the registering thread*, and registration happens exactly once (guarded
+/// by a CAS in `DefaultCallsite::register`).
+///
+/// In a multi-threaded `cargo test` binary a sibling test therefore routinely
+/// reaches a production callsite first, from a thread with no subscriber
+/// installed: the interest is derived from that thread's `NoSubscriber` and
+/// cached as `Interest::never()` for the whole process. From then on the
+/// callsite is dead for *every* later caller, including a test that installed
+/// its own subscriber — an `info_span!` silently evaluates to `Span::none()`, and
+/// a `warn!`/`info!` event never fires at all.
+///
+/// Registering a second, inert dispatcher closes both halves of that race:
+///
+/// * constructing it rebuilds every *already-registered* callsite's interest
+///   against the live dispatcher set — which includes the caller's subscriber —
+///   repairing whatever a sibling may already have poisoned; and
+/// * holding it alive keeps tracing-core off the single-dispatcher fast path, so
+///   a callsite registered *later* by any thread is resolved against that live
+///   set instead of the registering thread's `NoSubscriber`.
+///
+/// Call this **after** installing the test's subscriber, and hold the returned
+/// guard for the rest of the test. Only the `cargo test` fallback needs it:
+/// nextest runs each test in its own process, where there is no sibling thread
+/// to lose the race to (see `docs/testing/README.md`).
+#[must_use = "callsite interest is only pinned while the returned guard is held"]
+pub(crate) fn pin_callsite_interest_for_test() -> tracing::Dispatch {
+    tracing::Dispatch::new(tracing_core::subscriber::NoSubscriber::default())
+}

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<SwiftRoute> {
-        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 {

crates/s3select-api/Cargo.toml

@@ -25,6 +25,9 @@ keywords = ["s3-select", "api", "rustfs", "Minio", "object-store"]
 categories = ["web-programming", "development-tools", "asynchronous"]
 documentation = "https://docs.rs/rustfs-s3select-api/latest/rustfs_s3select_api/"
 
+[lints]
+workspace = true
+
 [dependencies]
 metrics = { workspace = true }
 async-trait.workspace = true

crates/s3select-query/Cargo.toml

@@ -25,6 +25,9 @@ keywords = ["s3-select", "query-engine", "rustfs", "Minio", "data-retrieval"]
 categories = ["web-programming", "development-tools", "data-structures"]
 documentation = "https://docs.rs/rustfs-s3select-query/latest/rustfs_s3select_query/"
 
+[lints]
+workspace = true
+
 [dependencies]
 rustfs-s3select-api = { workspace = true }
 async-recursion = { workspace = true }

crates/s3select-query/src/metadata/mod.rs

@@ -76,7 +76,7 @@ impl ContextProviderExtension for MetadataProvider {
 
         let table_handle = self.build_table_handle()?;
 
-        Ok(Arc::new(TableSourceAdapter::try_new(table_ref.clone(), table_name, table_handle)?))
+        Ok(Arc::new(TableSourceAdapter::try_new(table_ref, table_name, table_handle)?))
     }
 }
 

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::<request_context::RequestContext>().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<PutObjectInput>) -> S3Result<S3Response<PutObjectOutput>> {
         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::<request_context::RequestContext>().cloned();
-            spawn_background_with_context(request_context, async move {
+            spawn_background_with_context(Some(request_context.clone()), async move {
                 notify.notify(event_args).await;
             });
         }

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 {

rustfs/src/init.rs

@@ -15,6 +15,7 @@
 use crate::runtime_sources::current_region;
 use crate::server::ShutdownHandle;
 use crate::server::runtime_sources::current_notify_interface;
+use crate::storage_api::startup::bucket_metadata::contract::bucket::{BucketOperations, BucketOptions};
 use crate::storage_api::startup::init::{
     get_bucket_notification_config, process_lambda_configurations, process_queue_configurations, process_topic_configurations,
 };
@@ -23,11 +24,14 @@ use rustfs_config::{
     DEFAULT_BUFFER_MAX_SIZE, DEFAULT_BUFFER_MIN_SIZE, DEFAULT_BUFFER_PROFILE, DEFAULT_BUFFER_UNKNOWN_SIZE, DEFAULT_UPDATE_CHECK,
     ENV_RUSTFS_BUFFER_DEFAULT_SIZE, ENV_RUSTFS_BUFFER_MAX_SIZE, ENV_RUSTFS_BUFFER_MIN_SIZE, ENV_UPDATE_CHECK, RUSTFS_REGION,
 };
-use rustfs_targets::arn::{ARN, TargetIDError};
+use rustfs_notify::NotificationError;
+use rustfs_s3_types::EventName;
+use rustfs_targets::arn::{ARN, TargetID, TargetIDError};
 use rustfs_utils::get_env_usize;
 use s3s::s3_error;
 use std::env;
 use std::io::Error;
+use std::sync::Arc;
 use tracing::{debug, error, info, instrument, warn};
 
 const LOG_COMPONENT_INIT: &str = "init";
@@ -41,6 +45,8 @@ const LOG_SUBSYSTEM_PROTOCOL: &str = "protocol";
 const EVENT_PROTOCOL_RUNTIME_STATE: &str = "protocol_runtime_state";
 const EVENT_PROTOCOL_SERVER_STATE: &str = "protocol_server_state";
 
+type NotificationEventRule = (Vec<EventName>, String, String, Vec<TargetID>);
+
 #[instrument]
 pub fn print_server_info() {
     let current_year = jiff::Zoned::now().year();
@@ -151,6 +157,57 @@ fn arn_to_target_id(arn_str: &str) -> Result<rustfs_targets::arn::TargetID, Targ
         .map_err(|e| TargetIDError::InvalidFormat(e.to_string()))
 }
 
+fn notification_config_to_event_rules(
+    cfg: &s3s::dto::NotificationConfiguration,
+) -> Result<Vec<NotificationEventRule>, TargetIDError> {
+    let mut event_rules = Vec::new();
+    process_queue_configurations(&mut event_rules, cfg.queue_configurations.clone(), arn_to_target_id)?;
+    process_topic_configurations(&mut event_rules, cfg.topic_configurations.clone(), arn_to_target_id)?;
+    process_lambda_configurations(&mut event_rules, cfg.lambda_function_configurations.clone(), arn_to_target_id)?;
+    Ok(event_rules)
+}
+
+async fn apply_bucket_notification_configuration(bucket: &str, region: &str) -> Result<bool, NotificationError> {
+    let has_notification_config = get_bucket_notification_config(bucket)
+        .await
+        .map_err(|err| NotificationError::StorageNotAvailable(format!("load bucket notification config for {bucket}: {err}")))?;
+
+    match has_notification_config {
+        Some(cfg) => {
+            info!(
+                target: "rustfs::init",
+                event = "notification_config_loaded",
+                component = LOG_COMPONENT_INIT,
+                subsystem = LOG_SUBSYSTEM_NOTIFICATION,
+                bucket = %bucket,
+                queue_configuration_count = cfg.queue_configurations.as_ref().map_or(0, Vec::len),
+                topic_configuration_count = cfg.topic_configurations.as_ref().map_or(0, Vec::len),
+                lambda_configuration_count = cfg.lambda_function_configurations.as_ref().map_or(0, Vec::len),
+                "Loaded bucket notification configuration"
+            );
+
+            let event_rules =
+                notification_config_to_event_rules(&cfg).map_err(|err| NotificationError::BucketNotification(err.to_string()))?;
+            current_notify_interface()
+                .add_event_specific_rules(bucket, region, &event_rules)
+                .await?;
+            Ok(true)
+        }
+        None => {
+            info!(
+                target: "rustfs::init",
+                event = "notification_config_missing",
+                component = LOG_COMPONENT_INIT,
+                subsystem = LOG_SUBSYSTEM_NOTIFICATION,
+                bucket = %bucket,
+                "Bucket notification configuration not found"
+            );
+            current_notify_interface().clear_bucket_notification_rules(bucket).await?;
+            Ok(false)
+        }
+    }
+}
+
 /// Add existing bucket notification configurations to the global notifier system.
 /// This function retrieves notification configurations for each bucket
 /// and registers the corresponding event rules with the notifier system.
@@ -159,11 +216,52 @@ fn arn_to_target_id(arn_str: &str) -> Result<rustfs_targets::arn::TargetID, Targ
 /// * `buckets` - A vector of bucket names to process
 #[instrument(skip_all)]
 pub async fn add_bucket_notification_configuration(buckets: Vec<String>) {
+    let region = notification_region();
+    for bucket in buckets.iter() {
+        if let Err(err) = apply_bucket_notification_configuration(bucket, region.as_str()).await {
+            let err = s3_error!(InternalError, "Failed to add rules: {err}");
+            error!(
+                target: "rustfs::init",
+                event = "notification_rules_registration_failed",
+                component = LOG_COMPONENT_INIT,
+                subsystem = LOG_SUBSYSTEM_NOTIFICATION,
+                bucket = %bucket,
+                region,
+                error = ?err,
+                "Failed to register bucket notification rules"
+            );
+        }
+    }
+}
+
+pub(crate) async fn reconcile_persisted_bucket_notification_configurations(
+    store: Arc<rustfs_notify::NotifyStore>,
+) -> Result<usize, NotificationError> {
+    let bucket_infos = store
+        .list_bucket(&BucketOptions {
+            no_metadata: true,
+            ..Default::default()
+        })
+        .await
+        .map_err(|err| NotificationError::StorageNotAvailable(format!("list buckets for notification reconciliation: {err}")))?;
+    let region = notification_region();
+    let mut configured_bucket_count = 0;
+
+    for bucket in bucket_infos {
+        if apply_bucket_notification_configuration(&bucket.name, region.as_str()).await? {
+            configured_bucket_count += 1;
+        }
+    }
+
+    Ok(configured_bucket_count)
+}
+
+fn notification_region() -> String {
     let global_region = current_region();
-    let region = global_region
+    global_region
         .as_ref()
         .filter(|r| !r.as_str().is_empty())
-        .map(|r| r.as_str())
+        .map(|r| r.to_string())
         .unwrap_or_else(|| {
             warn!(
                 target: "rustfs::init",
@@ -173,107 +271,8 @@ pub async fn add_bucket_notification_configuration(buckets: Vec<String>) {
                 fallback_region = RUSTFS_REGION,
                 "Notification configuration falling back to default region"
             );
-            RUSTFS_REGION
-        });
-    for bucket in buckets.iter() {
-        let has_notification_config = get_bucket_notification_config(bucket).await.unwrap_or_else(|err| {
-            warn!(
-                target: "rustfs::init",
-                event = "notification_config_load_failed",
-                component = LOG_COMPONENT_INIT,
-                subsystem = LOG_SUBSYSTEM_NOTIFICATION,
-                bucket = %bucket,
-                error = ?err,
-                "Failed to load bucket notification configuration"
-            );
-            None
-        });
-
-        match has_notification_config {
-            Some(cfg) => {
-                info!(
-                    target: "rustfs::init",
-                    event = "notification_config_loaded",
-                    component = LOG_COMPONENT_INIT,
-                    subsystem = LOG_SUBSYSTEM_NOTIFICATION,
-                    bucket = %bucket,
-                    queue_configuration_count = cfg.queue_configurations.as_ref().map_or(0, Vec::len),
-                    topic_configuration_count = cfg.topic_configurations.as_ref().map_or(0, Vec::len),
-                    lambda_configuration_count = cfg.lambda_function_configurations.as_ref().map_or(0, Vec::len),
-                    "Loaded bucket notification configuration"
-                );
-
-                let mut event_rules = Vec::new();
-                if let Err(e) = process_queue_configurations(&mut event_rules, cfg.queue_configurations.clone(), arn_to_target_id)
-                {
-                    error!(
-                        target: "rustfs::init",
-                        event = "notification_config_parse_failed",
-                        component = LOG_COMPONENT_INIT,
-                        subsystem = LOG_SUBSYSTEM_NOTIFICATION,
-                        bucket = %bucket,
-                        config_type = "queue",
-                        error = ?e,
-                        "Failed to parse bucket notification configuration"
-                    );
-                }
-                if let Err(e) = process_topic_configurations(&mut event_rules, cfg.topic_configurations.clone(), arn_to_target_id)
-                {
-                    error!(
-                        target: "rustfs::init",
-                        event = "notification_config_parse_failed",
-                        component = LOG_COMPONENT_INIT,
-                        subsystem = LOG_SUBSYSTEM_NOTIFICATION,
-                        bucket = %bucket,
-                        config_type = "topic",
-                        error = ?e,
-                        "Failed to parse bucket notification configuration"
-                    );
-                }
-                if let Err(e) =
-                    process_lambda_configurations(&mut event_rules, cfg.lambda_function_configurations.clone(), arn_to_target_id)
-                {
-                    error!(
-                        target: "rustfs::init",
-                        event = "notification_config_parse_failed",
-                        component = LOG_COMPONENT_INIT,
-                        subsystem = LOG_SUBSYSTEM_NOTIFICATION,
-                        bucket = %bucket,
-                        config_type = "lambda",
-                        error = ?e,
-                        "Failed to parse bucket notification configuration"
-                    );
-                }
-
-                if let Err(e) = current_notify_interface()
-                    .add_event_specific_rules(bucket, region, &event_rules)
-                    .await
-                    .map_err(|e| s3_error!(InternalError, "Failed to add rules: {e}"))
-                {
-                    error!(
-                        target: "rustfs::init",
-                        event = "notification_rules_registration_failed",
-                        component = LOG_COMPONENT_INIT,
-                        subsystem = LOG_SUBSYSTEM_NOTIFICATION,
-                        bucket = %bucket,
-                        region,
-                        error = ?e,
-                        "Failed to register bucket notification rules"
-                    );
-                }
-            }
-            None => {
-                info!(
-                    target: "rustfs::init",
-                    event = "notification_config_missing",
-                    component = LOG_COMPONENT_INIT,
-                    subsystem = LOG_SUBSYSTEM_NOTIFICATION,
-                    bucket = %bucket,
-                    "Bucket notification configuration not found"
-                );
-            }
-        }
-    }
+            RUSTFS_REGION.to_string()
+        })
 }
 
 /// Build KMS configuration for local backend
@@ -1354,9 +1353,13 @@ pub async fn init_sftp_system() -> Result<Option<ShutdownHandle>, Box<dyn std::e
 
 #[cfg(test)]
 mod tests {
-    use super::resolve_buffer_profile_config;
+    use super::{notification_config_to_event_rules, resolve_buffer_profile_config};
     use crate::config::{BufferConfig, WorkloadProfile};
     use rustfs_config::KI_B;
+    use rustfs_s3_types::EventName;
+    use s3s::dto::{
+        FilterRule, FilterRuleName, NotificationConfiguration, NotificationConfigurationFilter, QueueConfiguration, S3KeyFilter,
+    };
 
     #[test]
     fn resolve_buffer_profile_config_returns_fallback_when_primary_is_invalid() {
@@ -1386,4 +1389,61 @@ mod tests {
 
         assert!(resolved.is_none());
     }
+
+    #[test]
+    fn notification_config_to_event_rules_preserves_target_and_filters() {
+        let cfg = NotificationConfiguration {
+            queue_configurations: Some(vec![QueueConfiguration {
+                events: vec!["s3:ObjectCreated:Put".to_string().into()],
+                queue_arn: "arn:rustfs:sqs:us-east-1:rustfs_to_activemq:mqtt".to_string(),
+                filter: Some(NotificationConfigurationFilter {
+                    key: Some(S3KeyFilter {
+                        filter_rules: Some(vec![
+                            FilterRule {
+                                name: Some(FilterRuleName::from_static(FilterRuleName::PREFIX)),
+                                value: Some("uploads/".to_string()),
+                            },
+                            FilterRule {
+                                name: Some(FilterRuleName::from_static(FilterRuleName::SUFFIX)),
+                                value: Some(".json".to_string()),
+                            },
+                        ]),
+                    }),
+                }),
+                id: Some("primary".to_string()),
+            }]),
+            topic_configurations: None,
+            lambda_function_configurations: None,
+            event_bridge_configuration: None,
+        };
+
+        let rules = notification_config_to_event_rules(&cfg).expect("valid notification config should map to event rules");
+
+        assert_eq!(rules.len(), 1);
+        assert_eq!(rules[0].0, vec![EventName::ObjectCreatedPut]);
+        assert_eq!(rules[0].1, "uploads/");
+        assert_eq!(rules[0].2, ".json");
+        assert_eq!(rules[0].3.len(), 1);
+        assert_eq!(rules[0].3[0].id, "rustfs_to_activemq");
+        assert_eq!(rules[0].3[0].name, "mqtt");
+    }
+
+    #[test]
+    fn notification_config_to_event_rules_rejects_invalid_arn() {
+        let cfg = NotificationConfiguration {
+            queue_configurations: Some(vec![QueueConfiguration {
+                events: vec!["s3:ObjectCreated:Put".to_string().into()],
+                queue_arn: "arn:aws:sqs:us-east-1:rustfs_to_activemq:mqtt".to_string(),
+                filter: None,
+                id: None,
+            }]),
+            topic_configurations: None,
+            lambda_function_configurations: None,
+            event_bridge_configuration: None,
+        };
+
+        let err = notification_config_to_event_rules(&cfg).expect_err("invalid ARN partition must fail");
+
+        assert!(err.to_string().contains("Invalid ARN"), "unexpected error: {err}");
+    }
 }

rustfs/src/server/event.rs

@@ -16,6 +16,7 @@ use super::{
     module_switch::{resolve_notify_module_state, validate_notify_module_env, with_refreshed_notify_module_state_from},
     refresh_persisted_module_switches_from, runtime_sources,
 };
+use crate::init::reconcile_persisted_bucket_notification_configurations;
 use crate::storage_api::server::event::{
     EventArgs as EcstoreEventArgs, StorageObjectInfo, read_existing_server_config_no_lock, register_event_dispatch_hook,
     with_server_config_read_lock,
@@ -38,6 +39,7 @@ use tracing::{info, instrument, warn};
 
 static NOTIFY_MODULE_ENABLED: AtomicBool = AtomicBool::new(rustfs_config::DEFAULT_NOTIFY_ENABLE);
 static NOTIFY_RUNTIME_RECONCILED: AtomicBool = AtomicBool::new(false);
+static NOTIFY_BUCKET_RULES_RECONCILED: AtomicBool = AtomicBool::new(false);
 static ECSTORE_EVENT_DISPATCH_HOOK: OnceLock<()> = OnceLock::new();
 
 const EVENT_NOTIFIER_RECONCILE_INTERVAL: Duration = Duration::from_secs(5);
@@ -54,6 +56,19 @@ pub(crate) fn mark_event_notifier_reconciled() {
 
 pub(crate) fn mark_event_notifier_unreconciled() {
     NOTIFY_RUNTIME_RECONCILED.store(false, Ordering::Release);
+    NOTIFY_BUCKET_RULES_RECONCILED.store(false, Ordering::Release);
+}
+
+fn are_bucket_notification_rules_reconciled() -> bool {
+    NOTIFY_BUCKET_RULES_RECONCILED.load(Ordering::Acquire)
+}
+
+fn mark_bucket_notification_rules_reconciled() {
+    NOTIFY_BUCKET_RULES_RECONCILED.store(true, Ordering::Release);
+}
+
+fn should_reconcile_bucket_notification_rules(runtime_changed: bool, notify_enabled: bool) -> bool {
+    notify_enabled && (runtime_changed || !are_bucket_notification_rules_reconciled())
 }
 
 pub fn refresh_notify_module_enabled() -> bool {
@@ -180,7 +195,7 @@ pub(crate) async fn reconcile_event_notifier_from_store(
         let system = ensure_live_events_initialized();
         let transition_system = system.clone();
         let transition_store = store.clone();
-        let transition = with_refreshed_notify_module_state_from(store, move |resolution| async move {
+        let transition = with_refreshed_notify_module_state_from(store.clone(), move |resolution| async move {
             NOTIFY_MODULE_ENABLED.store(resolution.enabled, Ordering::Relaxed);
             let read_store = transition_store.clone();
             let config_system = transition_system.clone();
@@ -208,11 +223,25 @@ pub(crate) async fn reconcile_event_notifier_from_store(
         .await
         .map_err(|err| NotificationError::Initialization(format!("failed to refresh notify module switch: {err}")))??;
 
+        let runtime_changed = transition.is_some();
         if let Some(transition) = transition {
             transition.wait().await?;
         }
 
-        ensure_event_notifier_converged(&system)
+        ensure_event_notifier_converged(&system)?;
+        if should_reconcile_bucket_notification_rules(runtime_changed, is_notify_module_enabled()) {
+            let configured_bucket_count = reconcile_persisted_bucket_notification_configurations(store).await?;
+            mark_bucket_notification_rules_reconciled();
+            info!(
+                event = EVENT_NOTIFY_RUNTIME_RECONCILE,
+                component = "notify",
+                subsystem = "bucket_rules",
+                configured_bucket_count,
+                "Persisted bucket notification rules reconciled"
+            );
+        }
+
+        Ok(())
     }
     .await;
 
@@ -405,7 +434,8 @@ mod tests {
         set_persisted_module_switches,
     };
     use super::{
-        convert_ecstore_object_info, init_event_notifier_with_store, parse_host_and_port, run_persisted_event_notifier_reconciler,
+        convert_ecstore_object_info, init_event_notifier_with_store, mark_bucket_notification_rules_reconciled,
+        parse_host_and_port, run_persisted_event_notifier_reconciler, should_reconcile_bucket_notification_rules,
     };
     use crate::server::is_event_notifier_reconciled;
     use crate::storage_api::server::event::StorageObjectInfo;
@@ -533,6 +563,30 @@ mod tests {
         assert_eq!(converted.transitioned_tier.as_deref(), Some("DEEP_ARCHIVE"));
     }
 
+    #[test]
+    fn bucket_rule_reconcile_runs_once_and_after_runtime_change() {
+        super::mark_event_notifier_unreconciled();
+
+        assert!(
+            should_reconcile_bucket_notification_rules(false, true),
+            "enabled notify must restore bucket rules until the first successful replay"
+        );
+
+        mark_bucket_notification_rules_reconciled();
+        assert!(
+            !should_reconcile_bucket_notification_rules(false, true),
+            "steady-state reconcile must not rescan buckets every tick"
+        );
+        assert!(
+            should_reconcile_bucket_notification_rules(true, true),
+            "target runtime changes must replay persisted bucket rules"
+        );
+        assert!(
+            !should_reconcile_bucket_notification_rules(true, false),
+            "disabled notify must not load external target rules"
+        );
+    }
+
     #[tokio::test(start_paused = true)]
     async fn persisted_reconciler_converges_after_one_injected_tick() {
         let persisted_generation = Arc::new(AtomicUsize::new(1));

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<A, B> {
     internode: B,
 }
 
-#[derive(Clone, Default)]
-struct InternodeRequestContextLiteLayer;
-
-impl<S> tower::Layer<S> for InternodeRequestContextLiteLayer {
-    type Service = InternodeRequestContextLiteService<S>;
-
-    fn layer(&self, inner: S) -> Self::Service {
-        InternodeRequestContextLiteService { inner }
-    }
-}
-
-#[derive(Clone)]
-struct InternodeRequestContextLiteService<S> {
-    inner: S,
-}
-
-impl<S, B> Service<HttpRequest<B>> for InternodeRequestContextLiteService<S>
-where
-    S: Service<HttpRequest<B>> + Clone,
-{
-    type Response = S::Response;
-    type Error = S::Error;
-    type Future = S::Future;
-
-    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<std::result::Result<(), Self::Error>> {
-        self.inner.poll_ready(cx)
-    }
-
-    fn call(&mut self, mut req: HttpRequest<B>) -> 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<A, B> PathDispatchService<A, B> {
     fn new(external: A, internode: B) -> Self {
         Self { external, internode }
@@ -1388,34 +1346,28 @@ fn process_connection(
         //  1. AddExtensionLayer<RemoteAddr>           — per-connection peer address
         //  2. AddExtensionLayer<SocketAddr>           — 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)

rustfs/src/server/hybrid.rs

@@ -28,6 +28,13 @@ pub(crate) fn hybrid<MakeRest, Grpc>(make_rest: MakeRest, grpc: Grpc) -> HybridS
     HybridService { rest: make_rest, grpc }
 }
 
+pub(crate) fn is_grpc_request<B>(req: &Request<B>) -> 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<Rest, Grpc> {
@@ -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<Incoming>) -> 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),
-            },
+            }
         }
     }
 }

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<B>) -> 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<B>(req: &HttpRequest<B>, 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<S> Layer<S> for ExternalRequestContextLayer {
+    type Service = ExternalRequestContextService<S>;
+
+    fn layer(&self, inner: S) -> Self::Service {
+        ExternalRequestContextService {
+            inner,
+            console_redirect_enabled: self.console_redirect_enabled,
+        }
+    }
+}
+
+#[derive(Clone)]
+pub struct ExternalRequestContextService<S> {
+    inner: S,
+    console_redirect_enabled: bool,
+}
+
+impl<S, B, ResBody> Service<HttpRequest<B>> for ExternalRequestContextService<S>
+where
+    S: Service<HttpRequest<B>, Response = Response<ResBody>>,
+{
+    type Response = Response<ResBody>;
+    type Error = S::Error;
+    type Future = ExternalRequestContextFuture<S::Future>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.inner.poll_ready(cx)
+    }
+
+    fn call(&mut self, mut req: HttpRequest<B>) -> 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<F> {
+        #[pin]
+        inner: F,
+        request_id: Option<HeaderValue>,
+        is_s3: bool,
+    }
+}
+
+impl<F, ResBody, E> Future for ExternalRequestContextFuture<F>
+where
+    F: Future<Output = Result<Response<ResBody>, E>>,
+{
+    type Output = Result<Response<ResBody>, E>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        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<S> {
     inner: S,
 }
 
+fn is_console_redirect_request<B>(req: &HttpRequest<B>) -> 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<S, RestBody, GrpcBody> Service<HttpRequest<Incoming>> for RedirectService<S>
 where
     S: Service<HttpRequest<Incoming>, Response = Response<HybridBody<RestBody, GrpcBody>>> + Clone + Send + 'static,
@@ -412,20 +552,8 @@ where
     }
 
     fn call(&mut self, req: HttpRequest<Incoming>) -> 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<Mutex<Option<HeaderMap>>>,
+        request_context: Arc<Mutex<Option<RequestContext>>>,
+        response_request_id: Option<HeaderValue>,
     }
 
     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<Mutex<Option<HeaderMap>>> {
             Arc::clone(&self.headers)
         }
+
+        fn request_context(&self) -> Arc<Mutex<Option<RequestContext>>> {
+            Arc::clone(&self.request_context)
+        }
     }
 
     impl<B: Send + 'static> Service<Request<B>> for HeaderCaptureService {
@@ -2065,10 +2206,350 @@ mod tests {
 
         fn call(&mut self, req: Request<B>) -> 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::<RequestContext>().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<AtomicUsize>,
@@ -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")

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<dyn std::error::Error + Send + Sync>;
 type BoxBody = http_body_util::combinators::UnsyncBoxBody<Bytes, BoxError>;
 
-/// 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<BoxBody> {
-    // 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<BoxBody> {
     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!("<RequestId>{id}</RequestId>"))
         .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<F, E, ReqBody>(
         "Request rejected by API rate limit"
     );
     Either::Right(ready(Ok(s3_too_many_requests_response(
-        req.headers().get(X_REQUEST_ID),
+        req.extensions()
+            .get::<RequestContext>()
+            .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("<Code>evil</Code>"));
+        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("<RequestId>"), "malformed id must be omitted entirely: {body}");
+        assert!(body.contains("<RequestId>server-request-id</RequestId>"), "body: {body}");
     }
 
     #[tokio::test]

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<T>(response: &S3Response<T>, request_id: &str) -> HashMap<String, String> {
+    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<EventArgsBuilder>,
     start_time: std::time::Instant,
-    request_context: Option<RequestContext>,
+    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<T>(&self, req: &S3Request<T>) -> 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");
             },
         );
     }

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<String>,
@@ -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";

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 {

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,
             };
         }
     }