perf: reduce per-request allocation pressure in hot path

- Async access log via unbounded channel (JSON serialization off hot path)
- Replace header_map HashMap with direct http::HeaderMap access (~20 fewer
  String allocs per request); update RouterTable::match_request and
  Rule::matches signatures accordingly
- reqwest pool: tcp_nodelay(true), pool_idle_timeout(90s)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: ZhiXiao-Lin

src/entrypoint.rs

@@ -90,8 +90,10 @@ pub struct GatewayState {
     pub mirrors: HashMap<String, Arc<crate::service::TrafficMirror>>,
     /// Failover selectors: service_name → FailoverSelector
     pub failovers: HashMap<String, Arc<crate::service::FailoverSelector>>,
-    /// Structured access log
+    /// Structured access log (counter + background task target)
     pub access_log: Arc<crate::observability::access_log::AccessLog>,
+    /// Channel for fire-and-forget log entries — background task does JSON + tracing
+    pub log_tx: tokio::sync::mpsc::UnboundedSender<crate::observability::access_log::AccessLogEntry>,
     /// Sticky session managers (only for services with sticky config)
     pub sticky_managers: HashMap<String, Arc<StickySessionManager>>,
     /// Passive health checkers for all services
@@ -327,7 +329,7 @@ async fn start_tcp_entrypoint(
             tokio::spawn(async move {
                 let _permit = permit;
 
-                let headers = HashMap::new();
+                let headers = http::HeaderMap::new();
                 if let Some(route) = state
                     .router_table
                     .match_request(None, "/", "TCP", &headers, &ep_name)
@@ -381,7 +383,7 @@ async fn start_udp_entrypoint(
     max_sessions: Option<usize>,
     state: Arc<GatewayState>,
 ) -> Result<tokio::task::JoinHandle<()>> {
-    let headers = HashMap::new();
+    let headers = http::HeaderMap::new();
     let upstream_addr = state
         .router_table
         .match_request(None, "/", "UDP", &headers, &name)
@@ -446,14 +448,6 @@ async fn handle_http_request(
     let method_str = req.method().as_str().to_string();
     let uri = req.uri().clone();
 
-    // Collect headers into a plain map for routing and middleware context.
-    let mut header_map = HashMap::new();
-    for (key, value) in req.headers().iter() {
-        if let Ok(v) = value.to_str() {
-            header_map.insert(key.as_str().to_string(), v.to_string());
-        }
-    }
-
     // Detect protocol from request headers.
     let is_ws = crate::proxy::websocket::is_websocket_upgrade(req.headers());
     let is_grpc = crate::proxy::grpc::is_grpc_request(req.headers());
@@ -462,7 +456,7 @@ async fn handle_http_request(
     let access_tracker = state.access_log.start_request();
 
     // Extract incoming trace context and create a child span.
-    let trace_ctx = crate::observability::tracing::extract_trace_context(&header_map)
+    let trace_ctx = crate::observability::tracing::extract_trace_context(req.headers())
         .map(|ctx| ctx.child())
         .unwrap_or_else(crate::observability::tracing::TraceContext::new_root);
 
@@ -471,7 +465,7 @@ async fn handle_http_request(
         host.as_deref(),
         &path,
         &method_str,
-        &header_map,
+        req.headers(),
         &entrypoint,
     ) {
         Some(route) => route,
@@ -644,8 +638,10 @@ async fn handle_http_request(
         .sticky_managers
         .get(&route.service_name)
         .and_then(|mgr| {
-            let session_id = header_map
+            let session_id = req_parts
+                .headers
                 .get("cookie")
+                .and_then(|v| v.to_str().ok())
                 .and_then(|cookie| mgr.extract_session_id(cookie))
                 .map(|s| s.to_string());
             match mgr.select_backend(session_id.as_deref(), lb.backends()) {
@@ -751,7 +747,7 @@ async fn handle_http_request(
         {
             Ok(grpc_resp) => {
                 let status_code = grpc_resp.http_status.as_u16();
-                state.access_log.record(&access_tracker.build_entry(
+                let _ = state.log_tx.send(access_tracker.build_entry(
                     remote_addr.ip().to_string(),
                     method_str,
                     path,
@@ -761,7 +757,7 @@ async fn handle_http_request(
                     Some(backend.url.clone()),
                     Some(route.router_name.clone()),
                     Some(entrypoint),
-                    header_map.get("user-agent").cloned(),
+                    req_parts.headers.get("user-agent").and_then(|v| v.to_str().ok()).map(|s| s.to_string()),
                 ));
 
                 // Record passive health from HTTP status.
@@ -806,7 +802,7 @@ async fn handle_http_request(
             }
             Err(e) => {
                 tracing::error!(error = %e, backend = backend.url, "gRPC proxy error");
-                state.access_log.record(&access_tracker.build_entry(
+                let _ = state.log_tx.send(access_tracker.build_entry(
                     remote_addr.ip().to_string(),
                     method_str,
                     path,
@@ -816,7 +812,7 @@ async fn handle_http_request(
                     Some(backend.url.clone()),
                     Some(route.router_name.clone()),
                     Some(entrypoint),
-                    header_map.get("user-agent").cloned(),
+                    req_parts.headers.get("user-agent").and_then(|v| v.to_str().ok()).map(|s| s.to_string()),
                 ));
                 if let Some(phc) = state.passive_health.get(&route.service_name) {
                     phc.record_error(&backend, 502);
@@ -864,7 +860,7 @@ async fn handle_http_request(
         {
             Ok(stream_resp) => {
                 let status_code = stream_resp.status.as_u16();
-                state.access_log.record(&access_tracker.build_entry(
+                let _ = state.log_tx.send(access_tracker.build_entry(
                     remote_addr.ip().to_string(),
                     method_str,
                     path,
@@ -874,7 +870,7 @@ async fn handle_http_request(
                     Some(backend.url.clone()),
                     Some(route.router_name.clone()),
                     Some(entrypoint),
-                    header_map.get("user-agent").cloned(),
+                    req_parts.headers.get("user-agent").and_then(|v| v.to_str().ok()).map(|s| s.to_string()),
                 ));
 
                 if let Some(phc) = state.passive_health.get(&route.service_name) {
@@ -982,7 +978,7 @@ async fn handle_http_request(
         Ok(proxy_resp) => {
             let status_code = proxy_resp.status.as_u16();
 
-            state.access_log.record(&access_tracker.build_entry(
+            let _ = state.log_tx.send(access_tracker.build_entry(
                 remote_addr.ip().to_string(),
                 method_str,
                 path,
@@ -992,7 +988,7 @@ async fn handle_http_request(
                 Some(backend.url.clone()),
                 Some(route.router_name.clone()),
                 Some(entrypoint),
-                header_map.get("user-agent").cloned(),
+                req_parts.headers.get("user-agent").and_then(|v| v.to_str().ok()).map(|s| s.to_string()),
             ));
 
             // Passive health: record 5xx errors.
@@ -1051,7 +1047,7 @@ async fn handle_http_request(
                 phc.record_error(&backend, 502);
             }
 
-            state.access_log.record(&access_tracker.build_entry(
+            let _ = state.log_tx.send(access_tracker.build_entry(
                 remote_addr.ip().to_string(),
                 method_str,
                 path,
@@ -1061,7 +1057,7 @@ async fn handle_http_request(
                 Some(backend.url.clone()),
                 Some(route.router_name.clone()),
                 Some(entrypoint),
-                header_map.get("user-agent").cloned(),
+                req_parts.headers.get("user-agent").and_then(|v| v.to_str().ok()).map(|s| s.to_string()),
             ));
 
             state.metrics.record_request(502, 0);
@@ -1130,6 +1126,7 @@ mod tests {
             mirrors: HashMap::new(),
             failovers: HashMap::new(),
             access_log: Arc::new(crate::observability::access_log::AccessLog::new()),
+            log_tx: tokio::sync::mpsc::unbounded_channel().0,
             sticky_managers: HashMap::new(),
             passive_health: HashMap::new(),
             metrics: Arc::new(crate::observability::metrics::GatewayMetrics::new()),

src/gateway.rs

@@ -122,6 +122,10 @@ impl Gateway {
         let middleware_configs = Arc::new(config.middlewares.clone());
         let pipeline_cache = Arc::new(build_pipeline_cache(&config, &middleware_configs));
 
+        let access_log = Arc::new(crate::observability::access_log::AccessLog::new());
+        let (log_tx, log_rx) = tokio::sync::mpsc::unbounded_channel();
+        spawn_log_task(log_rx, access_log.clone());
+
         let gw_state = Arc::new(entrypoint::GatewayState {
             router_table,
             service_registry,
@@ -132,7 +136,8 @@ impl Gateway {
             scaling: scaling_state,
             mirrors,
             failovers,
-            access_log: Arc::new(crate::observability::access_log::AccessLog::new()),
+            access_log,
+            log_tx,
             sticky_managers: build_sticky_managers(&config),
             passive_health: build_passive_health(&config),
             metrics: self.metrics.clone(),
@@ -365,6 +370,10 @@ impl Gateway {
         let middleware_configs = Arc::new(new_config.middlewares.clone());
         let pipeline_cache = Arc::new(build_pipeline_cache(&new_config, &middleware_configs));
 
+        let access_log = Arc::new(crate::observability::access_log::AccessLog::new());
+        let (log_tx, log_rx) = tokio::sync::mpsc::unbounded_channel();
+        spawn_log_task(log_rx, access_log.clone());
+
         let gw_state = Arc::new(entrypoint::GatewayState {
             router_table,
             service_registry,
@@ -375,7 +384,8 @@ impl Gateway {
             scaling: scaling_state,
             mirrors,
             failovers,
-            access_log: Arc::new(crate::observability::access_log::AccessLog::new()),
+            access_log,
+            log_tx,
             sticky_managers: build_sticky_managers(&new_config),
             passive_health: build_passive_health(&new_config),
             metrics: self.metrics.clone(),
@@ -809,6 +819,19 @@ fn spawn_autoscaler(
     Some(handle)
 }
 
+/// Spawn a background task that drains the access log channel and serializes entries.
+/// This keeps JSON serialization and tracing off the request hot path.
+fn spawn_log_task(
+    mut rx: tokio::sync::mpsc::UnboundedReceiver<crate::observability::access_log::AccessLogEntry>,
+    access_log: Arc<crate::observability::access_log::AccessLog>,
+) {
+    tokio::spawn(async move {
+        while let Some(entry) = rx.recv().await {
+            access_log.record(&entry);
+        }
+    });
+}
+
 /// Pre-compile middleware pipelines for all routers — avoids per-request Pipeline::from_config.
 fn build_pipeline_cache(
     config: &GatewayConfig,

src/observability/access_log.rs

@@ -57,7 +57,7 @@ impl AccessLog {
         }
     }
 
-    /// Record and emit a log entry
+    /// Record and emit a log entry (called from background logging task)
     pub fn record(&self, entry: &AccessLogEntry) {
         self.total_entries.fetch_add(1, Ordering::Relaxed);
         tracing::info!(
@@ -75,6 +75,13 @@ impl AccessLog {
         );
     }
 
+    /// Increment request counter only — for use with async logging channel.
+    /// Callers send the entry to the log channel; a background task calls record().
+    #[allow(dead_code)]
+    pub fn count(&self) {
+        self.total_entries.fetch_add(1, Ordering::Relaxed);
+    }
+
     /// Get total number of logged entries
     #[allow(dead_code)]
     pub fn total_entries(&self) -> u64 {

src/observability/tracing.rs

@@ -184,32 +184,38 @@ pub enum PropagationFormat {
     B3,
 }
 
-/// Extract trace context from request headers
-pub fn extract_trace_context(headers: &HashMap<String, String>) -> Option<TraceContext> {
+/// Extract trace context from request headers.
+///
+/// Accepts `&http::HeaderMap` directly — avoids the HashMap<String,String>
+/// allocation that was previously needed to convert hyper headers.
+pub fn extract_trace_context(headers: &http::HeaderMap) -> Option<TraceContext> {
+    let hdr = |name: &str| -> Option<&str> {
+        headers.get(name).and_then(|v| v.to_str().ok())
+    };
+
     // Try W3C traceparent first
-    if let Some(traceparent) = headers.get(TRACEPARENT_HEADER) {
+    if let Some(traceparent) = hdr(TRACEPARENT_HEADER) {
         if let Some(mut ctx) = TraceContext::from_traceparent(traceparent) {
-            ctx.trace_state = headers.get(TRACESTATE_HEADER).cloned();
+            ctx.trace_state = hdr(TRACESTATE_HEADER).map(|s| s.to_string());
             return Some(ctx);
         }
     }
 
     // Try B3 single header
-    if let Some(b3) = headers.get(B3_HEADER) {
+    if let Some(b3) = hdr(B3_HEADER) {
         return TraceContext::from_b3_single(b3);
     }
 
     // Try B3 multi-header
-    if let Some(trace_id) = headers.get(B3_TRACE_ID_HEADER) {
-        if let Some(span_id) = headers.get(B3_SPAN_ID_HEADER) {
-            let sampled = headers
-                .get(B3_SAMPLED_HEADER)
+    if let Some(trace_id) = hdr(B3_TRACE_ID_HEADER) {
+        if let Some(span_id) = hdr(B3_SPAN_ID_HEADER) {
+            let sampled = hdr(B3_SAMPLED_HEADER)
                 .map(|s| s == "1" || s == "true")
                 .unwrap_or(true);
 
             return Some(TraceContext {
-                trace_id: trace_id.clone(),
-                parent_span_id: span_id.clone(),
+                trace_id: trace_id.to_string(),
+                parent_span_id: span_id.to_string(),
                 span_id: generate_span_id(),
                 trace_flags: if sampled { 1 } else { 0 },
                 trace_state: None,
@@ -542,12 +548,17 @@ mod tests {
 
     #[test]
     fn test_extract_w3c() {
-        let mut headers = HashMap::new();
+        let mut headers = http::HeaderMap::new();
         headers.insert(
-            "traceparent".to_string(),
-            "00-0af7651916cd43dd8448eb211c80319c-00f067aa0ba902b7-01".to_string(),
+            http::header::HeaderName::from_static("traceparent"),
+            "00-0af7651916cd43dd8448eb211c80319c-00f067aa0ba902b7-01"
+                .parse()
+                .unwrap(),
+        );
+        headers.insert(
+            http::header::HeaderName::from_static("tracestate"),
+            "vendor=value".parse().unwrap(),
         );
-        headers.insert("tracestate".to_string(), "vendor=value".to_string());
 
         let ctx = extract_trace_context(&headers).unwrap();
         assert_eq!(ctx.trace_id, "0af7651916cd43dd8448eb211c80319c");
@@ -556,10 +567,12 @@ mod tests {
 
     #[test]
     fn test_extract_b3_single() {
-        let mut headers = HashMap::new();
+        let mut headers = http::HeaderMap::new();
         headers.insert(
-            "b3".to_string(),
-            "463ac35c9f6413ad48485a3953bb6124-0020000000000001-1".to_string(),
+            http::header::HeaderName::from_static("b3"),
+            "463ac35c9f6413ad48485a3953bb6124-0020000000000001-1"
+                .parse()
+                .unwrap(),
         );
 
         let ctx = extract_trace_context(&headers).unwrap();
@@ -568,13 +581,19 @@ mod tests {
 
     #[test]
     fn test_extract_b3_multi() {
-        let mut headers = HashMap::new();
+        let mut headers = http::HeaderMap::new();
+        headers.insert(
+            http::header::HeaderName::from_static("x-b3-traceid"),
+            "463ac35c9f6413ad48485a3953bb6124".parse().unwrap(),
+        );
+        headers.insert(
+            http::header::HeaderName::from_static("x-b3-spanid"),
+            "0020000000000001".parse().unwrap(),
+        );
         headers.insert(
-            "x-b3-traceid".to_string(),
-            "463ac35c9f6413ad48485a3953bb6124".to_string(),
+            http::header::HeaderName::from_static("x-b3-sampled"),
+            "1".parse().unwrap(),
         );
-        headers.insert("x-b3-spanid".to_string(), "0020000000000001".to_string());
-        headers.insert("x-b3-sampled".to_string(), "1".to_string());
 
         let ctx = extract_trace_context(&headers).unwrap();
         assert_eq!(ctx.trace_id, "463ac35c9f6413ad48485a3953bb6124");
@@ -583,7 +602,7 @@ mod tests {
 
     #[test]
     fn test_extract_no_trace() {
-        let headers = HashMap::new();
+        let headers = http::HeaderMap::new();
         assert!(extract_trace_context(&headers).is_none());
     }
 

src/proxy/http_proxy.rs

@@ -23,6 +23,8 @@ impl HttpProxy {
         let client = reqwest::Client::builder()
             .timeout(timeout)
             .pool_max_idle_per_host(100)
+            .pool_idle_timeout(Duration::from_secs(90))
+            .tcp_nodelay(true)
             .build()
             .unwrap_or_default();