v25.11.2

changelog.md

@@ -1,5 +1,76 @@
 # Changelog
 
+## 2026-03-15 - 25.11.2 - fix(rustproxy-http)
+avoid reusing HTTP/1 senders during streaming responses and relax HTTP/2 keep-alive timeouts
+
+- Stop returning HTTP/1 senders to the connection pool before upstream response bodies finish streaming to prevent unsafe reuse on active connections.
+- Increase HTTP/2 keep-alive timeout from 5 seconds to 30 seconds in proxy connection builders to better support longer-lived backend streams.
+- Improves reliability for large streaming payloads and backend fallback request handling.
+
+## 2026-03-15 - 25.11.1 - fix(rustproxy-http)
+keep connection idle tracking alive during streaming and tune HTTP/2 connection lifetimes
+
+- Propagate connection activity tracking through HTTP/1, HTTP/2, and WebSocket forwarding so active request and response body streams do not trigger the idle watchdog.
+- Update CountingBody to refresh connection activity timestamps while data frames are polled during uploads and downloads.
+- Increase pooled HTTP/2 max age and set explicit HTTP/2 connection window sizes to improve long-lived streaming behavior.
+
+## 2026-03-15 - 25.11.0 - feat(rustproxy-http)
+add HTTP/2 Extended CONNECT WebSocket proxy support
+
+- Enable HTTP/2 CONNECT protocol support on the Hyper auto connection builder
+- Detect WebSocket requests for both HTTP/1 Upgrade and HTTP/2 Extended CONNECT flows
+- Translate HTTP/2 WebSocket requests to an HTTP/1.1 backend handshake and return RFC-compliant client responses
+
+## 2026-03-12 - 25.10.7 - fix(rustproxy-http)
+remove Host header from HTTP/2 upstream requests while preserving it for HTTP/1 retries
+
+- strips the Host header before sending HTTP/2 upstream requests so :authority from the URI is used instead
+- avoids 400 responses from nginx caused by sending both Host and :authority headers
+- keeps a cloned header set for bodyless request retries so HTTP/1 fallback still retains the Host header
+
+## 2026-03-12 - 25.10.6 - fix(rustproxy-http)
+use the requested domain as HTTP/2 authority instead of the backend host and port
+
+- build HTTP/2 absolute URIs from the client-facing domain so the :authority pseudo-header matches the Host header
+- remove backend port from generated HTTP/2 request URIs and fall back to the upstream host only when no domain is available
+- apply the authority handling consistently across pooled, inline, and generic upstream request paths
+
+## 2026-03-12 - 25.10.5 - fix(rustproxy-http)
+configure HTTP/2 client builders with a Tokio timer for keep-alive handling
+
+- Adds TokioTimer to all HTTP/2 client builder instances in proxy_service.
+- Ensures configured HTTP/2 keep-alive interval and timeout settings have the required timer runtime support.
+
+## 2026-03-12 - 25.10.4 - fix(rustproxy-http)
+stabilize upstream HTTP/2 forwarding and fallback behavior
+
+- Remove hop-by-hop headers before forwarding requests to HTTP/2 backends to comply with RFC 9113.
+- Use ALPN-enabled TLS configuration whenever HTTP/2 is possible, including explicit H2 connections and retries.
+- Add HTTP/2 handshake timeouts, tuned connection settings, and fallback to HTTP/1 when H2 negotiation times out or fails.
+- Register pooled HTTP/2 senders only after a successful first request to avoid reusing broken connections.
+- Build absolute URIs for HTTP/2 upstream requests so pseudo-headers such as scheme and authority are derived correctly.
+
+## 2026-03-12 - 25.10.3 - fix(rustproxy-http)
+include request domain in backend proxy error and protocol detection logs
+
+- Adds domain context to backend TCP/TLS connect, handshake, request failure, retry, and fallback log entries in the Rust HTTP proxy service.
+- Propagates the resolved host/domain through H1, H2, pooled, and fallback forwarding paths so backend-level diagnostics can be correlated with the original request domain.
+
+## 2026-03-12 - 25.10.2 - fix(repo)
+no code changes to release
+
+
+## 2026-03-12 - 25.10.1 - fix(repo)
+no changes to commit
+
+
+## 2026-03-12 - 25.10.0 - feat(metrics)
+add per-backend connection, error, protocol, and pool metrics with stale backend pruning
+
+- tracks backend connection lifecycle, connect timing, protocol detection, pool hit/miss rates, handshake/request errors, and h2 fallback failures in Rust metrics
+- exposes backend metrics through the TypeScript metrics adapter with backend listings, protocol lookup, and top error summaries
+- prunes backend metrics for backends no longer referenced by active routes, including preserved-port targets expanded across listening ports
+
 ## 2026-03-11 - 25.9.3 - fix(rustproxy-http)
 Evict stale HTTP/2 pooled senders and retry bodyless requests with fresh backend connections to avoid 502s
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@push.rocks/smartproxy",
-  "version": "25.9.3",
+  "version": "25.11.2",
   "private": false,
   "description": "A powerful proxy package with unified route-based configuration for high traffic management. Features include SSL/TLS support, flexible routing patterns, WebSocket handling, advanced security options, and automatic ACME certificate management.",
   "main": "dist_ts/index.js",

rust/crates/rustproxy-http/src/connection_pool.rs

@@ -20,7 +20,7 @@ const IDLE_TIMEOUT: Duration = Duration::from_secs(90);
 const EVICTION_INTERVAL: Duration = Duration::from_secs(30);
 /// Maximum age for pooled HTTP/2 connections before proactive eviction.
 /// Prevents staleness from backends that close idle connections (e.g. nginx GOAWAY).
-const MAX_H2_AGE: Duration = Duration::from_secs(120);
+const MAX_H2_AGE: Duration = Duration::from_secs(300);
 
 /// Identifies a unique backend endpoint.
 #[derive(Clone, Debug, Hash, Eq, PartialEq)]

rust/crates/rustproxy-http/src/counting_body.rs

@@ -25,6 +25,11 @@ pub struct CountingBody<B> {
     direction: Direction,
     /// Whether we've already reported the bytes (to avoid double-reporting on drop).
     reported: bool,
+    /// Optional connection-level activity tracker. When set, poll_frame updates this
+    /// to keep the idle watchdog alive during active body streaming (uploads/downloads).
+    connection_activity: Option<Arc<AtomicU64>>,
+    /// Start instant for computing elapsed ms for connection_activity.
+    activity_start: Option<std::time::Instant>,
 }
 
 /// Which direction the bytes flow.
@@ -53,9 +58,20 @@ impl<B> CountingBody<B> {
             source_ip,
             direction,
             reported: false,
+            connection_activity: None,
+            activity_start: None,
         }
     }
 
+    /// Set the connection-level activity tracker. When set, each data frame
+    /// updates this timestamp to prevent the idle watchdog from killing the
+    /// connection during active body streaming.
+    pub fn with_connection_activity(mut self, activity: Arc<AtomicU64>, start: std::time::Instant) -> Self {
+        self.connection_activity = Some(activity);
+        self.activity_start = Some(start);
+        self
+    }
+
     /// Report accumulated bytes to the metrics collector.
     fn report(&mut self) {
         if self.reported {
@@ -103,6 +119,10 @@ where
             Poll::Ready(Some(Ok(frame))) => {
                 if let Some(data) = frame.data_ref() {
                     this.counted_bytes.fetch_add(data.len() as u64, Ordering::Relaxed);
+                    // Keep the connection-level idle watchdog alive during body streaming
+                    if let (Some(activity), Some(start)) = (&this.connection_activity, &this.activity_start) {
+                        activity.store(start.elapsed().as_millis() as u64, Ordering::Relaxed);
+                    }
                 }
                 Poll::Ready(Some(Ok(frame)))
             }

rust/crates/rustproxy-http/src/proxy_service.rs

@@ -33,6 +33,14 @@ use crate::request_filter::RequestFilter;
 use crate::response_filter::ResponseFilter;
 use crate::upstream_selector::UpstreamSelector;
 
+/// Per-connection context for keeping the idle watchdog alive during body streaming.
+/// Passed through the forwarding chain so CountingBody can update the timestamp.
+#[derive(Clone)]
+struct ConnActivity {
+    last_activity: Arc<AtomicU64>,
+    start: std::time::Instant,
+}
+
 /// Default upstream connect timeout (30 seconds).
 const DEFAULT_CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
 
@@ -294,8 +302,9 @@ impl HttpProxyService {
             let cn = cancel_inner.clone();
             let la = Arc::clone(&la_inner);
             let st = start;
+            let ca = ConnActivity { last_activity: Arc::clone(&la_inner), start };
             async move {
-                let result = svc.handle_request(req, peer, port, cn).await;
+                let result = svc.handle_request(req, peer, port, cn, ca).await;
                 // Mark request end — update activity timestamp before guard drops
                 la.store(st.elapsed().as_millis() as u64, Ordering::Relaxed);
                 drop(req_guard); // Explicitly drop to decrement active_requests
@@ -304,8 +313,13 @@ impl HttpProxyService {
         });
 
         // Auto-detect h1 vs h2 based on ALPN / connection preface.
-        // serve_connection_with_upgrades supports h1 Upgrade (WebSocket) and h2 CONNECT.
-        let builder = hyper_util::server::conn::auto::Builder::new(hyper_util::rt::TokioExecutor::new());
+        // serve_connection_with_upgrades supports h1 Upgrade (WebSocket) and h2 Extended CONNECT (RFC 8441).
+        let mut builder = hyper_util::server::conn::auto::Builder::new(hyper_util::rt::TokioExecutor::new());
+        // Configure H2 server settings: Extended CONNECT for WebSocket + flow control tuning
+        builder.http2()
+            .enable_connect_protocol()
+            .initial_stream_window_size(2 * 1024 * 1024)        // 2MB per stream (vs default 64KB)
+            .initial_connection_window_size(8 * 1024 * 1024);   // 8MB per client connection
         let conn = builder.serve_connection_with_upgrades(io, service);
         // Pin on the heap — auto::UpgradeableConnection is !Unpin
         let mut conn = Box::pin(conn);
@@ -365,6 +379,7 @@ impl HttpProxyService {
         peer_addr: std::net::SocketAddr,
         port: u16,
         cancel: CancellationToken,
+        conn_activity: ConnActivity,
     ) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {
         let host = req.headers()
             .get("host")
@@ -479,18 +494,26 @@ impl HttpProxyService {
         }
 
         let upstream_key = format!("{}:{}", upstream.host, upstream.port);
+        let domain_str = host.as_deref().unwrap_or("-");
         self.upstream_selector.connection_started(&upstream_key);
 
-        // Check for WebSocket upgrade
-        let is_websocket = req.headers()
+        // Check for WebSocket upgrade: H1 (Upgrade header) or H2 Extended CONNECT (RFC 8441)
+        let is_h1_websocket = req.headers()
             .get("upgrade")
             .and_then(|v| v.to_str().ok())
             .map(|v| v.eq_ignore_ascii_case("websocket"))
             .unwrap_or(false);
 
-        if is_websocket {
+        let is_h2_websocket = req.method() == hyper::Method::CONNECT
+            && req.extensions()
+                .get::<hyper::ext::Protocol>()
+                .map(|p| p.as_str().eq_ignore_ascii_case("websocket"))
+                .unwrap_or(false);
+
+        if is_h1_websocket || is_h2_websocket {
             let result = self.handle_websocket_upgrade(
-                req, peer_addr, &upstream, route_match.route, route_id, &upstream_key, cancel, &ip_str,
+                req, peer_addr, &upstream, route_match.route, route_id, &upstream_key, cancel, &ip_str, is_h2_websocket,
+                if is_h2_websocket { Some(conn_activity.clone()) } else { None },
             ).await;
             // Note: for WebSocket, connection_ended is called inside
             // the spawned tunnel task when the connection closes.
@@ -538,6 +561,15 @@ impl HttpProxyService {
             }
         }
 
+        // Remove hop-by-hop headers (RFC 9113 §8.2.2 forbids connection-specific headers in H2)
+        upstream_headers.remove("connection");
+        upstream_headers.remove("keep-alive");
+        upstream_headers.remove("proxy-connection");
+        upstream_headers.remove("transfer-encoding");
+        upstream_headers.remove("te");
+        upstream_headers.remove("trailer");
+        upstream_headers.remove("upgrade");
+
         // Add standard reverse-proxy headers (X-Forwarded-*)
         {
             let original_host = host.as_deref().unwrap_or("");
@@ -618,9 +650,11 @@ impl HttpProxyService {
             // H2 pool checkout (H2 senders are Clone and multiplexed)
             if use_h2 {
                 if let Some(sender) = self.connection_pool.checkout_h2(&pool_key) {
+                    self.metrics.backend_pool_hit(&upstream_key);
+                    self.metrics.set_backend_protocol(&upstream_key, "h2");
                     let result = self.forward_h2_pooled(
                         sender, parts, body, upstream_headers, &upstream_path,
-                        route_match.route, route_id, &ip_str, &pool_key,
+                        route_match.route, route_id, &ip_str, &pool_key, domain_str, &conn_activity,
                     ).await;
                     self.upstream_selector.connection_ended(&upstream_key);
                     return result;
@@ -629,14 +663,17 @@ impl HttpProxyService {
         }
 
         // --- Fresh connection path ---
-        // Choose TLS config: use ALPN config for auto-detect probe, plain config otherwise
-        let tls_config = if needs_alpn_probe {
+        self.metrics.backend_pool_miss(&upstream_key);
+
+        // Choose TLS config: use ALPN config when H2 is possible (auto-detect probe OR explicit H2)
+        let tls_config = if needs_alpn_probe || use_h2 {
             &self.backend_tls_config_alpn
         } else {
             &self.backend_tls_config
         };
 
         // Establish backend connection
+        let connect_start = std::time::Instant::now();
         let (backend, detected_h2) = if upstream.use_tls {
             match tokio::time::timeout(
                 self.connect_timeout,
@@ -661,25 +698,42 @@ impl HttpProxyService {
                         };
                         self.protocol_cache.insert(cache_key, detected);
 
-                        debug!(
-                            "Auto-detected {} for backend {}:{}",
-                            if is_h2 { "HTTP/2" } else { "HTTP/1.1" },
-                            upstream.host, upstream.port
+                        info!(
+                            backend = %upstream_key,
+                            domain = %domain_str,
+                            protocol = if is_h2 { "h2" } else { "h1" },
+                            connect_time_ms = %connect_start.elapsed().as_millis(),
+                            "Backend protocol detected via ALPN"
                         );
 
                         is_h2
                     } else {
                         use_h2
                     };
+                    self.metrics.backend_connection_opened(&upstream_key, connect_start.elapsed());
+                    self.metrics.set_backend_protocol(&upstream_key, if final_h2 { "h2" } else { "h1" });
                     (BackendStream::Tls(tls), final_h2)
                 }
                 Ok(Err(e)) => {
-                    error!("Failed TLS connect to upstream {}:{}: {}", upstream.host, upstream.port, e);
+                    error!(
+                        backend = %upstream_key,
+                        domain = %domain_str,
+                        connect_time_ms = %connect_start.elapsed().as_millis(),
+                        error = %e,
+                        "Backend TLS connect failed"
+                    );
+                    self.metrics.backend_connect_error(&upstream_key);
                     self.upstream_selector.connection_ended(&upstream_key);
                     return Ok(error_response(StatusCode::BAD_GATEWAY, "Backend TLS unavailable"));
                 }
                 Err(_) => {
-                    error!("Upstream TLS connect timeout for {}:{}", upstream.host, upstream.port);
+                    error!(
+                        backend = %upstream_key,
+                        domain = %domain_str,
+                        connect_time_ms = %connect_start.elapsed().as_millis(),
+                        "Backend TLS connect timeout"
+                    );
+                    self.metrics.backend_connect_error(&upstream_key);
                     self.upstream_selector.connection_ended(&upstream_key);
                     return Ok(error_response(StatusCode::GATEWAY_TIMEOUT, "Backend TLS connect timeout"));
                 }
@@ -694,15 +748,30 @@ impl HttpProxyService {
                     let _ = socket2::SockRef::from(&s).set_tcp_keepalive(
                         &socket2::TcpKeepalive::new().with_time(std::time::Duration::from_secs(60))
                     );
+                    self.metrics.backend_connection_opened(&upstream_key, connect_start.elapsed());
+                    self.metrics.set_backend_protocol(&upstream_key, if use_h2 { "h2" } else { "h1" });
                     (BackendStream::Plain(s), use_h2)
                 }
                 Ok(Err(e)) => {
-                    error!("Failed to connect to upstream {}:{}: {}", upstream.host, upstream.port, e);
+                    error!(
+                        backend = %upstream_key,
+                        domain = %domain_str,
+                        connect_time_ms = %connect_start.elapsed().as_millis(),
+                        error = %e,
+                        "Backend TCP connect failed"
+                    );
+                    self.metrics.backend_connect_error(&upstream_key);
                     self.upstream_selector.connection_ended(&upstream_key);
                     return Ok(error_response(StatusCode::BAD_GATEWAY, "Backend unavailable"));
                 }
                 Err(_) => {
-                    error!("Upstream connect timeout for {}:{}", upstream.host, upstream.port);
+                    error!(
+                        backend = %upstream_key,
+                        domain = %domain_str,
+                        connect_time_ms = %connect_start.elapsed().as_millis(),
+                        "Backend TCP connect timeout"
+                    );
+                    self.metrics.backend_connect_error(&upstream_key);
                     self.upstream_selector.connection_ended(&upstream_key);
                     return Ok(error_response(StatusCode::GATEWAY_TIMEOUT, "Backend connect timeout"));
                 }
@@ -724,22 +793,23 @@ impl HttpProxyService {
                 self.forward_h2_with_fallback(
                     io, parts, body, upstream_headers, &upstream_path,
                     &upstream, route_match.route, route_id, &ip_str, &final_pool_key,
-                    host.clone(),
+                    host.clone(), domain_str, &conn_activity,
                 ).await
             } else {
                 // Explicit H2 mode: hard-fail on handshake error (preserved behavior)
                 self.forward_h2(
                     io, parts, body, upstream_headers, &upstream_path,
-                    &upstream, route_match.route, route_id, &ip_str, &final_pool_key,
+                    &upstream, route_match.route, route_id, &ip_str, &final_pool_key, domain_str, &conn_activity,
                 ).await
             }
         } else {
             self.forward_h1(
                 io, parts, body, upstream_headers, &upstream_path,
-                &upstream, route_match.route, route_id, &ip_str, &final_pool_key,
+                &upstream, route_match.route, route_id, &ip_str, &final_pool_key, domain_str, &conn_activity,
             ).await
         };
         self.upstream_selector.connection_ended(&upstream_key);
+        self.metrics.backend_connection_closed(&upstream_key);
         result
     }
 
@@ -757,12 +827,17 @@ impl HttpProxyService {
         route_id: Option<&str>,
         source_ip: &str,
         pool_key: &crate::connection_pool::PoolKey,
+        domain: &str,
+        conn_activity: &ConnActivity,
     ) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {
+        let backend_key = format!("{}:{}", pool_key.host, pool_key.port);
+
         // Try pooled H1 connection first — avoids TCP+TLS handshake
         if let Some(pooled_sender) = self.connection_pool.checkout_h1(pool_key) {
+            self.metrics.backend_pool_hit(&backend_key);
             return self.forward_h1_with_sender(
                 pooled_sender, parts, body, upstream_headers, upstream_path,
-                route, route_id, source_ip, pool_key,
+                route, route_id, source_ip, pool_key, domain, conn_activity,
             ).await;
         }
 
@@ -773,7 +848,8 @@ impl HttpProxyService {
         ) = match hyper::client::conn::http1::handshake(io).await {
             Ok(h) => h,
             Err(e) => {
-                error!("Upstream handshake failed: {}", e);
+                error!(backend = %backend_key, domain = %domain, error = %e, "Backend H1 handshake failed");
+                self.metrics.backend_handshake_error(&backend_key);
                 return Ok(error_response(StatusCode::BAD_GATEWAY, "Backend handshake failed"));
             }
         };
@@ -784,7 +860,7 @@ impl HttpProxyService {
             }
         });
 
-        self.forward_h1_with_sender(sender, parts, body, upstream_headers, upstream_path, route, route_id, source_ip, pool_key).await
+        self.forward_h1_with_sender(sender, parts, body, upstream_headers, upstream_path, route, route_id, source_ip, pool_key, domain, conn_activity).await
     }
 
     /// Common H1 forwarding logic used by both fresh and pooled paths.
@@ -799,6 +875,8 @@ impl HttpProxyService {
         route_id: Option<&str>,
         source_ip: &str,
         pool_key: &crate::connection_pool::PoolKey,
+        domain: &str,
+        conn_activity: &ConnActivity,
     ) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {
         // Always use HTTP/1.1 for h1 backend connections (h2 incoming requests have version HTTP/2.0)
         let mut upstream_req = Request::builder()
@@ -817,7 +895,7 @@ impl HttpProxyService {
             route_id.map(|s| s.to_string()),
             Some(source_ip.to_string()),
             Direction::In,
-        );
+        ).with_connection_activity(Arc::clone(&conn_activity.last_activity), conn_activity.start);
         let boxed_body: BoxBody<Bytes, hyper::Error> = BoxBody::new(counting_req_body);
 
         let upstream_req = upstream_req.body(boxed_body).unwrap();
@@ -825,15 +903,24 @@ impl HttpProxyService {
         let upstream_response = match sender.send_request(upstream_req).await {
             Ok(resp) => resp,
             Err(e) => {
-                error!("Upstream request failed: {}", e);
+                let bk = format!("{}:{}", pool_key.host, pool_key.port);
+                error!(backend = %bk, domain = %domain, error = %e, "Backend H1 request failed");
+                self.metrics.backend_request_error(&bk);
                 return Ok(error_response(StatusCode::BAD_GATEWAY, "Backend request failed"));
             }
         };
 
-        // Return sender to pool (body streams lazily, sender is reusable once response head is received)
-        self.connection_pool.checkin_h1(pool_key.clone(), sender);
+        // Note: we do NOT return the sender to the pool here because the response body
+        // hasn't been fully streamed yet. Pooling a sender while its response body is still
+        // in-flight risks another request being dispatched on the same connection if is_ready()
+        // momentarily returns true between chunks. The sender is dropped after this scope,
+        // and the backend connection remains alive via the spawned conn driver task until
+        // the response body finishes streaming.
+        // For small/empty responses, the sender could theoretically be reused, but the safety
+        // of large streaming responses (e.g. 352MB Docker layers) takes priority.
+        drop(sender);
 
-        self.build_streaming_response(upstream_response, route, route_id, source_ip).await
+        self.build_streaming_response(upstream_response, route, route_id, source_ip, conn_activity).await
     }
 
     /// Forward request to backend via HTTP/2 with body streaming (fresh connection).
@@ -850,18 +937,33 @@ impl HttpProxyService {
         route_id: Option<&str>,
         source_ip: &str,
         pool_key: &crate::connection_pool::PoolKey,
+        domain: &str,
+        conn_activity: &ConnActivity,
     ) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {
+        let backend_key = format!("{}:{}", pool_key.host, pool_key.port);
         let exec = hyper_util::rt::TokioExecutor::new();
-        // Explicitly type the handshake with BoxBody for uniform pool type
+        let mut h2_builder = hyper::client::conn::http2::Builder::new(exec);
+        h2_builder
+            .timer(hyper_util::rt::TokioTimer::new())
+            .keep_alive_interval(std::time::Duration::from_secs(10))
+            .keep_alive_timeout(std::time::Duration::from_secs(30))
+            .initial_stream_window_size(2 * 1024 * 1024)
+            .initial_connection_window_size(16 * 1024 * 1024);
         let (sender, conn): (
             hyper::client::conn::http2::SendRequest<BoxBody<Bytes, hyper::Error>>,
             hyper::client::conn::http2::Connection<TokioIo<BackendStream>, BoxBody<Bytes, hyper::Error>, hyper_util::rt::TokioExecutor>,
-        ) = match hyper::client::conn::http2::handshake(exec, io).await {
-            Ok(h) => h,
-            Err(e) => {
-                error!("HTTP/2 upstream handshake failed: {}", e);
+        ) = match tokio::time::timeout(self.connect_timeout, h2_builder.handshake(io)).await {
+            Ok(Ok(h)) => h,
+            Ok(Err(e)) => {
+                error!(backend = %backend_key, domain = %domain, error = %e, "Backend H2 handshake failed");
+                self.metrics.backend_handshake_error(&backend_key);
                 return Ok(error_response(StatusCode::BAD_GATEWAY, "Backend H2 handshake failed"));
             }
+            Err(_) => {
+                error!(backend = %backend_key, domain = %domain, "Backend H2 handshake timeout");
+                self.metrics.backend_handshake_error(&backend_key);
+                return Ok(error_response(StatusCode::GATEWAY_TIMEOUT, "Backend H2 handshake timeout"));
+            }
         };
 
         tokio::spawn(async move {
@@ -870,10 +972,13 @@ impl HttpProxyService {
             }
         });
 
-        // Register for multiplexed reuse
-        self.connection_pool.register_h2(pool_key.clone(), sender.clone());
-
-        self.forward_h2_with_sender(sender, parts, body, upstream_headers, upstream_path, route, route_id, source_ip, Some(pool_key)).await
+        // Clone sender for potential pool registration; register only after first request succeeds
+        let sender_for_pool = sender.clone();
+        let result = self.forward_h2_with_sender(sender, parts, body, upstream_headers, upstream_path, route, route_id, source_ip, Some(pool_key), domain, conn_activity).await;
+        if matches!(&result, Ok(ref resp) if resp.status() != StatusCode::BAD_GATEWAY) {
+            self.connection_pool.register_h2(pool_key.clone(), sender_for_pool);
+        }
+        result
     }
 
     /// Forward request using an existing (pooled) HTTP/2 sender.
@@ -890,6 +995,8 @@ impl HttpProxyService {
         route_id: Option<&str>,
         source_ip: &str,
         pool_key: &crate::connection_pool::PoolKey,
+        domain: &str,
+        conn_activity: &ConnActivity,
     ) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {
         // Save retry state for bodyless requests (cheap: Method is an enum, HeaderMap clones Arc-backed Bytes)
         let retry_state = if body.is_end_stream() {
@@ -900,18 +1007,18 @@ impl HttpProxyService {
 
         let result = self.forward_h2_with_sender(
             sender, parts, body, upstream_headers, upstream_path,
-            route, route_id, source_ip, Some(pool_key),
+            route, route_id, source_ip, Some(pool_key), domain, conn_activity,
         ).await;
 
         // If the request failed (502) and we can retry with an empty body, do so
         let is_502 = matches!(&result, Ok(resp) if resp.status() == StatusCode::BAD_GATEWAY);
         if is_502 {
             if let Some((method, headers)) = retry_state {
-                warn!("Stale pooled H2 sender for {}:{}, retrying with fresh connection",
-                    pool_key.host, pool_key.port);
+                warn!(backend = %format!("{}:{}", pool_key.host, pool_key.port), domain = %domain,
+                    "Stale pooled H2 sender, retrying with fresh connection");
                 return self.retry_h2_with_fresh_connection(
                     method, headers, upstream_path,
-                    pool_key, route, route_id, source_ip,
+                    pool_key, route, route_id, source_ip, domain, conn_activity,
                 ).await;
             }
         }
@@ -929,20 +1036,27 @@ impl HttpProxyService {
         route: &rustproxy_config::RouteConfig,
         route_id: Option<&str>,
         source_ip: &str,
+        domain: &str,
+        conn_activity: &ConnActivity,
     ) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {
+        let backend_key = format!("{}:{}", pool_key.host, pool_key.port);
+
         // Establish fresh backend connection
+        let retry_connect_start = std::time::Instant::now();
         let backend = if pool_key.use_tls {
             match tokio::time::timeout(
                 self.connect_timeout,
-                connect_tls_backend(&self.backend_tls_config, &pool_key.host, pool_key.port),
+                connect_tls_backend(&self.backend_tls_config_alpn, &pool_key.host, pool_key.port),
             ).await {
                 Ok(Ok(tls)) => BackendStream::Tls(tls),
                 Ok(Err(e)) => {
-                    error!("H2 retry: TLS connect failed for {}:{}: {}", pool_key.host, pool_key.port, e);
+                    error!(backend = %backend_key, domain = %domain, error = %e, "H2 retry: TLS connect failed");
+                    self.metrics.backend_connect_error(&backend_key);
                     return Ok(error_response(StatusCode::BAD_GATEWAY, "Backend unavailable on H2 retry"));
                 }
                 Err(_) => {
-                    error!("H2 retry: TLS connect timeout for {}:{}", pool_key.host, pool_key.port);
+                    error!(backend = %backend_key, domain = %domain, "H2 retry: TLS connect timeout");
+                    self.metrics.backend_connect_error(&backend_key);
                     return Ok(error_response(StatusCode::GATEWAY_TIMEOUT, "Backend timeout on H2 retry"));
                 }
             }
@@ -956,27 +1070,45 @@ impl HttpProxyService {
                     BackendStream::Plain(s)
                 }
                 Ok(Err(e)) => {
-                    error!("H2 retry: connect failed for {}:{}: {}", pool_key.host, pool_key.port, e);
+                    error!(backend = %backend_key, domain = %domain, error = %e, "H2 retry: TCP connect failed");
+                    self.metrics.backend_connect_error(&backend_key);
                     return Ok(error_response(StatusCode::BAD_GATEWAY, "Backend unavailable on H2 retry"));
                 }
                 Err(_) => {
-                    error!("H2 retry: connect timeout for {}:{}", pool_key.host, pool_key.port);
+                    error!(backend = %backend_key, domain = %domain, "H2 retry: TCP connect timeout");
+                    self.metrics.backend_connect_error(&backend_key);
                     return Ok(error_response(StatusCode::GATEWAY_TIMEOUT, "Backend timeout on H2 retry"));
                 }
             }
         };
+        self.metrics.backend_connection_opened(&backend_key, retry_connect_start.elapsed());
 
         let io = TokioIo::new(backend);
         let exec = hyper_util::rt::TokioExecutor::new();
+        let mut h2_builder = hyper::client::conn::http2::Builder::new(exec);
+        h2_builder
+            .timer(hyper_util::rt::TokioTimer::new())
+            .keep_alive_interval(std::time::Duration::from_secs(10))
+            .keep_alive_timeout(std::time::Duration::from_secs(30))
+            .initial_stream_window_size(2 * 1024 * 1024)
+            .initial_connection_window_size(16 * 1024 * 1024);
         let (mut sender, conn): (
             hyper::client::conn::http2::SendRequest<BoxBody<Bytes, hyper::Error>>,
             hyper::client::conn::http2::Connection<TokioIo<BackendStream>, BoxBody<Bytes, hyper::Error>, hyper_util::rt::TokioExecutor>,
-        ) = match hyper::client::conn::http2::handshake(exec, io).await {
-            Ok(h) => h,
-            Err(e) => {
-                error!("H2 retry: handshake failed for {}:{}: {}", pool_key.host, pool_key.port, e);
+        ) = match tokio::time::timeout(self.connect_timeout, h2_builder.handshake(io)).await {
+            Ok(Ok(h)) => h,
+            Ok(Err(e)) => {
+                error!(backend = %backend_key, domain = %domain, error = %e, "H2 retry: handshake failed");
+                self.metrics.backend_handshake_error(&backend_key);
+                self.metrics.backend_connection_closed(&backend_key);
                 return Ok(error_response(StatusCode::BAD_GATEWAY, "Backend H2 retry handshake failed"));
             }
+            Err(_) => {
+                error!(backend = %backend_key, domain = %domain, "H2 retry: handshake timeout");
+                self.metrics.backend_handshake_error(&backend_key);
+                self.metrics.backend_connection_closed(&backend_key);
+                return Ok(error_response(StatusCode::GATEWAY_TIMEOUT, "Backend H2 retry handshake timeout"));
+            }
         };
 
         tokio::spawn(async move {
@@ -985,13 +1117,17 @@ impl HttpProxyService {
             }
         });
 
-        // Register fresh sender in pool for future requests
-        self.connection_pool.register_h2(pool_key.clone(), sender.clone());
-
-        // Build request with empty body
+        // Build request with empty body using absolute URI for H2 pseudo-headers
+        let scheme = if pool_key.use_tls { "https" } else { "http" };
+        let authority = if domain != "-" { domain } else { pool_key.host.as_str() };
+        let h2_uri = format!("{}://{}{}", scheme, authority, upstream_path);
         let mut upstream_req = Request::builder()
             .method(method)
-            .uri(upstream_path);
+            .uri(&h2_uri);
+
+        // Remove Host header for H2 — :authority pseudo-header (from URI) is sufficient
+        let mut upstream_headers = upstream_headers;
+        upstream_headers.remove("host");
 
         if let Some(headers) = upstream_req.headers_mut() {
             *headers = upstream_headers;
@@ -1004,11 +1140,18 @@ impl HttpProxyService {
 
         match sender.send_request(upstream_req).await {
             Ok(resp) => {
-                self.build_streaming_response(resp, route, route_id, source_ip).await
+                // Register in pool only after request succeeds
+                self.connection_pool.register_h2(pool_key.clone(), sender);
+                let result = self.build_streaming_response(resp, route, route_id, source_ip, conn_activity).await;
+                // Close the fresh backend connection (opened above)
+                self.metrics.backend_connection_closed(&backend_key);
+                result
             }
             Err(e) => {
-                error!("H2 retry: request failed for {}:{}: {}", pool_key.host, pool_key.port, e);
-                self.connection_pool.remove_h2(pool_key);
+                error!(backend = %backend_key, domain = %domain, error = %e, "H2 retry: request failed");
+                self.metrics.backend_request_error(&backend_key);
+                // Close the fresh backend connection (opened above)
+                self.metrics.backend_connection_closed(&backend_key);
                 Ok(error_response(StatusCode::BAD_GATEWAY, "Backend H2 request failed on retry"))
             }
         }
@@ -1026,7 +1169,7 @@ impl HttpProxyService {
         io: TokioIo<BackendStream>,
         parts: hyper::http::request::Parts,
         body: Incoming,
-        upstream_headers: hyper::HeaderMap,
+        mut upstream_headers: hyper::HeaderMap,
         upstream_path: &str,
         upstream: &crate::upstream_selector::UpstreamSelection,
         route: &rustproxy_config::RouteConfig,
@@ -1034,15 +1177,63 @@ impl HttpProxyService {
         source_ip: &str,
         pool_key: &crate::connection_pool::PoolKey,
         requested_host: Option<String>,
+        domain: &str,
+        conn_activity: &ConnActivity,
     ) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {
         let exec = hyper_util::rt::TokioExecutor::new();
-        let handshake_result: Result<(
-            hyper::client::conn::http2::SendRequest<BoxBody<Bytes, hyper::Error>>,
-            hyper::client::conn::http2::Connection<TokioIo<BackendStream>, BoxBody<Bytes, hyper::Error>, hyper_util::rt::TokioExecutor>,
-        ), hyper::Error> = hyper::client::conn::http2::handshake(exec, io).await;
+        let mut h2_builder = hyper::client::conn::http2::Builder::new(exec);
+        h2_builder
+            .timer(hyper_util::rt::TokioTimer::new())
+            .keep_alive_interval(std::time::Duration::from_secs(10))
+            .keep_alive_timeout(std::time::Duration::from_secs(30))
+            .initial_stream_window_size(2 * 1024 * 1024)
+            .initial_connection_window_size(16 * 1024 * 1024);
+        let handshake_result = tokio::time::timeout(
+            self.connect_timeout,
+            h2_builder.handshake(io),
+        ).await;
 
         match handshake_result {
-            Ok((mut sender, conn)) => {
+            Err(_) => {
+                // H2 handshake timed out — fall back to H1
+                let bk = format!("{}:{}", upstream.host, upstream.port);
+                warn!(
+                    backend = %bk,
+                    domain = %domain,
+                    "H2 handshake timeout, falling back to H1"
+                );
+                self.metrics.backend_h2_failure(&bk);
+                self.metrics.backend_handshake_error(&bk);
+
+                let cache_key = crate::protocol_cache::ProtocolCacheKey {
+                    host: upstream.host.clone(),
+                    port: upstream.port,
+                    requested_host: requested_host.clone(),
+                };
+                self.protocol_cache.insert(cache_key, crate::protocol_cache::DetectedProtocol::H1);
+
+                match self.reconnect_backend(upstream, domain).await {
+                    Some(fallback_backend) => {
+                        let h1_pool_key = crate::connection_pool::PoolKey {
+                            host: upstream.host.clone(),
+                            port: upstream.port,
+                            use_tls: upstream.use_tls,
+                            h2: false,
+                        };
+                        let fallback_io = TokioIo::new(fallback_backend);
+                        let result = self.forward_h1(
+                            fallback_io, parts, body, upstream_headers, upstream_path,
+                            upstream, route, route_id, source_ip, &h1_pool_key, domain, conn_activity,
+                        ).await;
+                        self.metrics.backend_connection_closed(&bk);
+                        result
+                    }
+                    None => {
+                        Ok(error_response(StatusCode::BAD_GATEWAY, "Backend unavailable after H2 timeout fallback"))
+                    }
+                }
+            }
+            Ok(Ok((mut sender, conn))) => {
                 tokio::spawn(async move {
                     if let Err(e) = conn.await {
                         debug!("HTTP/2 upstream connection error: {}", e);
@@ -1050,18 +1241,25 @@ impl HttpProxyService {
                 });
 
                 // Save retry state before consuming parts/body (for bodyless requests like GET)
+                // Clone BEFORE removing Host — H1 fallback needs Host header
                 let retry_state = if body.is_end_stream() {
                     Some((parts.method.clone(), upstream_headers.clone()))
                 } else {
                     None
                 };
 
+                // Remove Host header for H2 — :authority pseudo-header (from URI) is sufficient
+                upstream_headers.remove("host");
+
                 // Build and send the h2 request inline (don't register in pool yet —
                 // we need to verify the request actually succeeds first, because some
                 // backends advertise h2 via ALPN but don't speak the h2 binary protocol).
+                let scheme = if upstream.use_tls { "https" } else { "http" };
+                let authority = if domain != "-" { domain } else { upstream.host.as_str() };
+                let h2_uri = format!("{}://{}{}", scheme, authority, upstream_path);
                 let mut upstream_req = Request::builder()
                     .method(parts.method)
-                    .uri(upstream_path);
+                    .uri(&h2_uri);
 
                 if let Some(headers) = upstream_req.headers_mut() {
                     *headers = upstream_headers;
@@ -1073,7 +1271,7 @@ impl HttpProxyService {
                     route_id.map(|s| s.to_string()),
                     Some(source_ip.to_string()),
                     Direction::In,
-                );
+                ).with_connection_activity(Arc::clone(&conn_activity.last_activity), conn_activity.start);
                 let boxed_body: BoxBody<Bytes, hyper::Error> = BoxBody::new(counting_req_body);
                 let upstream_req = upstream_req.body(boxed_body).unwrap();
 
@@ -1081,15 +1279,19 @@ impl HttpProxyService {
                     Ok(upstream_response) => {
                         // H2 works! Register sender in pool for multiplexed reuse
                         self.connection_pool.register_h2(pool_key.clone(), sender);
-                        self.build_streaming_response(upstream_response, route, route_id, source_ip).await
+                        self.build_streaming_response(upstream_response, route, route_id, source_ip, conn_activity).await
                     }
                     Err(e) => {
                         // H2 request failed — backend advertises h2 via ALPN but doesn't
                         // actually speak it. Update cache so future requests use H1.
+                        let bk = format!("{}:{}", upstream.host, upstream.port);
                         warn!(
-                            "Auto-detect: H2 request failed for {}:{}, falling back to H1: {}",
-                            upstream.host, upstream.port, e
+                            backend = %bk,
+                            domain = %domain,
+                            error = %e,
+                            "Auto-detect: H2 request failed, falling back to H1"
                         );
+                        self.metrics.backend_h2_failure(&bk);
                         let cache_key = crate::protocol_cache::ProtocolCacheKey {
                             host: upstream.host.clone(),
                             port: upstream.port,
@@ -1099,7 +1301,7 @@ impl HttpProxyService {
 
                         // Retry as H1 for bodyless requests; return 502 for requests with bodies
                         if let Some((method, headers)) = retry_state {
-                            match self.reconnect_backend(upstream).await {
+                            match self.reconnect_backend(upstream, domain).await {
                                 Some(fallback_backend) => {
                                     let h1_pool_key = crate::connection_pool::PoolKey {
                                         host: upstream.host.clone(),
@@ -1108,10 +1310,13 @@ impl HttpProxyService {
                                         h2: false,
                                     };
                                     let fallback_io = TokioIo::new(fallback_backend);
-                                    self.forward_h1_empty_body(
+                                    let result = self.forward_h1_empty_body(
                                         fallback_io, method, headers, upstream_path,
-                                        route, route_id, source_ip, &h1_pool_key,
-                                    ).await
+                                        route, route_id, source_ip, &h1_pool_key, domain, conn_activity,
+                                    ).await;
+                                    // Close the reconnected backend connection (opened in reconnect_backend)
+                                    self.metrics.backend_connection_closed(&bk);
+                                    result
                                 }
                                 None => {
                                     Ok(error_response(StatusCode::BAD_GATEWAY, "Backend unavailable after H2 fallback"))
@@ -1123,13 +1328,18 @@ impl HttpProxyService {
                     }
                 }
             }
-            Err(e) => {
+            Ok(Err(e)) => {
                 // H2 handshake truly failed — fall back to H1
                 // Body is NOT consumed yet, so we can retry the full request.
+                let bk = format!("{}:{}", upstream.host, upstream.port);
                 warn!(
-                    "H2 handshake failed for {}:{}, falling back to H1: {}",
-                    upstream.host, upstream.port, e
+                    backend = %bk,
+                    domain = %domain,
+                    error = %e,
+                    "H2 handshake failed, falling back to H1"
                 );
+                self.metrics.backend_h2_failure(&bk);
+                self.metrics.backend_handshake_error(&bk);
 
                 // Update cache to H1 so subsequent requests skip H2
                 let cache_key = crate::protocol_cache::ProtocolCacheKey {
@@ -1140,7 +1350,7 @@ impl HttpProxyService {
                 self.protocol_cache.insert(cache_key, crate::protocol_cache::DetectedProtocol::H1);
 
                 // Reconnect for H1 (the original io was consumed by the failed h2 handshake)
-                match self.reconnect_backend(upstream).await {
+                match self.reconnect_backend(upstream, domain).await {
                     Some(fallback_backend) => {
                         let h1_pool_key = crate::connection_pool::PoolKey {
                             host: upstream.host.clone(),
@@ -1149,10 +1359,13 @@ impl HttpProxyService {
                             h2: false,
                         };
                         let fallback_io = TokioIo::new(fallback_backend);
-                        self.forward_h1(
+                        let result = self.forward_h1(
                             fallback_io, parts, body, upstream_headers, upstream_path,
-                            upstream, route, route_id, source_ip, &h1_pool_key,
-                        ).await
+                            upstream, route, route_id, source_ip, &h1_pool_key, domain, conn_activity,
+                        ).await;
+                        // Close the reconnected backend connection (opened in reconnect_backend)
+                        self.metrics.backend_connection_closed(&bk);
+                        result
                     }
                     None => {
                         Ok(error_response(StatusCode::BAD_GATEWAY, "Backend unavailable after H2 fallback"))
@@ -1174,14 +1387,18 @@ impl HttpProxyService {
         route_id: Option<&str>,
         source_ip: &str,
         pool_key: &crate::connection_pool::PoolKey,
+        domain: &str,
+        conn_activity: &ConnActivity,
     ) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {
+        let backend_key = format!("{}:{}", pool_key.host, pool_key.port);
         let (mut sender, conn): (
             hyper::client::conn::http1::SendRequest<BoxBody<Bytes, hyper::Error>>,
             hyper::client::conn::http1::Connection<TokioIo<BackendStream>, BoxBody<Bytes, hyper::Error>>,
         ) = match hyper::client::conn::http1::handshake(io).await {
             Ok(h) => h,
             Err(e) => {
-                error!("H1 fallback: handshake failed: {}", e);
+                error!(backend = %backend_key, domain = %domain, error = %e, "H1 fallback: handshake failed");
+                self.metrics.backend_handshake_error(&backend_key);
                 return Ok(error_response(StatusCode::BAD_GATEWAY, "Backend H1 fallback handshake failed"));
             }
         };
@@ -1209,34 +1426,43 @@ impl HttpProxyService {
         let upstream_response = match sender.send_request(upstream_req).await {
             Ok(resp) => resp,
             Err(e) => {
-                error!("H1 fallback: request failed: {}", e);
+                error!(backend = %backend_key, domain = %domain, error = %e, "H1 fallback: request failed");
+                self.metrics.backend_request_error(&backend_key);
                 return Ok(error_response(StatusCode::BAD_GATEWAY, "Backend H1 fallback request failed"));
             }
         };
 
-        // Return sender to pool for keep-alive reuse
-        self.connection_pool.checkin_h1(pool_key.clone(), sender);
+        // Don't pool the sender while response body is still streaming (same safety as forward_h1_with_sender)
+        drop(sender);
 
-        self.build_streaming_response(upstream_response, route, route_id, source_ip).await
+        self.build_streaming_response(upstream_response, route, route_id, source_ip, conn_activity).await
     }
 
     /// Reconnect to a backend (used for H2→H1 fallback).
     async fn reconnect_backend(
         &self,
         upstream: &crate::upstream_selector::UpstreamSelection,
+        domain: &str,
     ) -> Option<BackendStream> {
+        let backend_key = format!("{}:{}", upstream.host, upstream.port);
+        let reconnect_start = std::time::Instant::now();
         if upstream.use_tls {
             match tokio::time::timeout(
                 self.connect_timeout,
                 connect_tls_backend(&self.backend_tls_config, &upstream.host, upstream.port),
             ).await {
-                Ok(Ok(tls)) => Some(BackendStream::Tls(tls)),
+                Ok(Ok(tls)) => {
+                    self.metrics.backend_connection_opened(&backend_key, reconnect_start.elapsed());
+                    Some(BackendStream::Tls(tls))
+                }
                 Ok(Err(e)) => {
-                    error!("H1 fallback: TLS reconnect failed for {}:{}: {}", upstream.host, upstream.port, e);
+                    error!(backend = %backend_key, domain = %domain, error = %e, "H1 fallback: TLS reconnect failed");
+                    self.metrics.backend_connect_error(&backend_key);
                     None
                 }
                 Err(_) => {
-                    error!("H1 fallback: TLS reconnect timeout for {}:{}", upstream.host, upstream.port);
+                    error!(backend = %backend_key, domain = %domain, "H1 fallback: TLS reconnect timeout");
+                    self.metrics.backend_connect_error(&backend_key);
                     None
                 }
             }
@@ -1250,14 +1476,17 @@ impl HttpProxyService {
                     let _ = socket2::SockRef::from(&s).set_tcp_keepalive(
                         &socket2::TcpKeepalive::new().with_time(std::time::Duration::from_secs(60))
                     );
+                    self.metrics.backend_connection_opened(&backend_key, reconnect_start.elapsed());
                     Some(BackendStream::Plain(s))
                 }
                 Ok(Err(e)) => {
-                    error!("H1 fallback: reconnect failed for {}:{}: {}", upstream.host, upstream.port, e);
+                    error!(backend = %backend_key, domain = %domain, error = %e, "H1 fallback: TCP reconnect failed");
+                    self.metrics.backend_connect_error(&backend_key);
                     None
                 }
                 Err(_) => {
-                    error!("H1 fallback: reconnect timeout for {}:{}", upstream.host, upstream.port);
+                    error!(backend = %backend_key, domain = %domain, "H1 fallback: TCP reconnect timeout");
+                    self.metrics.backend_connect_error(&backend_key);
                     None
                 }
             }
@@ -1276,10 +1505,24 @@ impl HttpProxyService {
         route_id: Option<&str>,
         source_ip: &str,
         pool_key: Option<&crate::connection_pool::PoolKey>,
+        domain: &str,
+        conn_activity: &ConnActivity,
     ) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {
+        // Build absolute URI for H2 pseudo-headers (:scheme, :authority)
+        // Use the requested domain as authority (not backend address) so :authority matches Host header
+        let scheme = if pool_key.map(|pk| pk.use_tls).unwrap_or(false) { "https" } else { "http" };
+        let authority = if domain != "-" { domain } else {
+            pool_key.map(|pk| pk.host.as_str()).unwrap_or("localhost")
+        };
+        let h2_uri = format!("{}://{}{}", scheme, authority, upstream_path);
         let mut upstream_req = Request::builder()
             .method(parts.method)
-            .uri(upstream_path);
+            .uri(&h2_uri);
+
+        // Remove Host header for H2 — :authority pseudo-header (from URI) is sufficient
+        // Having both Host and :authority causes nginx to return 400
+        let mut upstream_headers = upstream_headers;
+        upstream_headers.remove("host");
 
         if let Some(headers) = upstream_req.headers_mut() {
             *headers = upstream_headers;
@@ -1292,7 +1535,7 @@ impl HttpProxyService {
             route_id.map(|s| s.to_string()),
             Some(source_ip.to_string()),
             Direction::In,
-        );
+        ).with_connection_activity(Arc::clone(&conn_activity.last_activity), conn_activity.start);
         let boxed_body: BoxBody<Bytes, hyper::Error> = BoxBody::new(counting_req_body);
 
         let upstream_req = upstream_req.body(boxed_body).unwrap();
@@ -1300,16 +1543,20 @@ impl HttpProxyService {
         let upstream_response = match sender.send_request(upstream_req).await {
             Ok(resp) => resp,
             Err(e) => {
-                error!("HTTP/2 upstream request failed: {}", e);
                 // Evict the dead sender so subsequent requests get fresh connections
                 if let Some(key) = pool_key {
+                    let bk = format!("{}:{}", key.host, key.port);
+                    error!(backend = %bk, domain = %domain, error = %e, "Backend H2 request failed");
+                    self.metrics.backend_request_error(&bk);
                     self.connection_pool.remove_h2(key);
+                } else {
+                    error!(domain = %domain, error = %e, "Backend H2 request failed");
                 }
                 return Ok(error_response(StatusCode::BAD_GATEWAY, "Backend H2 request failed"));
             }
         };
 
-        self.build_streaming_response(upstream_response, route, route_id, source_ip).await
+        self.build_streaming_response(upstream_response, route, route_id, source_ip, conn_activity).await
     }
 
     /// Build the client-facing response from an upstream response, streaming the body.
@@ -1322,6 +1569,7 @@ impl HttpProxyService {
         route: &rustproxy_config::RouteConfig,
         route_id: Option<&str>,
         source_ip: &str,
+        conn_activity: &ConnActivity,
     ) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {
         let (resp_parts, resp_body) = upstream_response.into_parts();
 
@@ -1342,14 +1590,14 @@ impl HttpProxyService {
             route_id.map(|s| s.to_string()),
             Some(source_ip.to_string()),
             Direction::Out,
-        );
+        ).with_connection_activity(Arc::clone(&conn_activity.last_activity), conn_activity.start);
 
         let body: BoxBody<Bytes, hyper::Error> = BoxBody::new(counting_body);
 
         Ok(response.body(body).unwrap())
     }
 
-    /// Handle a WebSocket upgrade request.
+    /// Handle a WebSocket upgrade request (H1 Upgrade or H2 Extended CONNECT per RFC 8441).
     async fn handle_websocket_upgrade(
         &self,
         req: Request<Incoming>,
@@ -1360,6 +1608,8 @@ impl HttpProxyService {
         upstream_key: &str,
         cancel: CancellationToken,
         source_ip: &str,
+        is_h2: bool,
+        conn_activity: Option<ConnActivity>,
     ) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {
         use tokio::io::{AsyncReadExt, AsyncWriteExt};
 
@@ -1445,9 +1695,11 @@ impl HttpProxyService {
 
         let (parts, _body) = req.into_parts();
 
+        // H2 Extended CONNECT uses method=CONNECT, but the H1.1 backend expects GET
+        let backend_method = if is_h2 { "GET" } else { parts.method.as_str() };
         let mut raw_request = format!(
             "{} {} HTTP/1.1\r\n",
-            parts.method, upstream_path
+            backend_method, upstream_path
         );
 
         // Copy all original headers (preserving the client's Host header).
@@ -1475,6 +1727,23 @@ impl HttpProxyService {
             }
         }
 
+        // H2 Extended CONNECT doesn't carry H1 WebSocket handshake headers;
+        // inject them so the H1.1 backend can complete the upgrade.
+        if is_h2 {
+            if !parts.headers.contains_key("upgrade") {
+                raw_request.push_str("upgrade: websocket\r\n");
+            }
+            if !parts.headers.contains_key("connection") {
+                raw_request.push_str("connection: Upgrade\r\n");
+            }
+            if !parts.headers.contains_key("sec-websocket-version") {
+                raw_request.push_str("sec-websocket-version: 13\r\n");
+            }
+            if !parts.headers.contains_key("sec-websocket-key") {
+                raw_request.push_str("sec-websocket-key: dGhlIHNhbXBsZSBub25jZQ==\r\n");
+            }
+        }
+
         // Add standard reverse-proxy headers (X-Forwarded-*)
         {
             let original_host = parts.headers.get("host")
@@ -1577,8 +1846,12 @@ impl HttpProxyService {
             ));
         }
 
-        let mut client_resp = Response::builder()
-            .status(StatusCode::SWITCHING_PROTOCOLS);
+        // H1: 101 Switching Protocols; H2: 200 OK (RFC 8441 — hyper requires 2xx for Extended CONNECT upgrade)
+        let mut client_resp = if is_h2 {
+            Response::builder().status(StatusCode::OK)
+        } else {
+            Response::builder().status(StatusCode::SWITCHING_PROTOCOLS)
+        };
 
         if let Some(resp_headers) = client_resp.headers_mut() {
             for line in response_str.lines().skip(1) {
@@ -1589,6 +1862,17 @@ impl HttpProxyService {
                 if let Some((name, value)) = line.split_once(':') {
                     let name = name.trim();
                     let value = value.trim();
+                    // Skip hop-by-hop headers for H2 (forbidden by RFC 9113 §8.2.2)
+                    if is_h2 {
+                        let name_lower = name.to_lowercase();
+                        if name_lower == "upgrade" || name_lower == "connection"
+                            || name_lower == "sec-websocket-accept"
+                            || name_lower == "transfer-encoding"
+                            || name_lower == "keep-alive"
+                        {
+                            continue;
+                        }
+                    }
                     if let Ok(header_name) = hyper::header::HeaderName::from_bytes(name.as_bytes()) {
                         if let Ok(header_value) = hyper::header::HeaderValue::from_str(value) {
                             resp_headers.insert(header_name, header_value);
@@ -1629,6 +1913,11 @@ impl HttpProxyService {
             let last_activity = Arc::new(AtomicU64::new(0));
             let start = std::time::Instant::now();
 
+            // For H2 WebSocket: also update the connection-level activity tracker
+            // to prevent the idle watchdog from killing the H2 connection
+            let conn_act_c2u = conn_activity.as_ref().map(|ca| (Arc::clone(&ca.last_activity), ca.start));
+            let conn_act_u2c = conn_activity.as_ref().map(|ca| (Arc::clone(&ca.last_activity), ca.start));
+
             let la1 = Arc::clone(&last_activity);
             let c2u = tokio::spawn(async move {
                 let mut buf = vec![0u8; 65536];
@@ -1643,6 +1932,9 @@ impl HttpProxyService {
                     }
                     total += n as u64;
                     la1.store(start.elapsed().as_millis() as u64, Ordering::Relaxed);
+                    if let Some((ref ca, ca_start)) = conn_act_c2u {
+                        ca.store(ca_start.elapsed().as_millis() as u64, Ordering::Relaxed);
+                    }
                 }
                 let _ = uw.shutdown().await;
                 total
@@ -1662,6 +1954,9 @@ impl HttpProxyService {
                     }
                     total += n as u64;
                     la2.store(start.elapsed().as_millis() as u64, Ordering::Relaxed);
+                    if let Some((ref ca, ca_start)) = conn_act_u2c {
+                        ca.store(ca_start.elapsed().as_millis() as u64, Ordering::Relaxed);
+                    }
                 }
                 let _ = cw.shutdown().await;
                 total

rust/crates/rustproxy-metrics/src/collector.rs

@@ -3,6 +3,7 @@ use serde::{Deserialize, Serialize};
 use std::collections::HashSet;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Mutex;
+use std::time::Duration;
 
 use crate::throughput::{ThroughputSample, ThroughputTracker};
 
@@ -20,6 +21,7 @@ pub struct Metrics {
     pub throughput_recent_out_bytes_per_sec: u64,
     pub routes: std::collections::HashMap<String, RouteMetrics>,
     pub ips: std::collections::HashMap<String, IpMetrics>,
+    pub backends: std::collections::HashMap<String, BackendMetrics>,
     pub throughput_history: Vec<ThroughputSample>,
     pub total_http_requests: u64,
     pub http_requests_per_sec: u64,
@@ -52,6 +54,23 @@ pub struct IpMetrics {
     pub throughput_out_bytes_per_sec: u64,
 }
 
+/// Per-backend metrics (keyed by "host:port").
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct BackendMetrics {
+    pub active_connections: u64,
+    pub total_connections: u64,
+    pub protocol: String,
+    pub connect_errors: u64,
+    pub handshake_errors: u64,
+    pub request_errors: u64,
+    pub total_connect_time_us: u64,
+    pub connect_count: u64,
+    pub pool_hits: u64,
+    pub pool_misses: u64,
+    pub h2_failures: u64,
+}
+
 /// Statistics snapshot.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
@@ -69,6 +88,9 @@ const DEFAULT_RETENTION_SECONDS: usize = 3600;
 /// Maximum number of IPs to include in a snapshot (top by active connections).
 const MAX_IPS_IN_SNAPSHOT: usize = 100;
 
+/// Maximum number of backends to include in a snapshot (top by total connections).
+const MAX_BACKENDS_IN_SNAPSHOT: usize = 100;
+
 /// Metrics collector tracking connections and throughput.
 ///
 /// Design: The hot path (`record_bytes`) is entirely lock-free — it only touches
@@ -96,6 +118,19 @@ pub struct MetricsCollector {
     ip_pending_tp: DashMap<String, (AtomicU64, AtomicU64)>,
     ip_throughput: DashMap<String, Mutex<ThroughputTracker>>,
 
+    // ── Per-backend tracking (keyed by "host:port") ──
+    backend_active: DashMap<String, AtomicU64>,
+    backend_total: DashMap<String, AtomicU64>,
+    backend_protocol: DashMap<String, String>,
+    backend_connect_errors: DashMap<String, AtomicU64>,
+    backend_handshake_errors: DashMap<String, AtomicU64>,
+    backend_request_errors: DashMap<String, AtomicU64>,
+    backend_connect_time_us: DashMap<String, AtomicU64>,
+    backend_connect_count: DashMap<String, AtomicU64>,
+    backend_pool_hits: DashMap<String, AtomicU64>,
+    backend_pool_misses: DashMap<String, AtomicU64>,
+    backend_h2_failures: DashMap<String, AtomicU64>,
+
     // ── HTTP request tracking ──
     total_http_requests: AtomicU64,
     pending_http_requests: AtomicU64,
@@ -134,6 +169,17 @@ impl MetricsCollector {
             ip_bytes_out: DashMap::new(),
             ip_pending_tp: DashMap::new(),
             ip_throughput: DashMap::new(),
+            backend_active: DashMap::new(),
+            backend_total: DashMap::new(),
+            backend_protocol: DashMap::new(),
+            backend_connect_errors: DashMap::new(),
+            backend_handshake_errors: DashMap::new(),
+            backend_request_errors: DashMap::new(),
+            backend_connect_time_us: DashMap::new(),
+            backend_connect_count: DashMap::new(),
+            backend_pool_hits: DashMap::new(),
+            backend_pool_misses: DashMap::new(),
+            backend_h2_failures: DashMap::new(),
             total_http_requests: AtomicU64::new(0),
             pending_http_requests: AtomicU64::new(0),
             http_request_throughput: Mutex::new(ThroughputTracker::new(retention_seconds)),
@@ -268,6 +314,113 @@ impl MetricsCollector {
         self.pending_http_requests.fetch_add(1, Ordering::Relaxed);
     }
 
+    // ── Per-backend recording methods ──
+
+    /// Record a successful backend connection with its connect duration.
+    pub fn backend_connection_opened(&self, key: &str, connect_time: Duration) {
+        self.backend_active
+            .entry(key.to_string())
+            .or_insert_with(|| AtomicU64::new(0))
+            .fetch_add(1, Ordering::Relaxed);
+        self.backend_total
+            .entry(key.to_string())
+            .or_insert_with(|| AtomicU64::new(0))
+            .fetch_add(1, Ordering::Relaxed);
+        self.backend_connect_time_us
+            .entry(key.to_string())
+            .or_insert_with(|| AtomicU64::new(0))
+            .fetch_add(connect_time.as_micros() as u64, Ordering::Relaxed);
+        self.backend_connect_count
+            .entry(key.to_string())
+            .or_insert_with(|| AtomicU64::new(0))
+            .fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// Record a backend connection closing.
+    pub fn backend_connection_closed(&self, key: &str) {
+        if let Some(counter) = self.backend_active.get(key) {
+            let val = counter.load(Ordering::Relaxed);
+            if val > 0 {
+                counter.fetch_sub(1, Ordering::Relaxed);
+            }
+        }
+    }
+
+    /// Record a backend connect error (TCP or TLS connect failure/timeout).
+    pub fn backend_connect_error(&self, key: &str) {
+        self.backend_connect_errors
+            .entry(key.to_string())
+            .or_insert_with(|| AtomicU64::new(0))
+            .fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// Record a backend handshake error (H1 or H2 handshake failure).
+    pub fn backend_handshake_error(&self, key: &str) {
+        self.backend_handshake_errors
+            .entry(key.to_string())
+            .or_insert_with(|| AtomicU64::new(0))
+            .fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// Record a backend request error (send_request failure).
+    pub fn backend_request_error(&self, key: &str) {
+        self.backend_request_errors
+            .entry(key.to_string())
+            .or_insert_with(|| AtomicU64::new(0))
+            .fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// Record a connection pool hit for a backend.
+    pub fn backend_pool_hit(&self, key: &str) {
+        self.backend_pool_hits
+            .entry(key.to_string())
+            .or_insert_with(|| AtomicU64::new(0))
+            .fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// Record a connection pool miss for a backend.
+    pub fn backend_pool_miss(&self, key: &str) {
+        self.backend_pool_misses
+            .entry(key.to_string())
+            .or_insert_with(|| AtomicU64::new(0))
+            .fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// Record an H2 failure (h2 attempted but fell back to h1).
+    pub fn backend_h2_failure(&self, key: &str) {
+        self.backend_h2_failures
+            .entry(key.to_string())
+            .or_insert_with(|| AtomicU64::new(0))
+            .fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// Set the protocol in use for a backend ("h1" or "h2").
+    pub fn set_backend_protocol(&self, key: &str, protocol: &str) {
+        self.backend_protocol
+            .entry(key.to_string())
+            .and_modify(|v| {
+                if v != protocol {
+                    *v = protocol.to_string();
+                }
+            })
+            .or_insert_with(|| protocol.to_string());
+    }
+
+    /// Remove per-backend metrics for backends no longer in any route target.
+    pub fn retain_backends(&self, active_backends: &HashSet<String>) {
+        self.backend_active.retain(|k, _| active_backends.contains(k));
+        self.backend_total.retain(|k, _| active_backends.contains(k));
+        self.backend_protocol.retain(|k, _| active_backends.contains(k));
+        self.backend_connect_errors.retain(|k, _| active_backends.contains(k));
+        self.backend_handshake_errors.retain(|k, _| active_backends.contains(k));
+        self.backend_request_errors.retain(|k, _| active_backends.contains(k));
+        self.backend_connect_time_us.retain(|k, _| active_backends.contains(k));
+        self.backend_connect_count.retain(|k, _| active_backends.contains(k));
+        self.backend_pool_hits.retain(|k, _| active_backends.contains(k));
+        self.backend_pool_misses.retain(|k, _| active_backends.contains(k));
+        self.backend_h2_failures.retain(|k, _| active_backends.contains(k));
+    }
+
     /// Take a throughput sample on all trackers (cold path, call at 1Hz or configured interval).
     ///
     /// Drains the lock-free pending counters and feeds the accumulated bytes
@@ -488,6 +641,72 @@ impl MetricsCollector {
             });
         }
 
+        // Collect per-backend metrics, capped at top MAX_BACKENDS_IN_SNAPSHOT by total connections
+        let mut backend_entries: Vec<(String, BackendMetrics)> = Vec::new();
+        for entry in self.backend_total.iter() {
+            let key = entry.key().clone();
+            let total = entry.value().load(Ordering::Relaxed);
+            let active = self.backend_active
+                .get(&key)
+                .map(|c| c.load(Ordering::Relaxed))
+                .unwrap_or(0);
+            let protocol = self.backend_protocol
+                .get(&key)
+                .map(|v| v.value().clone())
+                .unwrap_or_else(|| "unknown".to_string());
+            let connect_errors = self.backend_connect_errors
+                .get(&key)
+                .map(|c| c.load(Ordering::Relaxed))
+                .unwrap_or(0);
+            let handshake_errors = self.backend_handshake_errors
+                .get(&key)
+                .map(|c| c.load(Ordering::Relaxed))
+                .unwrap_or(0);
+            let request_errors = self.backend_request_errors
+                .get(&key)
+                .map(|c| c.load(Ordering::Relaxed))
+                .unwrap_or(0);
+            let total_connect_time_us = self.backend_connect_time_us
+                .get(&key)
+                .map(|c| c.load(Ordering::Relaxed))
+                .unwrap_or(0);
+            let connect_count = self.backend_connect_count
+                .get(&key)
+                .map(|c| c.load(Ordering::Relaxed))
+                .unwrap_or(0);
+            let pool_hits = self.backend_pool_hits
+                .get(&key)
+                .map(|c| c.load(Ordering::Relaxed))
+                .unwrap_or(0);
+            let pool_misses = self.backend_pool_misses
+                .get(&key)
+                .map(|c| c.load(Ordering::Relaxed))
+                .unwrap_or(0);
+            let h2_failures = self.backend_h2_failures
+                .get(&key)
+                .map(|c| c.load(Ordering::Relaxed))
+                .unwrap_or(0);
+
+            backend_entries.push((key, BackendMetrics {
+                active_connections: active,
+                total_connections: total,
+                protocol,
+                connect_errors,
+                handshake_errors,
+                request_errors,
+                total_connect_time_us,
+                connect_count,
+                pool_hits,
+                pool_misses,
+                h2_failures,
+            }));
+        }
+        // Sort by total connections descending, then cap
+        backend_entries.sort_by(|a, b| b.1.total_connections.cmp(&a.1.total_connections));
+        backend_entries.truncate(MAX_BACKENDS_IN_SNAPSHOT);
+
+        let backends: std::collections::HashMap<String, BackendMetrics> = backend_entries.into_iter().collect();
+
         // HTTP request rates
         let (http_rps, http_rps_recent) = self.http_request_throughput
             .lock()
@@ -509,6 +728,7 @@ impl MetricsCollector {
             throughput_recent_out_bytes_per_sec: global_recent_out,
             routes,
             ips,
+            backends,
             throughput_history,
             total_http_requests: self.total_http_requests.load(Ordering::Relaxed),
             http_requests_per_sec: http_rps,
@@ -805,4 +1025,120 @@ mod tests {
         assert_eq!(snapshot.throughput_history[0].bytes_in, 100);
         assert_eq!(snapshot.throughput_history[4].bytes_in, 500);
     }
+
+    #[test]
+    fn test_backend_metrics_basic() {
+        let collector = MetricsCollector::new();
+        let key = "backend1:8080";
+
+        // Open connections with timing
+        collector.backend_connection_opened(key, Duration::from_millis(15));
+        collector.backend_connection_opened(key, Duration::from_millis(25));
+
+        assert_eq!(collector.backend_active.get(key).unwrap().load(Ordering::Relaxed), 2);
+        assert_eq!(collector.backend_total.get(key).unwrap().load(Ordering::Relaxed), 2);
+        assert_eq!(collector.backend_connect_count.get(key).unwrap().load(Ordering::Relaxed), 2);
+        // 15ms + 25ms = 40ms = 40_000us
+        assert_eq!(collector.backend_connect_time_us.get(key).unwrap().load(Ordering::Relaxed), 40_000);
+
+        // Close one
+        collector.backend_connection_closed(key);
+        assert_eq!(collector.backend_active.get(key).unwrap().load(Ordering::Relaxed), 1);
+        // total stays
+        assert_eq!(collector.backend_total.get(key).unwrap().load(Ordering::Relaxed), 2);
+
+        // Record errors
+        collector.backend_connect_error(key);
+        collector.backend_handshake_error(key);
+        collector.backend_request_error(key);
+        collector.backend_h2_failure(key);
+        collector.backend_pool_hit(key);
+        collector.backend_pool_hit(key);
+        collector.backend_pool_miss(key);
+
+        assert_eq!(collector.backend_connect_errors.get(key).unwrap().load(Ordering::Relaxed), 1);
+        assert_eq!(collector.backend_handshake_errors.get(key).unwrap().load(Ordering::Relaxed), 1);
+        assert_eq!(collector.backend_request_errors.get(key).unwrap().load(Ordering::Relaxed), 1);
+        assert_eq!(collector.backend_h2_failures.get(key).unwrap().load(Ordering::Relaxed), 1);
+        assert_eq!(collector.backend_pool_hits.get(key).unwrap().load(Ordering::Relaxed), 2);
+        assert_eq!(collector.backend_pool_misses.get(key).unwrap().load(Ordering::Relaxed), 1);
+
+        // Protocol
+        collector.set_backend_protocol(key, "h1");
+        assert_eq!(collector.backend_protocol.get(key).unwrap().value(), "h1");
+        collector.set_backend_protocol(key, "h2");
+        assert_eq!(collector.backend_protocol.get(key).unwrap().value(), "h2");
+    }
+
+    #[test]
+    fn test_backend_metrics_in_snapshot() {
+        let collector = MetricsCollector::new();
+
+        collector.backend_connection_opened("b1:443", Duration::from_millis(10));
+        collector.backend_connection_opened("b2:8080", Duration::from_millis(20));
+        collector.set_backend_protocol("b1:443", "h2");
+        collector.set_backend_protocol("b2:8080", "h1");
+        collector.backend_connect_error("b1:443");
+
+        let snapshot = collector.snapshot();
+        assert_eq!(snapshot.backends.len(), 2);
+
+        let b1 = snapshot.backends.get("b1:443").unwrap();
+        assert_eq!(b1.active_connections, 1);
+        assert_eq!(b1.total_connections, 1);
+        assert_eq!(b1.protocol, "h2");
+        assert_eq!(b1.connect_errors, 1);
+        assert_eq!(b1.total_connect_time_us, 10_000);
+        assert_eq!(b1.connect_count, 1);
+
+        let b2 = snapshot.backends.get("b2:8080").unwrap();
+        assert_eq!(b2.protocol, "h1");
+        assert_eq!(b2.connect_errors, 0);
+    }
+
+    #[test]
+    fn test_retain_backends_prunes_stale() {
+        let collector = MetricsCollector::new();
+
+        collector.backend_connection_opened("active:443", Duration::from_millis(5));
+        collector.backend_connection_opened("stale:8080", Duration::from_millis(10));
+        collector.set_backend_protocol("active:443", "h1");
+        collector.set_backend_protocol("stale:8080", "h2");
+        collector.backend_connect_error("stale:8080");
+
+        let active = HashSet::from(["active:443".to_string()]);
+        collector.retain_backends(&active);
+
+        // active:443 should still exist
+        assert!(collector.backend_total.get("active:443").is_some());
+        assert!(collector.backend_protocol.get("active:443").is_some());
+
+        // stale:8080 should be fully removed
+        assert!(collector.backend_active.get("stale:8080").is_none());
+        assert!(collector.backend_total.get("stale:8080").is_none());
+        assert!(collector.backend_protocol.get("stale:8080").is_none());
+        assert!(collector.backend_connect_errors.get("stale:8080").is_none());
+        assert!(collector.backend_connect_time_us.get("stale:8080").is_none());
+        assert!(collector.backend_connect_count.get("stale:8080").is_none());
+        assert!(collector.backend_pool_hits.get("stale:8080").is_none());
+        assert!(collector.backend_pool_misses.get("stale:8080").is_none());
+        assert!(collector.backend_h2_failures.get("stale:8080").is_none());
+    }
+
+    #[test]
+    fn test_backend_connection_closed_saturates() {
+        let collector = MetricsCollector::new();
+        let key = "b:80";
+
+        // Close without opening — should not underflow
+        collector.backend_connection_closed(key);
+        // No entry created
+        assert!(collector.backend_active.get(key).is_none());
+
+        // Open one, close two — should saturate at 0
+        collector.backend_connection_opened(key, Duration::from_millis(1));
+        collector.backend_connection_closed(key);
+        collector.backend_connection_closed(key);
+        assert_eq!(collector.backend_active.get(key).unwrap().load(Ordering::Relaxed), 0);
+    }
 }

rust/crates/rustproxy/src/lib.rs

@@ -603,6 +603,31 @@ impl RustProxy {
             .collect();
         self.metrics.retain_routes(&active_route_ids);
 
+        // Prune per-backend metrics for backends no longer in any route target.
+        // For PortSpec::Preserve routes, expand across all listening ports since
+        // the actual runtime port depends on the incoming connection.
+        let listening_ports = self.get_listening_ports();
+        let active_backends: HashSet<String> = routes.iter()
+            .filter_map(|r| r.action.targets.as_ref())
+            .flat_map(|targets| targets.iter())
+            .flat_map(|target| {
+                let hosts: Vec<String> = target.host.to_vec().into_iter().map(|s| s.to_string()).collect();
+                match &target.port {
+                    rustproxy_config::PortSpec::Fixed(p) => {
+                        hosts.into_iter().map(|h| format!("{}:{}", h, p)).collect::<Vec<_>>()
+                    }
+                    _ => {
+                        // Preserve/special: expand across all listening ports
+                        let lp = &listening_ports;
+                        hosts.into_iter()
+                            .flat_map(|h| lp.iter().map(move |p| format!("{}:{}", h, *p)))
+                            .collect::<Vec<_>>()
+                    }
+                }
+            })
+            .collect();
+        self.metrics.retain_backends(&active_backends);
+
         // Atomically swap the route table
         let new_manager = Arc::new(new_manager);
         self.route_table.store(Arc::clone(&new_manager));

ts/00_commitinfo_data.ts

@@ -3,6 +3,6 @@
  */
 export const commitinfo = {
   name: '@push.rocks/smartproxy',
-  version: '25.9.3',
+  version: '25.11.2',
   description: 'A powerful proxy package with unified route-based configuration for high traffic management. Features include SSL/TLS support, flexible routing patterns, WebSocket handling, advanced security options, and automatic ACME certificate management.'
 }

ts/proxies/smart-proxy/models/metrics-types.ts

@@ -67,6 +67,13 @@ export interface IMetrics {
     connections(): number;
   };
   
+  // Backend metrics
+  backends: {
+    byBackend(): Map<string, IBackendMetrics>;
+    protocols(): Map<string, string>;
+    topByErrors(limit?: number): Array<{ backend: string; errors: number }>;
+  };
+
   // Performance metrics
   percentiles: {
     connectionDuration(): { p50: number; p95: number; p99: number };
@@ -98,6 +105,21 @@ export interface IMetricsConfig {
   prometheusPrefix: string;        // Default: smartproxy_
 }
 
+/**
+ * Per-backend metrics
+ */
+export interface IBackendMetrics {
+  protocol: string;
+  activeConnections: number;
+  totalConnections: number;
+  connectErrors: number;
+  handshakeErrors: number;
+  requestErrors: number;
+  avgConnectTimeMs: number;
+  poolHitRate: number;
+  h2Failures: number;
+}
+
 /**
  * Internal interface for connection byte tracking
  */

ts/proxies/smart-proxy/rust-metrics-adapter.ts

@@ -1,4 +1,4 @@
-import type { IMetrics, IThroughputData, IThroughputHistoryPoint } from './models/metrics-types.js';
+import type { IMetrics, IBackendMetrics, IThroughputData, IThroughputHistoryPoint } from './models/metrics-types.js';
 import type { RustProxyBridge } from './rust-proxy-bridge.js';
 
 /**
@@ -169,6 +169,55 @@ export class RustMetricsAdapter implements IMetrics {
     },
   };
 
+  public backends = {
+    byBackend: (): Map<string, IBackendMetrics> => {
+      const result = new Map<string, IBackendMetrics>();
+      if (this.cache?.backends) {
+        for (const [key, bm] of Object.entries(this.cache.backends)) {
+          const m = bm as any;
+          const totalTimeUs = m.totalConnectTimeUs ?? 0;
+          const count = m.connectCount ?? 0;
+          const poolHits = m.poolHits ?? 0;
+          const poolMisses = m.poolMisses ?? 0;
+          const poolTotal = poolHits + poolMisses;
+          result.set(key, {
+            protocol: m.protocol ?? 'unknown',
+            activeConnections: m.activeConnections ?? 0,
+            totalConnections: m.totalConnections ?? 0,
+            connectErrors: m.connectErrors ?? 0,
+            handshakeErrors: m.handshakeErrors ?? 0,
+            requestErrors: m.requestErrors ?? 0,
+            avgConnectTimeMs: count > 0 ? (totalTimeUs / count) / 1000 : 0,
+            poolHitRate: poolTotal > 0 ? poolHits / poolTotal : 0,
+            h2Failures: m.h2Failures ?? 0,
+          });
+        }
+      }
+      return result;
+    },
+    protocols: (): Map<string, string> => {
+      const result = new Map<string, string>();
+      if (this.cache?.backends) {
+        for (const [key, bm] of Object.entries(this.cache.backends)) {
+          result.set(key, (bm as any).protocol ?? 'unknown');
+        }
+      }
+      return result;
+    },
+    topByErrors: (limit: number = 10): Array<{ backend: string; errors: number }> => {
+      const result: Array<{ backend: string; errors: number }> = [];
+      if (this.cache?.backends) {
+        for (const [key, bm] of Object.entries(this.cache.backends)) {
+          const m = bm as any;
+          const errors = (m.connectErrors ?? 0) + (m.handshakeErrors ?? 0) + (m.requestErrors ?? 0);
+          if (errors > 0) result.push({ backend: key, errors });
+        }
+      }
+      result.sort((a, b) => b.errors - a.errors);
+      return result.slice(0, limit);
+    },
+  };
+
   public percentiles = {
     connectionDuration: (): { p50: number; p95: number; p99: number } => {
       return { p50: 0, p95: 0, p99: 0 };