refactor(admin): read persisted notification config for target subscriptions

collect_target_subscriptions walks PersistedNotificationConfiguration, which
get_notification_config already returns, instead of round-tripping it
through XML, the s3s notification DTO and serde_json. Queue, topic, then
lambda order, exact ARN matching, and the Prefix-before-prefix rule
precedence with value-less rules skipped are unchanged; a new test pins
that precedence.

.config/s3s-footprint-baseline.txt

@@ -12,6 +12,6 @@
 # s3_error_lines  lines holding s3_error!, same scope, minus those the script
 #                 header's rule clears (rustfs_s3_types's macro); fails closed
 # ecstore_files   Rust files referencing s3s paths under crates/ecstore/src
-files=65
+files=59
 s3_error_lines=289
 ecstore_files=0

.config/test-count-baseline.txt

@@ -10,7 +10,7 @@
 # intentionally removed. Regenerate with:
 #   scripts/check_test_count_ratchet.sh --update
 e2e_test	993	20
-rustfs	6078	64
+rustfs	6079	64
 rustfs-audit	70	0
 rustfs-checksums	26	0
 rustfs-common	36	0

rustfs/src/admin/handlers/event.rs

@@ -34,18 +34,17 @@ use crate::app::storage_api::bucket::metadata_sys::get_notification_config;
 use crate::server::{
     ADMIN_PREFIX, is_notify_module_enabled, refresh_notify_module_enabled, refresh_persisted_module_switches_from_store,
 };
-use crate::storage::storage_api::persisted_to_s3s;
 use http::StatusCode;
 use hyper::Method;
 use matchit::Params;
 use rustfs_config::notify::NOTIFY_ROUTE_PREFIX;
 use rustfs_config::server_config::Config;
 use rustfs_config::{EVENT_DEFAULT_DIR, MAX_ADMIN_REQUEST_BODY_SIZE};
+use rustfs_gateway_types::persistence::PersistedNotificationConfiguration;
 use rustfs_policy::policy::action::{Action, AdminAction};
 use rustfs_s3_types::{Body, S3Error, S3ErrorCode, S3Result, s3_error};
 use rustfs_targets::catalog::builtin::builtin_notify_target_admin_descriptors;
 use serde::{Deserialize, Serialize};
-use serde_json::Value;
 use std::collections::HashMap;
 use std::sync::LazyLock;
 use tracing::{error, info, warn};
@@ -278,55 +277,49 @@ fn subscription_target_arn(params: &Params<'_, '_>, region: &str) -> S3Result<St
 
 fn collect_target_subscriptions(
     bucket: &str,
-    config: s3s::dto::NotificationConfiguration,
+    config: &PersistedNotificationConfiguration,
     target_arn: &str,
-) -> S3Result<Vec<TargetSubscription>> {
-    let value = serde_json::to_value(config).map_err(|e| {
-        S3Error::with_message(S3ErrorCode::InternalError, format!("failed to serialize notification config: {e}"))
-    })?;
-    let mut subscriptions = Vec::new();
-    for key in [
-        "queue_configurations",
-        "topic_configurations",
-        "lambda_function_configurations",
-    ] {
-        if let Some(entries) = value.get(key).and_then(Value::as_array) {
-            for entry in entries {
-                let arn = ["queue_arn", "topic_arn", "lambda_function_arn"]
-                    .iter()
-                    .find_map(|field| entry.get(field).and_then(Value::as_str));
-                if arn != Some(target_arn) {
-                    continue;
-                }
-                let filters = entry
-                    .get("filter")
-                    .and_then(|v| v.get("key"))
-                    .and_then(|v| v.get("filter_rules"))
-                    .and_then(Value::as_array);
-                let filter_value = |name: &str| {
-                    filters.and_then(|rules| {
-                        rules.iter().find_map(|rule| {
-                            (rule.get("name").and_then(Value::as_str) == Some(name))
-                                .then(|| rule.get("value").and_then(Value::as_str).map(str::to_string))
-                                .flatten()
-                        })
-                    })
-                };
-                subscriptions.push(TargetSubscription {
-                    bucket: bucket.to_string(),
-                    id: entry.get("id").and_then(Value::as_str).map(str::to_string),
-                    events: entry
-                        .get("events")
-                        .and_then(Value::as_array)
-                        .map(|events| events.iter().filter_map(Value::as_str).map(str::to_string).collect())
-                        .unwrap_or_default(),
-                    prefix: filter_value("Prefix").or_else(|| filter_value("prefix")),
-                    suffix: filter_value("Suffix").or_else(|| filter_value("suffix")),
-                });
+) -> Vec<TargetSubscription> {
+    // Queue, topic, then lambda: the order the subscriptions route has always reported.
+    let queues = config
+        .queue_configurations
+        .iter()
+        .flatten()
+        .map(|c| (c.id.as_deref(), c.queue_arn.as_str(), &c.events, c.filter.as_ref()));
+    let topics = config
+        .topic_configurations
+        .iter()
+        .flatten()
+        .map(|c| (c.id.as_deref(), c.topic_arn.as_str(), &c.events, c.filter.as_ref()));
+    let lambdas = config
+        .lambda_function_configurations
+        .iter()
+        .flatten()
+        .map(|c| (c.id.as_deref(), c.lambda_function_arn.as_str(), &c.events, c.filter.as_ref()));
+
+    queues
+        .chain(topics)
+        .chain(lambdas)
+        .filter(|(_, arn, _, _)| *arn == target_arn)
+        .map(|(id, _, events, filter)| {
+            let rules = filter.and_then(|f| f.key.as_ref()).and_then(|k| k.filter_rules.as_deref());
+            // A matching rule without a value does not end the search, so a later rule can supply it.
+            let filter_value = |name: &str| {
+                rules.and_then(|rules| {
+                    rules
+                        .iter()
+                        .find_map(|rule| (rule.name.as_deref() == Some(name)).then(|| rule.value.clone()).flatten())
+                })
+            };
+            TargetSubscription {
+                bucket: bucket.to_string(),
+                id: id.map(str::to_string),
+                events: events.clone(),
+                prefix: filter_value("Prefix").or_else(|| filter_value("prefix")),
+                suffix: filter_value("Suffix").or_else(|| filter_value("suffix")),
             }
-        }
-    }
-    Ok(subscriptions)
+        })
+        .collect()
 }
 
 #[async_trait::async_trait]
@@ -350,7 +343,7 @@ impl Operation for ListTargetSubscriptions {
             else {
                 continue;
             };
-            subscriptions.extend(collect_target_subscriptions(&bucket.name, persisted_to_s3s(&config)?, &target_arn)?);
+            subscriptions.extend(collect_target_subscriptions(&bucket.name, &config, &target_arn));
         }
         let data = serde_json::to_vec(&subscriptions)
             .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("failed to serialize subscriptions: {e}")))?;
@@ -685,6 +678,7 @@ mod tests {
     use rustfs_config::server_config::{KV, KVS};
     use rustfs_config::{DEFAULT_DELIMITER, ENABLE_KEY};
     use rustfs_targets::arn::TargetID;
+    use serde_json::Value;
     use serial_test::serial;
     use std::collections::{HashMap, HashSet};
     use temp_env::{with_var, with_vars};
@@ -1215,36 +1209,59 @@ mod tests {
 
     #[test]
     fn target_subscriptions_match_all_config_kinds_without_crossing_target_identity() {
-        let config: s3s::dto::NotificationConfiguration = serde_json::from_value(serde_json::json!({
-            "queue_configurations": [
-                {
-                    "id": "uploads",
-                    "queue_arn": "arn:rustfs:sqs:eu-west-1:primary:webhook",
-                    "events": ["s3:ObjectCreated:*", "s3:ObjectRemoved:*"],
-                    "filter": { "key": { "filter_rules": [
-                        { "name": "prefix", "value": "photos/" },
-                        { "name": "suffix", "value": ".jpg" }
-                    ] } }
+        use rustfs_gateway_types::persistence::{
+            PersistedFilterRule, PersistedLambdaFunctionConfiguration, PersistedNotificationConfigurationFilter,
+            PersistedQueueConfiguration, PersistedS3KeyFilter, PersistedTopicConfiguration,
+        };
+        let events = |names: &[&str]| names.iter().map(|name| name.to_string()).collect::<Vec<_>>();
+        let queue = |id: &str, arn: &str, names: &[&str]| PersistedQueueConfiguration {
+            events: events(names),
+            filter: None,
+            id: Some(id.to_string()),
+            queue_arn: arn.to_string(),
+        };
+        let rule = |name: &str, value: &str| PersistedFilterRule {
+            name: Some(name.to_string()),
+            value: Some(value.to_string()),
+        };
+        let config = PersistedNotificationConfiguration {
+            queue_configurations: Some(vec![
+                PersistedQueueConfiguration {
+                    filter: Some(PersistedNotificationConfigurationFilter {
+                        key: Some(PersistedS3KeyFilter {
+                            filter_rules: Some(vec![rule("prefix", "photos/"), rule("suffix", ".jpg")]),
+                        }),
+                    }),
+                    ..queue(
+                        "uploads",
+                        "arn:rustfs:sqs:eu-west-1:primary:webhook",
+                        &["s3:ObjectCreated:*", "s3:ObjectRemoved:*"],
+                    )
                 },
-                { "id": "other-service", "queue_arn": "arn:rustfs:sqs:eu-west-1:primary:kafka", "events": ["s3:ObjectCreated:*"] },
-                { "id": "other-target", "queue_arn": "arn:rustfs:sqs:eu-west-1:secondary:webhook", "events": ["s3:ObjectCreated:*"] },
-                { "id": "other-region", "queue_arn": "arn:rustfs:sqs:us-east-1:primary:webhook", "events": ["s3:ObjectCreated:*"] }
-            ],
-            "topic_configurations": [
-                { "id": "topic", "topic_arn": "arn:rustfs:sqs:eu-west-1:primary:webhook", "events": ["s3:ObjectCreated:Put"] }
-            ],
-            "lambda_function_configurations": [
-                { "lambda_function_arn": "arn:rustfs:sqs:eu-west-1:primary:webhook", "events": ["s3:ObjectRemoved:Delete"] }
-            ]
-        }))
-        .expect("notification DTO should deserialize");
+                queue("other-service", "arn:rustfs:sqs:eu-west-1:primary:kafka", &["s3:ObjectCreated:*"]),
+                queue("other-target", "arn:rustfs:sqs:eu-west-1:secondary:webhook", &["s3:ObjectCreated:*"]),
+                queue("other-region", "arn:rustfs:sqs:us-east-1:primary:webhook", &["s3:ObjectCreated:*"]),
+            ]),
+            topic_configurations: Some(vec![PersistedTopicConfiguration {
+                events: events(&["s3:ObjectCreated:Put"]),
+                filter: None,
+                id: Some("topic".to_string()),
+                topic_arn: "arn:rustfs:sqs:eu-west-1:primary:webhook".to_string(),
+            }]),
+            lambda_function_configurations: Some(vec![PersistedLambdaFunctionConfiguration {
+                events: events(&["s3:ObjectRemoved:Delete"]),
+                filter: None,
+                id: None,
+                lambda_function_arn: "arn:rustfs:sqs:eu-west-1:primary:webhook".to_string(),
+            }]),
+            ..Default::default()
+        };
         let router = subscription_router();
         for target_type in ["webhook", "notify_webhook"] {
             let path = format!("/v3/target/{target_type}/primary/subscriptions");
             let params = router.at(&path).expect("subscription route should match").params;
             let arn = subscription_target_arn(&params, "eu-west-1").expect("webhook ARN should resolve");
-            let subscriptions =
-                collect_target_subscriptions("photos", config.clone(), &arn).expect("subscriptions should collect");
+            let subscriptions = collect_target_subscriptions("photos", &config, &arn);
             assert_eq!(
                 serde_json::to_value(subscriptions).expect("subscriptions should serialize"),
                 serde_json::json!([
@@ -1253,23 +1270,52 @@ mod tests {
                     { "bucket": "photos", "id": null, "events": ["s3:ObjectRemoved:Delete"], "prefix": null, "suffix": null }
                 ])
             );
-            assert!(
-                collect_target_subscriptions("photos", config.clone(), "arn:rustfs:sqs:eu-west-1:unused:webhook")
-                    .expect("unmatched subscriptions should collect")
-                    .is_empty()
-            );
+            assert!(collect_target_subscriptions("photos", &config, "arn:rustfs:sqs:eu-west-1:unused:webhook").is_empty());
         }
         assert!(
             collect_target_subscriptions(
                 "empty",
-                s3s::dto::NotificationConfiguration::default(),
+                &PersistedNotificationConfiguration::default(),
                 "arn:rustfs:sqs::primary:webhook"
             )
-            .expect("empty configuration should collect")
             .is_empty()
         );
     }
 
+    #[test]
+    fn target_subscription_filters_prefer_capitalized_rules_and_skip_rules_without_values() {
+        use rustfs_gateway_types::persistence::{
+            PersistedFilterRule, PersistedNotificationConfigurationFilter, PersistedQueueConfiguration, PersistedS3KeyFilter,
+        };
+        let rule = |name: &str, value: Option<&str>| PersistedFilterRule {
+            name: Some(name.to_string()),
+            value: value.map(str::to_string),
+        };
+        let arn = "arn:rustfs:sqs:eu-west-1:primary:webhook";
+        let config = PersistedNotificationConfiguration {
+            queue_configurations: Some(vec![PersistedQueueConfiguration {
+                events: vec!["s3:ObjectCreated:*".to_string()],
+                filter: Some(PersistedNotificationConfigurationFilter {
+                    key: Some(PersistedS3KeyFilter {
+                        filter_rules: Some(vec![
+                            rule("prefix", Some("lower/")),
+                            rule("Prefix", None),
+                            rule("Prefix", Some("upper/")),
+                            rule("suffix", None),
+                        ]),
+                    }),
+                }),
+                id: None,
+                queue_arn: arn.to_string(),
+            }]),
+            ..Default::default()
+        };
+        let subscriptions = collect_target_subscriptions("photos", &config, arn);
+        assert_eq!(subscriptions.len(), 1);
+        assert_eq!(subscriptions[0].prefix.as_deref(), Some("upper/"));
+        assert_eq!(subscriptions[0].suffix, None);
+    }
+
     #[tokio::test]
     async fn target_subscriptions_authorize_before_resolving_either_type_form() {
         let router = subscription_router();

rustfs/src/admin/handlers/heal.rs

@@ -2293,33 +2293,21 @@ mod tests {
 
     #[tokio::test]
     async fn test_nested_heal_routes_still_require_authentication() {
-        use s3s::route::S3Route;
-
         let mut router = super::S3Router::new(false);
         super::register_heal_route(&mut router).expect("register production Heal routes");
         for prefix in ["/rustfs/admin", "/minio/admin"] {
             for object in ["dir/object.bin", "dir%2Fobject.bin", "literal%252Fslash"] {
-                let mut req = s3s::S3Request {
-                    input: s3s::Body::empty(),
-                    method: http::Method::POST,
-                    uri: format!("{prefix}/v3/heal/test-bucket/{object}").parse().expect("Heal URI"),
-                    headers: http::HeaderMap::new(),
-                    extensions: http::Extensions::new(),
-                    credentials: None,
-                    region: None,
-                    service: None,
-                    trailing_headers: None,
-                };
+                let uri: http::Uri = format!("{prefix}/v3/heal/test-bucket/{object}").parse().expect("Heal URI");
                 let err = router
-                    .check_access(&mut req)
+                    .check_unsigned_access(http::Method::POST, uri.clone())
                     .await
                     .expect_err("router must require a signature");
-                assert_eq!(err.code(), &S3ErrorCode::AccessDenied.into());
+                assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
                 let err = router
-                    .call(req)
+                    .call_unsigned(http::Method::POST, uri)
                     .await
                     .expect_err("handler must independently require authentication");
-                assert_eq!(err.code(), &S3ErrorCode::InvalidRequest.into());
+                assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
                 assert!(err.to_string().contains("authentication required"));
             }
         }
@@ -2327,34 +2315,22 @@ mod tests {
 
     #[tokio::test]
     async fn replacement_recovery_diagnostics_require_authentication() {
-        use s3s::route::S3Route;
-
         let mut router = super::S3Router::new(false);
         super::register_heal_route(&mut router).expect("register production Heal routes");
         for prefix in ["/rustfs/admin", "/minio/admin"] {
             let path = format!("{prefix}/v4/heal/replacement-recovery");
             assert!(router.contains_compatible_route(http::Method::GET, &path));
-            let mut req = s3s::S3Request {
-                input: s3s::Body::empty(),
-                method: http::Method::GET,
-                uri: path.parse().expect("replacement diagnostics URI"),
-                headers: http::HeaderMap::new(),
-                extensions: http::Extensions::new(),
-                credentials: None,
-                region: None,
-                service: None,
-                trailing_headers: None,
-            };
+            let uri: http::Uri = path.parse().expect("replacement diagnostics URI");
             let err = router
-                .check_access(&mut req)
+                .check_unsigned_access(http::Method::GET, uri.clone())
                 .await
                 .expect_err("diagnostics router must require a signature");
-            assert_eq!(err.code(), &S3ErrorCode::AccessDenied.into());
+            assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
             let err = router
-                .call(req)
+                .call_unsigned(http::Method::GET, uri)
                 .await
                 .expect_err("diagnostics handler must authenticate before inspecting local records");
-            assert_eq!(err.code(), &S3ErrorCode::InvalidRequest.into());
+            assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
             assert!(err.to_string().contains("authentication required"));
         }
     }

rustfs/src/admin/handlers/health.rs

@@ -91,56 +91,42 @@ impl Operation for HealthCheckHandler {
 mod tests {
     use super::*;
     use http_body_util::BodyExt as _;
-    use s3s::route::S3Route as _;
     use serial_test::serial;
     use temp_env::with_var;
 
-    fn health_request(method: Method) -> s3s::S3Request<s3s::Body> {
-        s3s::S3Request {
-            input: s3s::Body::empty(),
-            method,
-            uri: HEALTH_PREFIX.parse().expect("health URI"),
-            headers: HeaderMap::new(),
-            extensions: http::Extensions::new(),
-            credentials: None,
-            region: None,
-            service: None,
-            trailing_headers: None,
-        }
-    }
-
-    async fn call_real_health_handler(method: Method) -> s3s::S3Response<s3s::Body> {
+    async fn call_real_health_handler(method: Method) -> (StatusCode, HeaderMap, Body) {
         let mut router = S3Router::new(false);
         router
             .insert(method.clone(), HEALTH_PREFIX, AdminOperation(&HealthCheckHandler {}))
             .expect("health route should register");
         router
-            .call(health_request(method))
+            .call_unsigned(method, HEALTH_PREFIX.parse().expect("health URI"))
             .await
             .expect("health handler should return a response")
+            .into_parts()
     }
 
     #[tokio::test]
     async fn real_health_handler_get_and_head_preserve_wire_branches() {
-        let get = call_real_health_handler(Method::GET).await;
-        assert_eq!(get.status, Some(StatusCode::OK));
+        let (get_status, get_headers, get_body) = call_real_health_handler(Method::GET).await;
+        assert_eq!(get_status, StatusCode::OK);
         assert_eq!(
-            get.headers.get(CONTENT_TYPE).and_then(|value| value.to_str().ok()),
+            get_headers.get(CONTENT_TYPE).and_then(|value| value.to_str().ok()),
             Some("application/json")
         );
         let payload: serde_json::Value =
-            serde_json::from_slice(&get.output.collect().await.expect("GET body should read").to_bytes())
+            serde_json::from_slice(&get_body.collect().await.expect("GET body should read").to_bytes())
                 .expect("GET health response should be JSON");
         assert_eq!(payload["status"], "ok");
 
-        let head = call_real_health_handler(Method::HEAD).await;
-        assert_eq!(head.status, Some(StatusCode::OK));
+        let (head_status, head_headers, head_body) = call_real_health_handler(Method::HEAD).await;
+        assert_eq!(head_status, StatusCode::OK);
         assert_eq!(
-            head.headers.get(CONTENT_TYPE).and_then(|value| value.to_str().ok()),
+            head_headers.get(CONTENT_TYPE).and_then(|value| value.to_str().ok()),
             Some("application/json")
         );
         assert!(
-            head.output
+            head_body
                 .collect()
                 .await
                 .expect("HEAD body should read")
@@ -151,18 +137,12 @@ mod tests {
 
     #[tokio::test]
     async fn real_health_handler_rejects_unsupported_methods_with_allow_header() {
-        let response = call_real_health_handler(Method::POST).await;
+        let (status, headers, body) = call_real_health_handler(Method::POST).await;
 
-        assert_eq!(response.status, Some(StatusCode::METHOD_NOT_ALLOWED));
+        assert_eq!(status, StatusCode::METHOD_NOT_ALLOWED);
+        assert_eq!(headers.get(http::header::ALLOW).and_then(|value| value.to_str().ok()), Some("GET, HEAD"));
         assert_eq!(
-            response
-                .headers
-                .get(http::header::ALLOW)
-                .and_then(|value| value.to_str().ok()),
-            Some("GET, HEAD")
-        );
-        assert_eq!(
-            response.output.collect().await.expect("error body should read").to_bytes(),
+            body.collect().await.expect("error body should read").to_bytes(),
             bytes::Bytes::from_static(b"Method Not Allowed")
         );
     }

rustfs/src/admin/handlers/pools.rs

@@ -1299,9 +1299,9 @@ mod pools_handler_tests {
 
         let err = decommission_start_api_error(err);
 
-        assert_eq!(err.code, s3s::S3ErrorCode::ServiceUnavailable);
+        assert_eq!(err.code, rustfs_s3_types::S3ErrorCode::ServiceUnavailable.into());
         assert!(err.message.contains("data movement gate (decommission activation)"), "{}", err.message);
-        let err = s3s::S3Error::from(contextualize_admin_pool_api_error(err, "start decommission", "pools [0]"));
+        let err = rustfs_s3_types::S3Error::from(contextualize_admin_pool_api_error(err, "start decommission", "pools [0]"));
         assert_eq!(
             err.headers().and_then(|headers| headers.get(http::header::RETRY_AFTER)),
             Some(&http::HeaderValue::from_static("1")),
@@ -1313,8 +1313,8 @@ mod pools_handler_tests {
     fn test_decommission_start_api_error_reports_a_refused_preflight_heal_as_retryable() {
         let err = decommission_start_api_error(crate::storage_api::error::StorageError::SlowDown);
 
-        assert_eq!(err.code, s3s::S3ErrorCode::ServiceUnavailable);
-        let err = s3s::S3Error::from(contextualize_admin_pool_api_error(err, "start decommission", "pools [0]"));
+        assert_eq!(err.code, rustfs_s3_types::S3ErrorCode::ServiceUnavailable.into());
+        let err = rustfs_s3_types::S3Error::from(contextualize_admin_pool_api_error(err, "start decommission", "pools [0]"));
         assert_eq!(err.status_code(), Some(http::StatusCode::SERVICE_UNAVAILABLE));
     }
 

rustfs/src/admin/router.rs

@@ -29,7 +29,7 @@ use crate::admin::console::{is_console_path, make_console_server};
 use crate::admin::handlers::oidc::is_oidc_path;
 use crate::admin::operation::{AdminRequest, Operation as OwnedOperation};
 use crate::admin::runtime_sources::{
-    ServerContextSlot, app_context_from_req, current_boot_time, current_bucket_monitor_handle, current_deployment_id,
+    ServerContextSlot, app_context_from_extensions, current_boot_time, current_bucket_monitor_handle, current_deployment_id,
     current_notification_system, current_object_store_handle, current_region, current_replication_pool_handle,
     current_server_config, default_object_usecase,
 };
@@ -3021,7 +3021,7 @@ async fn handle_replication_extension_request(
     match ext_req.route {
         ReplicationExtRoute::MetricsV1 | ReplicationExtRoute::MetricsV2 => {
             ensure_replication_config_exists(&ext_req.bucket).await?;
-            build_replication_metrics_response(&ext_req.bucket, ext_req.route, app_context_from_req(req)).await
+            build_replication_metrics_response(&ext_req.bucket, ext_req.route, app_context_from_extensions(&req.extensions)).await
         }
         ReplicationExtRoute::Check => {
             let (versioning, _) = metadata_sys::get_versioning_config(&ext_req.bucket)
@@ -3233,6 +3233,51 @@ impl<T: Operation> S3Router<T> {
     pub(crate) fn registered_routes(&self) -> &[String] {
         &self.registered_routes
     }
+
+    /// Runs an unsigned request (no presented credentials, no headers, empty
+    /// body) through this router's `check_access` gate, so handler tests can
+    /// pin the router's authentication with owned types only.
+    pub(crate) async fn check_unsigned_access(&self, method: Method, uri: Uri) -> rustfs_s3_types::S3Result<()> {
+        let mut req = unsigned_legacy_request(method, uri);
+        S3Route::check_access(self, &mut req)
+            .await
+            .map_err(rustfs_s3_types::S3Error::from)
+    }
+
+    /// Dispatches an unsigned request (no presented credentials, no headers,
+    /// empty body) through this router's `call`, returning the owned admin
+    /// response. Every dispatched response carries a status, so a missing one
+    /// fails the test.
+    pub(crate) async fn call_unsigned(
+        &self,
+        method: Method,
+        uri: Uri,
+    ) -> rustfs_s3_types::S3Result<crate::admin::operation::AdminResponse> {
+        let resp = S3Route::call(self, unsigned_legacy_request(method, uri))
+            .await
+            .map_err(rustfs_s3_types::S3Error::from)?;
+        let status = resp.status.expect("a dispatched admin response carries a status");
+        Ok(crate::admin::operation::AdminResponse::with_headers(
+            status,
+            resp.output.into(),
+            resp.headers,
+        ))
+    }
+}
+
+#[cfg(test)]
+fn unsigned_legacy_request(method: Method, uri: Uri) -> S3Request<Body> {
+    S3Request {
+        input: Body::empty(),
+        method,
+        uri,
+        headers: HeaderMap::new(),
+        extensions: Extensions::new(),
+        credentials: None,
+        region: None,
+        service: None,
+        trailing_headers: None,
+    }
 }
 
 impl<T: Operation> Default for S3Router<T> {

rustfs/src/admin/runtime_sources.rs

@@ -66,10 +66,6 @@ pub(crate) fn current_object_data_cache() -> Option<Arc<ObjectDataCacheAdapter>>
     current_object_data_cache_handle_for_context(context.as_deref())
 }
 
-pub(crate) fn app_context_from_req<B>(req: &s3s::S3Request<B>) -> Option<Arc<AppContext>> {
-    app_context_from_extensions(&req.extensions)
-}
-
 pub(crate) fn app_context_from_envelope(envelope: &RequestEnvelope) -> Option<Arc<AppContext>> {
     match envelope.server_context() {
         Some(slot) => slot.installed_app_context(),
@@ -160,17 +156,17 @@ pub(crate) fn publish_storage_class_config(config: StorageClassConfig) {
 mod tests {
     use super::{
         AppContext, IamInterface, KmsInterface, ServerContextSlot, app_context_from_envelope, app_context_from_extensions,
-        app_context_from_req, current_app_context, object_store_from_envelope, publish_test_app_context,
+        current_app_context, object_store_from_envelope, publish_test_app_context,
     };
-    use crate::admin::router::{Operation, S3Router};
-    use crate::app::RequestEnvelope;
+    use crate::admin::operation::{AdminRequest, AdminResponse, Operation};
+    use crate::admin::router::S3Router;
+    use crate::app::{RequestEnvelope, RequestHead};
     use hyper::{Method, StatusCode};
     use matchit::Params;
     use rustfs_iam::{store::object::ObjectStore, sys::IamSys};
     use rustfs_kms::KmsServiceManager;
+    use rustfs_s3_types::Body;
     use rustfs_s3_types::{S3ErrorCode, s3_error};
-    use s3s::route::S3Route;
-    use s3s::{Body, S3Request, S3Response};
     use std::sync::Arc;
 
     struct UnreadyIam;
@@ -213,22 +209,16 @@ mod tests {
         Arc::new(AppContext::new(context.object_store(), Arc::new(UnreadyIam), Arc::new(TestKms)))
     }
 
-    fn request(extensions: http::Extensions) -> S3Request<Body> {
-        S3Request {
-            input: Body::empty(),
+    fn envelope(extensions: http::Extensions) -> RequestEnvelope {
+        let head = RequestHead {
             method: Method::GET,
             uri: "/context-probe".parse().expect("test URI"),
             headers: http::HeaderMap::new(),
-            extensions,
             credentials: None,
             region: None,
             service: None,
-            trailing_headers: None,
-        }
-    }
-
-    fn envelope(extensions: http::Extensions) -> RequestEnvelope {
-        RequestEnvelope::from_s3s(&request(extensions))
+        };
+        RequestEnvelope::from_parts(head, extensions)
     }
 
     #[tokio::test]
@@ -240,8 +230,7 @@ mod tests {
         extensions.insert(slot);
 
         assert!(app_context_from_extensions(&extensions).is_none());
-        assert!(object_store_from_envelope(&envelope(extensions.clone())).is_none());
-        assert!(app_context_from_req(&request(extensions)).is_none());
+        assert!(object_store_from_envelope(&envelope(extensions)).is_none());
         assert!(Arc::ptr_eq(
             &current_app_context().expect("ambient context must remain available"),
             &ambient
@@ -333,13 +322,10 @@ mod tests {
 
     #[async_trait::async_trait]
     impl Operation for ContextDependentAdminRoute {
-        async fn call(
-            &self,
-            req: S3Request<Body>,
-            _params: Params<'_, '_>,
-        ) -> rustfs_s3_types::S3Result<S3Response<(StatusCode, Body)>> {
-            app_context_from_req(&req).ok_or_else(|| s3_error!(ServiceUnavailable, "server context is not ready"))?;
-            Ok(S3Response::new((StatusCode::NO_CONTENT, Body::empty())))
+        async fn call(&self, req: AdminRequest, _params: Params<'_, '_>) -> rustfs_s3_types::S3Result<AdminResponse> {
+            app_context_from_envelope(req.envelope())
+                .ok_or_else(|| s3_error!(ServiceUnavailable, "server context is not ready"))?;
+            Ok(AdminResponse::new(StatusCode::NO_CONTENT, Body::empty()))
         }
     }
 
@@ -355,16 +341,16 @@ mod tests {
             .expect("register test route");
 
         let err = router
-            .call(request(http::Extensions::new()))
+            .call_unsigned(Method::GET, "/context-probe".parse().expect("test URI"))
             .await
             .expect_err("empty slot must fail closed");
-        assert_eq!(err.code(), &S3ErrorCode::ServiceUnavailable.into());
+        assert_eq!(err.code(), &S3ErrorCode::ServiceUnavailable);
 
         assert!(slot.install(distinct_context(&ambient)));
         let response = router
-            .call(request(http::Extensions::new()))
+            .call_unsigned(Method::GET, "/context-probe".parse().expect("test URI"))
             .await
             .expect("installed slot must serve request");
-        assert_eq!(response.status, Some(StatusCode::NO_CONTENT));
+        assert_eq!(response.into_parts().0, StatusCode::NO_CONTENT);
     }
 }

rustfs/src/admin/utils.rs

@@ -165,9 +165,10 @@ mod tests {
         let payload = b"{\"ok\":true}".to_vec();
         let encrypted = encrypt_data(b"secret", &payload).expect("encrypt payload");
 
-        let decoded = read_compatible_admin_body(s3s::Body::from(encrypted), 1024, "/minio/admin/v3/list-users", "secret")
-            .await
-            .expect("decode payload");
+        let decoded =
+            read_compatible_admin_body(rustfs_s3_types::Body::from(encrypted), 1024, "/minio/admin/v3/list-users", "secret")
+                .await
+                .expect("decode payload");
 
         assert_eq!(decoded, payload);
     }
@@ -191,12 +192,17 @@ mod tests {
         assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
         assert_eq!(err.message(), Some("failed to read request body: length limit exceeded"));
 
-        let legacy_body = s3s::Body::http_body(http_body_util::Full::new(bytes::Bytes::from_static(b"nine bytes!")));
-        let legacy_err = read_compatible_admin_body(legacy_body, 8, "/rustfs/admin/v3/list-users", "secret")
+        // A streaming body has no exact length up front, so the limit trips
+        // while frames accumulate rather than on the size hint.
+        let streaming_body = rustfs_s3_types::Body::from_stream(futures_util::stream::iter([
+            Ok::<_, std::io::Error>(bytes::Bytes::from_static(b"nine ")),
+            Ok(bytes::Bytes::from_static(b"bytes!")),
+        ]));
+        let streaming_err = read_compatible_admin_body(streaming_body, 8, "/rustfs/admin/v3/list-users", "secret")
             .await
-            .expect_err("legacy streaming bodies retain the same limit error");
-        assert_eq!(legacy_err.code(), err.code());
-        assert_eq!(legacy_err.message(), err.message());
+            .expect_err("streaming bodies retain the same limit error");
+        assert_eq!(streaming_err.code(), err.code());
+        assert_eq!(streaming_err.message(), err.message());
 
         let exact =
             read_compatible_admin_body(rustfs_s3_types::Body::from("12345678"), 8, "/rustfs/admin/v3/list-users", "secret")