Explicit Reasons of Session Fallback Cleanup + ME Close

- bound API and metrics connection handling
- default metrics listener to localhost
- reject untrusted PROXY protocol peers before parsing headers
- cap API request body size and PROXY v2 payload allocation
- validate route usernames and TLS domains consistently

AGENTS.md

@@ -191,6 +191,11 @@ When facing a non-trivial modification, follow this sequence:
 4. **Implement**: Make the minimal, isolated change.
 5. **Verify**: Explain why the change preserves existing behavior and architectural integrity.
 
+When the repository contains a `PLAN.md` for the current task, maintain it as
+a working checkbox plan while implementing changes. Mark completed and partial
+items in `PLAN.md` as the code changes land, so the remaining work stays
+explicit and future passes do not waste time rediscovering status.
+
 ---
 
 ### 9. Context Awareness
@@ -222,10 +227,9 @@ Your response MUST consist of two sections:
 
 **Section 2: `## Changes`**
 
-- For each modified or created file: the filename on a separate line in backticks, followed by the code block.
-- For files **under 200 lines**: return the full file with all changes applied.
-- For files **over 200 lines**: return only the changed functions/blocks with at least 3 lines of surrounding context above and below. If the user requests the full file, provide it.
-- New files: full file content.
+- For each modified or created file: the filename on a separate line in backticks, followed by a concise description of what changed.
+- Do not include full file contents or long code blocks in `## Changes` unless the user explicitly asks for code text.
+- If code snippets are necessary, include only the minimal relevant excerpt.
 - End with a suggested git commit message in English.
 
 #### Reporting Out-of-Scope Issues
@@ -429,4 +433,3 @@ Every patch must be **atomic and production-safe**.
 * **No transitional states** — no placeholders, incomplete refactors, or temporary inconsistencies.
 
 **Invariant:** After any single patch, the repository remains fully functional and buildable.
-

src/api/mod.rs

@@ -5,6 +5,7 @@ use std::net::{IpAddr, SocketAddr};
 use std::path::PathBuf;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::time::Duration;
 
 use http_body_util::Full;
 use hyper::body::{Bytes, Incoming};
@@ -12,8 +13,10 @@ use hyper::header::AUTHORIZATION;
 use hyper::server::conn::http1;
 use hyper::service::service_fn;
 use hyper::{Method, Request, Response, StatusCode};
+use subtle::ConstantTimeEq;
 use tokio::net::TcpListener;
-use tokio::sync::{Mutex, RwLock, watch};
+use tokio::sync::{Mutex, RwLock, Semaphore, watch};
+use tokio::time::timeout;
 use tracing::{debug, info, warn};
 
 use crate::config::{ApiGrayAction, ProxyConfig};
@@ -44,6 +47,7 @@ use http_utils::{error_response, read_json, read_optional_json, success_response
 use model::{
     ApiFailure, ClassCount, CreateUserRequest, DeleteUserResponse, HealthData, HealthReadyData,
     PatchUserRequest, ResetUserQuotaResponse, RotateSecretRequest, SummaryData, UserActiveIps,
+    is_valid_username,
 };
 use runtime_edge::{
     EdgeConnectionsCacheEntry, build_runtime_connections_summary_data,
@@ -66,6 +70,10 @@ use runtime_zero::{
 };
 use users::{create_user, delete_user, patch_user, rotate_secret, users_from_config};
 
+const API_MAX_CONTROL_CONNECTIONS: usize = 1024;
+const API_HTTP_CONNECTION_TIMEOUT: Duration = Duration::from_secs(15);
+const ROUTE_USERNAME_ERROR: &str = "username must match [A-Za-z0-9_.-] and be 1..64 chars";
+
 pub(super) struct ApiRuntimeState {
     pub(super) process_started_at_epoch_secs: u64,
     pub(super) config_reload_count: AtomicU64,
@@ -103,6 +111,18 @@ impl ApiShared {
     }
 }
 
+fn auth_header_matches(actual: &str, expected: &str) -> bool {
+    actual.as_bytes().ct_eq(expected.as_bytes()).into()
+}
+
+fn parse_route_username(user: &str) -> Result<&str, ApiFailure> {
+    if is_valid_username(user) {
+        Ok(user)
+    } else {
+        Err(ApiFailure::bad_request(ROUTE_USERNAME_ERROR))
+    }
+}
+
 pub async fn serve(
     listen: SocketAddr,
     stats: Arc<Stats>,
@@ -167,6 +187,8 @@ pub async fn serve(
         shared.runtime_events.clone(),
     );
 
+    let connection_permits = Arc::new(Semaphore::new(API_MAX_CONTROL_CONNECTIONS));
+
     loop {
         let (stream, peer) = match listener.accept().await {
             Ok(v) => v,
@@ -176,20 +198,45 @@ pub async fn serve(
             }
         };
 
+        let connection_permit = match connection_permits.clone().try_acquire_owned() {
+            Ok(permit) => permit,
+            Err(_) => {
+                debug!(
+                    peer = %peer,
+                    max_connections = API_MAX_CONTROL_CONNECTIONS,
+                    "Dropping API connection: control-plane connection budget exhausted"
+                );
+                continue;
+            }
+        };
+
         let shared_conn = shared.clone();
         let config_rx_conn = config_rx.clone();
         tokio::spawn(async move {
+            let _connection_permit = connection_permit;
             let svc = service_fn(move |req: Request<Incoming>| {
                 let shared_req = shared_conn.clone();
                 let config_rx_req = config_rx_conn.clone();
                 async move { handle(req, peer, shared_req, config_rx_req).await }
             });
-            if let Err(error) = http1::Builder::new()
-                .serve_connection(hyper_util::rt::TokioIo::new(stream), svc)
-                .await
+            match timeout(
+                API_HTTP_CONNECTION_TIMEOUT,
+                http1::Builder::new().serve_connection(hyper_util::rt::TokioIo::new(stream), svc),
+            )
+            .await
             {
-                if !error.is_user() {
-                    debug!(error = %error, "API connection error");
+                Ok(Ok(())) => {}
+                Ok(Err(error)) => {
+                    if !error.is_user() {
+                        debug!(error = %error, "API connection error");
+                    }
+                }
+                Err(_) => {
+                    debug!(
+                        peer = %peer,
+                        timeout_ms = API_HTTP_CONNECTION_TIMEOUT.as_millis() as u64,
+                        "API connection timed out"
+                    );
                 }
             }
         });
@@ -245,7 +292,7 @@ async fn handle(
             .headers()
             .get(AUTHORIZATION)
             .and_then(|v| v.to_str().ok())
-            .map(|v| v == api_cfg.auth_header)
+            .map(|v| auth_header_matches(v, &api_cfg.auth_header))
             .unwrap_or(false);
         if !auth_ok {
             return Ok(error_response(
@@ -501,6 +548,7 @@ async fn handle(
                     && !user.is_empty()
                     && !user.contains('/')
                 {
+                    let user = parse_route_username(user)?;
                     if api_cfg.read_only {
                         return Ok(error_response(
                             request_id,
@@ -544,10 +592,64 @@ async fn handle(
                         revision,
                     ));
                 }
+                if method == Method::POST
+                    && let Some(base_user) = normalized_path
+                        .strip_prefix("/v1/users/")
+                        .and_then(|path| path.strip_suffix("/rotate-secret"))
+                    && !base_user.is_empty()
+                    && !base_user.contains('/')
+                {
+                    let base_user = parse_route_username(base_user)?;
+                    if api_cfg.read_only {
+                        return Ok(error_response(
+                            request_id,
+                            ApiFailure::new(
+                                StatusCode::FORBIDDEN,
+                                "read_only",
+                                "API runs in read-only mode",
+                            ),
+                        ));
+                    }
+                    let expected_revision = parse_if_match(req.headers());
+                    let body =
+                        read_optional_json::<RotateSecretRequest>(req.into_body(), body_limit)
+                            .await?;
+                    let result = rotate_secret(
+                        base_user,
+                        body.unwrap_or_default(),
+                        expected_revision,
+                        &shared,
+                    )
+                    .await;
+                    let (mut data, revision) = match result {
+                        Ok(ok) => ok,
+                        Err(error) => {
+                            shared.runtime_events.record(
+                                "api.user.rotate_secret.failed",
+                                format!("username={} code={}", base_user, error.code),
+                            );
+                            return Err(error);
+                        }
+                    };
+                    let runtime_cfg = config_rx.borrow().clone();
+                    data.user.in_runtime =
+                        runtime_cfg.access.users.contains_key(&data.user.username);
+                    shared.runtime_events.record(
+                        "api.user.rotate_secret.ok",
+                        format!("username={}", base_user),
+                    );
+                    let status = if data.user.in_runtime {
+                        StatusCode::OK
+                    } else {
+                        StatusCode::ACCEPTED
+                    };
+                    return Ok(success_response(status, data, revision));
+                }
                 if let Some(user) = normalized_path.strip_prefix("/v1/users/")
                     && !user.is_empty()
                     && !user.contains('/')
                 {
+                    let user = parse_route_username(user)?;
                     if method == Method::GET {
                         let revision = current_revision(&shared.config_path).await?;
                         let disk_cfg = load_config_from_disk(&shared.config_path).await?;
@@ -648,56 +750,6 @@ async fn handle(
                         };
                         return Ok(success_response(status, response, revision));
                     }
-                    if method == Method::POST
-                        && let Some(base_user) = user.strip_suffix("/rotate-secret")
-                        && !base_user.is_empty()
-                        && !base_user.contains('/')
-                    {
-                        if api_cfg.read_only {
-                            return Ok(error_response(
-                                request_id,
-                                ApiFailure::new(
-                                    StatusCode::FORBIDDEN,
-                                    "read_only",
-                                    "API runs in read-only mode",
-                                ),
-                            ));
-                        }
-                        let expected_revision = parse_if_match(req.headers());
-                        let body =
-                            read_optional_json::<RotateSecretRequest>(req.into_body(), body_limit)
-                                .await?;
-                        let result = rotate_secret(
-                            base_user,
-                            body.unwrap_or_default(),
-                            expected_revision,
-                            &shared,
-                        )
-                        .await;
-                        let (mut data, revision) = match result {
-                            Ok(ok) => ok,
-                            Err(error) => {
-                                shared.runtime_events.record(
-                                    "api.user.rotate_secret.failed",
-                                    format!("username={} code={}", base_user, error.code),
-                                );
-                                return Err(error);
-                            }
-                        };
-                        let runtime_cfg = config_rx.borrow().clone();
-                        data.user.in_runtime =
-                            runtime_cfg.access.users.contains_key(&data.user.username);
-                        shared.runtime_events.record(
-                            "api.user.rotate_secret.ok",
-                            format!("username={}", base_user),
-                        );
-                        let status = if data.user.in_runtime {
-                            StatusCode::OK
-                        } else {
-                            StatusCode::ACCEPTED
-                        };
-                        return Ok(success_response(status, data, revision));
-                    }
                     if method == Method::POST {
                         return Ok(error_response(
                             request_id,

src/api/users.rs

@@ -465,12 +465,7 @@ pub(super) async fn users_from_config(
             .map(|secret| {
                 build_user_links(cfg, secret, startup_detected_ip_v4, startup_detected_ip_v6)
             })
-            .unwrap_or(UserLinks {
-                classic: Vec::new(),
-                secure: Vec::new(),
-                tls: Vec::new(),
-                tls_domains: Vec::new(),
-            });
+            .unwrap_or_else(empty_user_links);
         users.push(UserInfo {
             in_runtime: runtime_cfg
                 .map(|runtime| runtime.access.users.contains_key(&username))
@@ -511,6 +506,15 @@ pub(super) async fn users_from_config(
     users
 }
 
+fn empty_user_links() -> UserLinks {
+    UserLinks {
+        classic: Vec::new(),
+        secure: Vec::new(),
+        tls: Vec::new(),
+        tls_domains: Vec::new(),
+    }
+}
+
 fn build_user_links(
     cfg: &ProxyConfig,
     secret: &str,

src/config/load.rs

@@ -22,6 +22,14 @@ const MAX_ME_ROUTE_CHANNEL_CAPACITY: usize = 8_192;
 const MAX_ME_C2ME_CHANNEL_CAPACITY: usize = 8_192;
 const MIN_MAX_CLIENT_FRAME_BYTES: usize = 4 * 1024;
 const MAX_MAX_CLIENT_FRAME_BYTES: usize = 16 * 1024 * 1024;
+const MAX_API_REQUEST_BODY_LIMIT_BYTES: usize = 1024 * 1024;
+
+fn is_valid_tls_domain_name(domain: &str) -> bool {
+    !domain.is_empty()
+        && !domain
+            .chars()
+            .any(|ch| ch.is_whitespace() || matches!(ch, '/' | '\\'))
+}
 
 const TOP_LEVEL_CONFIG_KEYS: &[&str] = &[
     "general",
@@ -1773,9 +1781,11 @@ impl ProxyConfig {
             ));
         }
 
-        if config.server.api.request_body_limit_bytes == 0 {
+        if !(1..=MAX_API_REQUEST_BODY_LIMIT_BYTES)
+            .contains(&config.server.api.request_body_limit_bytes)
+        {
             return Err(ProxyError::Config(
-                "server.api.request_body_limit_bytes must be > 0".to_string(),
+                "server.api.request_body_limit_bytes must be within [1, 1048576]".to_string(),
             ));
         }
 
@@ -2103,13 +2113,22 @@ impl ProxyConfig {
             return Err(ProxyError::Config("No modes enabled".to_string()));
         }
 
-        if self.censorship.tls_domain.contains(' ') || self.censorship.tls_domain.contains('/') {
+        if !is_valid_tls_domain_name(&self.censorship.tls_domain) {
             return Err(ProxyError::Config(format!(
                 "Invalid tls_domain: '{}'. Must be a valid domain name",
                 self.censorship.tls_domain
             )));
         }
 
+        for domain in &self.censorship.tls_domains {
+            if !is_valid_tls_domain_name(domain) {
+                return Err(ProxyError::Config(format!(
+                    "Invalid tls_domains entry: '{}'. Must be a valid domain name",
+                    domain
+                )));
+            }
+        }
+
         for (user, tag) in &self.access.user_ad_tags {
             let zeros = "00000000000000000000000000000000";
             if !is_valid_ad_tag(tag) {

src/config/types.rs

@@ -543,10 +543,17 @@ pub struct GeneralConfig {
     pub me_d2c_frame_buf_shrink_threshold_bytes: usize,
 
     /// Copy buffer size for client->DC direction in direct relay.
+    ///
+    /// This is also the upper bound for one amortized upload rate-limit burst:
+    /// upload debt is settled before the next relay read instead of blocking
+    /// inside the completed read path.
     #[serde(default = "default_direct_relay_copy_buf_c2s_bytes")]
     pub direct_relay_copy_buf_c2s_bytes: usize,
 
     /// Copy buffer size for DC->client direction in direct relay.
+    ///
+    /// This bounds one direct download rate-limit grant because writes are
+    /// clipped to the currently available shaper budget.
     #[serde(default = "default_direct_relay_copy_buf_s2c_bytes")]
     pub direct_relay_copy_buf_s2c_bytes: usize,
 
@@ -1891,14 +1898,17 @@ pub struct AccessConfig {
     ///
     /// Each entry supports independent upload (`up_bps`) and download
     /// (`down_bps`) ceilings. A value of `0` in one direction means
-    /// "unlimited" for that direction.
+    /// "unlimited" for that direction. Limits are amortized: a relay quantum
+    /// may pass as a bounded burst, and the limiter applies the resulting wait
+    /// before later traffic in the same direction proceeds.
     #[serde(default)]
     pub user_rate_limits: HashMap<String, RateLimitBps>,
 
     /// Per-CIDR aggregate transport rate limits in bits-per-second.
     ///
     /// Matching uses longest-prefix-wins semantics. A value of `0` in one
-    /// direction means "unlimited" for that direction.
+    /// direction means "unlimited" for that direction. Limits are amortized
+    /// with the same bounded-burst contract as per-user rate limits.
     #[serde(default)]
     pub cidr_rate_limits: HashMap<IpNetwork, RateLimitBps>,
 

src/error.rs

@@ -222,6 +222,9 @@ pub enum ProxyError {
     #[error("Proxy error: {0}")]
     Proxy(String),
 
+    #[error("ME connection lost")]
+    MiddleConnectionLost,
+
     // ============= Config Errors =============
     #[error("Config error: {0}")]
     Config(String),

src/maestro/listeners.rs

@@ -494,7 +494,7 @@ pub(crate) fn spawn_tcp_accept_loops(
 
                                 let me_closed = matches!(
                                     &e,
-                                    crate::error::ProxyError::Proxy(msg) if msg == "ME connection lost"
+                                    crate::error::ProxyError::MiddleConnectionLost
                                 );
                                 let route_switched = matches!(
                                     &e,

src/maestro/runtime_tasks.rs

@@ -73,6 +73,11 @@ pub(crate) async fn spawn_runtime_tasks(
         rc_clone.run_periodic_cleanup().await;
     });
 
+    let stats_maintenance = stats.clone();
+    tokio::spawn(async move {
+        stats_maintenance.run_periodic_user_stats_maintenance().await;
+    });
+
     let detected_ip_v4: Option<IpAddr> = probe.detected_ipv4.map(IpAddr::V4);
     let detected_ip_v6: Option<IpAddr> = probe.detected_ipv6.map(IpAddr::V6);
     debug!(

src/metrics.rs

@@ -11,6 +11,8 @@ use hyper::service::service_fn;
 use hyper::{Request, Response, StatusCode};
 use ipnetwork::IpNetwork;
 use tokio::net::TcpListener;
+use tokio::sync::Semaphore;
+use tokio::time::timeout;
 use tracing::{debug, info, warn};
 
 use crate::config::ProxyConfig;
@@ -27,6 +29,8 @@ use crate::transport::{ListenOptions, create_listener};
 const USER_LABELED_METRICS_MAX_USERS: usize = 4096;
 // Keeps TLS-front per-domain health series bounded for large generated configs.
 const TLS_FRONT_PROFILE_HEALTH_MAX_DOMAINS: usize = 256;
+const METRICS_MAX_CONTROL_CONNECTIONS: usize = 512;
+const METRICS_HTTP_CONNECTION_TIMEOUT: Duration = Duration::from_secs(15);
 
 pub async fn serve(
     port: u16,
@@ -74,11 +78,11 @@ pub async fn serve(
         return;
     }
 
-    // Fallback: bind on 0.0.0.0 and [::] using metrics_port.
+    // Fallback: keep metrics local unless an explicit metrics_listen is configured.
     let mut listener_v4 = None;
     let mut listener_v6 = None;
 
-    let addr_v4 = SocketAddr::from(([0, 0, 0, 0], port));
+    let addr_v4 = SocketAddr::from(([127, 0, 0, 1], port));
     match bind_metrics_listener(addr_v4, false, listen_backlog) {
         Ok(listener) => {
             info!(
@@ -92,11 +96,11 @@ pub async fn serve(
         }
     }
 
-    let addr_v6 = SocketAddr::from(([0, 0, 0, 0, 0, 0, 0, 0], port));
+    let addr_v6 = SocketAddr::from(([0, 0, 0, 0, 0, 0, 0, 1], port));
     match bind_metrics_listener(addr_v6, true, listen_backlog) {
         Ok(listener) => {
             info!(
-                "Metrics endpoint: http://[::]:{}/metrics and /beobachten",
+                "Metrics endpoint: http://[::1]:{}/metrics and /beobachten",
                 port
             );
             listener_v6 = Some(listener);
@@ -184,6 +188,8 @@ async fn serve_listener(
     config_rx: tokio::sync::watch::Receiver<Arc<ProxyConfig>>,
     whitelist: Arc<Vec<IpNetwork>>,
 ) {
+    let connection_permits = Arc::new(Semaphore::new(METRICS_MAX_CONTROL_CONNECTIONS));
+
     loop {
         let (stream, peer) = match listener.accept().await {
             Ok(v) => v,
@@ -198,6 +204,18 @@ async fn serve_listener(
             continue;
         }
 
+        let connection_permit = match connection_permits.clone().try_acquire_owned() {
+            Ok(permit) => permit,
+            Err(_) => {
+                debug!(
+                    peer = %peer,
+                    max_connections = METRICS_MAX_CONTROL_CONNECTIONS,
+                    "Dropping metrics connection: control-plane connection budget exhausted"
+                );
+                continue;
+            }
+        };
+
         let stats = stats.clone();
         let beobachten = beobachten.clone();
         let shared_state = shared_state.clone();
@@ -205,6 +223,7 @@ async fn serve_listener(
         let tls_cache = tls_cache.clone();
         let config_rx_conn = config_rx.clone();
         tokio::spawn(async move {
+            let _connection_permit = connection_permit;
             let svc = service_fn(move |req| {
                 let stats = stats.clone();
                 let beobachten = beobachten.clone();
@@ -225,11 +244,23 @@ async fn serve_listener(
                     .await
                 }
             });
-            if let Err(e) = http1::Builder::new()
-                .serve_connection(hyper_util::rt::TokioIo::new(stream), svc)
-                .await
+            match timeout(
+                METRICS_HTTP_CONNECTION_TIMEOUT,
+                http1::Builder::new().serve_connection(hyper_util::rt::TokioIo::new(stream), svc),
+            )
+            .await
             {
-                debug!(error = %e, "Metrics connection error");
+                Ok(Ok(())) => {}
+                Ok(Err(e)) => {
+                    debug!(error = %e, "Metrics connection error");
+                }
+                Err(_) => {
+                    debug!(
+                        peer = %peer,
+                        timeout_ms = METRICS_HTTP_CONNECTION_TIMEOUT.as_millis() as u64,
+                        "Metrics connection timed out"
+                    );
+                }
             }
         });
     }
@@ -708,6 +739,66 @@ async fn render_metrics(
         }
     );
 
+    let _ = writeln!(
+        out,
+        "# HELP telemt_quota_refund_bytes_total Reserved quota bytes returned before commit"
+    );
+    let _ = writeln!(out, "# TYPE telemt_quota_refund_bytes_total counter");
+    let _ = writeln!(
+        out,
+        "telemt_quota_refund_bytes_total {}",
+        if core_enabled {
+            stats.get_quota_refund_bytes_total()
+        } else {
+            0
+        }
+    );
+    let _ = writeln!(
+        out,
+        "# HELP telemt_quota_contention_total Quota reservation CAS contention events"
+    );
+    let _ = writeln!(out, "# TYPE telemt_quota_contention_total counter");
+    let _ = writeln!(
+        out,
+        "telemt_quota_contention_total {}",
+        if core_enabled {
+            stats.get_quota_contention_total()
+        } else {
+            0
+        }
+    );
+    let _ = writeln!(
+        out,
+        "# HELP telemt_quota_contention_timeout_total Quota reservations that hit the bounded contention budget"
+    );
+    let _ = writeln!(
+        out,
+        "# TYPE telemt_quota_contention_timeout_total counter"
+    );
+    let _ = writeln!(
+        out,
+        "telemt_quota_contention_timeout_total {}",
+        if core_enabled {
+            stats.get_quota_contention_timeout_total()
+        } else {
+            0
+        }
+    );
+    let _ = writeln!(
+        out,
+        "# HELP telemt_quota_acquire_cancelled_total Quota acquisitions cancelled before reservation completed"
+    );
+    let _ = writeln!(out, "# TYPE telemt_quota_acquire_cancelled_total counter");
+    let _ = writeln!(
+        out,
+        "telemt_quota_acquire_cancelled_total {}",
+        if core_enabled {
+            stats.get_quota_acquire_cancelled_total()
+        } else {
+            0
+        }
+    );
+
     let _ = writeln!(
         out,
         "# HELP telemt_conntrack_control_state Runtime conntrack control state flags"
@@ -822,6 +913,32 @@ async fn render_metrics(
     );
 
     let limiter_metrics = shared_state.traffic_limiter.metrics_snapshot();
+    let _ = writeln!(
+        out,
+        "# HELP telemt_rate_limiter_burst_bound_bytes Configured upper bound for one direct relay rate-limit burst"
+    );
+    let _ = writeln!(
+        out,
+        "# TYPE telemt_rate_limiter_burst_bound_bytes gauge"
+    );
+    let _ = writeln!(
+        out,
+        "telemt_rate_limiter_burst_bound_bytes{{direction=\"up\"}} {}",
+        if core_enabled {
+            config.general.direct_relay_copy_buf_c2s_bytes
+        } else {
+            0
+        }
+    );
+    let _ = writeln!(
+        out,
+        "telemt_rate_limiter_burst_bound_bytes{{direction=\"down\"}} {}",
+        if core_enabled {
+            config.general.direct_relay_copy_buf_s2c_bytes
+        } else {
+            0
+        }
+    );
     let _ = writeln!(
         out,
         "# HELP telemt_rate_limiter_throttle_total Traffic limiter throttle events by scope and direction"
@@ -1924,6 +2041,85 @@ async fn render_metrics(
             0
         }
     );
+    let _ = writeln!(
+        out,
+        "# HELP telemt_me_child_join_timeout_total Middle relay child tasks that did not join before cleanup deadline"
+    );
+    let _ = writeln!(out, "# TYPE telemt_me_child_join_timeout_total counter");
+    let _ = writeln!(
+        out,
+        "telemt_me_child_join_timeout_total {}",
+        if core_enabled {
+            stats.get_me_child_join_timeout_total()
+        } else {
+            0
+        }
+    );
+    let _ = writeln!(
+        out,
+        "# HELP telemt_me_child_abort_total Middle relay child tasks aborted after bounded cleanup timeout"
+    );
+    let _ = writeln!(out, "# TYPE telemt_me_child_abort_total counter");
+    let _ = writeln!(
+        out,
+        "telemt_me_child_abort_total {}",
+        if core_enabled {
+            stats.get_me_child_abort_total()
+        } else {
+            0
+        }
+    );
+    let _ = writeln!(
+        out,
+        "# HELP telemt_flow_wait_events_total Flow wait events by reason, direction, and outcome"
+    );
+    let _ = writeln!(out, "# TYPE telemt_flow_wait_events_total counter");
+    let _ = writeln!(
+        out,
+        "telemt_flow_wait_events_total{{reason=\"middle_rate_limit\",direction=\"down\",outcome=\"waited\"}} {}",
+        if core_enabled {
+            stats.get_flow_wait_middle_rate_limit_total()
+        } else {
+            0
+        }
+    );
+    let _ = writeln!(
+        out,
+        "telemt_flow_wait_events_total{{reason=\"middle_rate_limit\",direction=\"down\",outcome=\"cancelled\"}} {}",
+        if core_enabled {
+            stats.get_flow_wait_middle_rate_limit_cancelled_total()
+        } else {
+            0
+        }
+    );
+    let _ = writeln!(
+        out,
+        "# HELP telemt_flow_wait_ms_total Flow wait time in milliseconds by reason and direction"
+    );
+    let _ = writeln!(out, "# TYPE telemt_flow_wait_ms_total counter");
+    let _ = writeln!(
+        out,
+        "telemt_flow_wait_ms_total{{reason=\"middle_rate_limit\",direction=\"down\"}} {}",
+        if core_enabled {
+            stats.get_flow_wait_middle_rate_limit_ms_total()
+        } else {
+            0
+        }
+    );
+    let _ = writeln!(
+        out,
+        "# HELP telemt_session_drop_fallback_total Session reservations cleaned by Drop instead of explicit async release"
+    );
+    let _ = writeln!(out, "# TYPE telemt_session_drop_fallback_total counter");
+    let _ = writeln!(
+        out,
+        "telemt_session_drop_fallback_total {}",
+        if core_enabled {
+            stats.get_session_drop_fallback_total()
+        } else {
+            0
+        }
+    );
 
     let _ = writeln!(
         out,

src/proxy/client.rs

@@ -32,7 +32,13 @@ struct UserConnectionReservation {
     user: String,
     ip: IpAddr,
     tracks_ip: bool,
-    active: bool,
+    state: SessionReservationState,
+}
+
+#[derive(Clone, Copy, PartialEq, Eq)]
+enum SessionReservationState {
+    Active,
+    Released,
 }
 
 impl UserConnectionReservation {
@@ -49,28 +55,29 @@ impl UserConnectionReservation {
             user,
             ip,
             tracks_ip,
-            active: true,
+            state: SessionReservationState::Active,
         }
     }
 
     async fn release(mut self) {
-        if !self.active {
+        if self.state != SessionReservationState::Active {
             return;
         }
         if self.tracks_ip {
             self.ip_tracker.remove_ip(&self.user, self.ip).await;
         }
-        self.active = false;
+        self.state = SessionReservationState::Released;
         self.stats.decrement_user_curr_connects(&self.user);
     }
 }
 
 impl Drop for UserConnectionReservation {
     fn drop(&mut self) {
-        if !self.active {
+        if self.state != SessionReservationState::Active {
             return;
         }
-        self.active = false;
+        self.state = SessionReservationState::Released;
+        self.stats.increment_session_drop_fallback_total();
         self.stats.decrement_user_curr_connects(&self.user);
         if self.tracks_ip {
             self.ip_tracker.enqueue_cleanup(self.user.clone(), self.ip);
@@ -466,6 +473,17 @@ where
     let mut local_addr = synthetic_local_addr(config.server.port);
 
     if proxy_protocol_enabled {
+        if !is_trusted_proxy_source(peer.ip(), &config.server.proxy_protocol_trusted_cidrs) {
+            stats.increment_connects_bad_with_class("proxy_protocol_untrusted");
+            warn!(
+                peer = %peer,
+                trusted = ?config.server.proxy_protocol_trusted_cidrs,
+                "Rejecting PROXY protocol header from untrusted source"
+            );
+            record_beobachten_class(&beobachten, &config, peer.ip(), "other");
+            return Err(ProxyError::InvalidProxyProtocol);
+        }
+
         let proxy_header_timeout =
             Duration::from_millis(config.server.proxy_protocol_header_timeout_ms.max(1));
         match timeout(
@@ -475,17 +493,6 @@ where
         .await
         {
             Ok(Ok(info)) => {
-                if !is_trusted_proxy_source(peer.ip(), &config.server.proxy_protocol_trusted_cidrs)
-                {
-                    stats.increment_connects_bad_with_class("proxy_protocol_untrusted");
-                    warn!(
-                        peer = %peer,
-                        trusted = ?config.server.proxy_protocol_trusted_cidrs,
-                        "Rejecting PROXY protocol header from untrusted source"
-                    );
-                    record_beobachten_class(&beobachten, &config, peer.ip(), "other");
-                    return Err(ProxyError::InvalidProxyProtocol);
-                }
                 debug!(
                     peer = %peer,
                     client = %info.src_addr,
@@ -978,6 +985,26 @@ impl RunningClientHandler {
         let mut local_addr = self.stream.local_addr().map_err(ProxyError::Io)?;
 
         if self.proxy_protocol_enabled {
+            if !is_trusted_proxy_source(
+                self.peer.ip(),
+                &self.config.server.proxy_protocol_trusted_cidrs,
+            ) {
+                self.stats
+                    .increment_connects_bad_with_class("proxy_protocol_untrusted");
+                warn!(
+                    peer = %self.peer,
+                    trusted = ?self.config.server.proxy_protocol_trusted_cidrs,
+                    "Rejecting PROXY protocol header from untrusted source"
+                );
+                record_beobachten_class(
+                    &self.beobachten,
+                    &self.config,
+                    self.peer.ip(),
+                    "other",
+                );
+                return Err(ProxyError::InvalidProxyProtocol);
+            }
+
             let proxy_header_timeout =
                 Duration::from_millis(self.config.server.proxy_protocol_header_timeout_ms.max(1));
             match timeout(
@@ -987,25 +1014,6 @@ impl RunningClientHandler {
             .await
             {
                 Ok(Ok(info)) => {
-                    if !is_trusted_proxy_source(
-                        self.peer.ip(),
-                        &self.config.server.proxy_protocol_trusted_cidrs,
-                    ) {
-                        self.stats
-                            .increment_connects_bad_with_class("proxy_protocol_untrusted");
-                        warn!(
-                            peer = %self.peer,
-                            trusted = ?self.config.server.proxy_protocol_trusted_cidrs,
-                            "Rejecting PROXY protocol header from untrusted source"
-                        );
-                        record_beobachten_class(
-                            &self.beobachten,
-                            &self.config,
-                            self.peer.ip(),
-                            "other",
-                        );
-                        return Err(ProxyError::InvalidProxyProtocol);
-                    }
                     debug!(
                         peer = %self.peer,
                         client = %info.src_addr,

src/proxy/middle_relay.rs

@@ -14,6 +14,7 @@ use std::time::{Duration, Instant};
 use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
 use tokio::sync::{OwnedSemaphorePermit, Semaphore, mpsc, oneshot, watch};
 use tokio::time::timeout;
+use tokio_util::sync::CancellationToken;
 use tracing::{debug, info, trace, warn};
 
 use crate::config::{ConntrackPressureProfile, ProxyConfig};
@@ -65,6 +66,14 @@ const ME_D2C_SINGLE_WRITE_COALESCE_MAX_BYTES: usize = 128 * 1024;
 const QUOTA_RESERVE_SPIN_RETRIES: usize = 32;
 const QUOTA_RESERVE_BACKOFF_MIN_MS: u64 = 1;
 const QUOTA_RESERVE_BACKOFF_MAX_MS: u64 = 16;
+const QUOTA_RESERVE_MAX_BACKOFF_ROUNDS: usize = 16;
+const ME_CHILD_JOIN_TIMEOUT: Duration = Duration::from_secs(2);
+
+enum MiddleQuotaReserveError {
+    LimitExceeded,
+    Contended,
+    Cancelled,
+}
 
 #[derive(Default)]
 pub(crate) struct DesyncDedupRotationState {
@@ -622,21 +631,38 @@ async fn reserve_user_quota_with_yield(
     user_stats: &UserStats,
     bytes: u64,
     limit: u64,
-) -> std::result::Result<u64, QuotaReserveError> {
+    stats: &Stats,
+    cancel: &CancellationToken,
+) -> std::result::Result<u64, MiddleQuotaReserveError> {
     let mut backoff_ms = QUOTA_RESERVE_BACKOFF_MIN_MS;
+    let mut backoff_rounds = 0usize;
     loop {
         for _ in 0..QUOTA_RESERVE_SPIN_RETRIES {
             match user_stats.quota_try_reserve(bytes, limit) {
                 Ok(total) => return Ok(total),
                 Err(QuotaReserveError::LimitExceeded) => {
-                    return Err(QuotaReserveError::LimitExceeded);
+                    return Err(MiddleQuotaReserveError::LimitExceeded);
+                }
+                Err(QuotaReserveError::Contended) => {
+                    stats.increment_quota_contention_total();
+                    std::hint::spin_loop();
                 }
-                Err(QuotaReserveError::Contended) => std::hint::spin_loop(),
             }
         }
 
         tokio::task::yield_now().await;
-        tokio::time::sleep(Duration::from_millis(backoff_ms)).await;
+        tokio::select! {
+            _ = tokio::time::sleep(Duration::from_millis(backoff_ms)) => {}
+            _ = cancel.cancelled() => {
+                stats.increment_quota_acquire_cancelled_total();
+                return Err(MiddleQuotaReserveError::Cancelled);
+            }
+        }
+        backoff_rounds = backoff_rounds.saturating_add(1);
+        if backoff_rounds >= QUOTA_RESERVE_MAX_BACKOFF_ROUNDS {
+            stats.increment_quota_contention_timeout_total();
+            return Err(MiddleQuotaReserveError::Contended);
+        }
         backoff_ms = backoff_ms
             .saturating_mul(2)
             .min(QUOTA_RESERVE_BACKOFF_MAX_MS);
@@ -647,12 +673,12 @@ async fn wait_for_traffic_budget(
     lease: Option<&Arc<TrafficLease>>,
     direction: RateDirection,
     bytes: u64,
-) {
+) -> Result<()> {
     if bytes == 0 {
-        return;
+        return Ok(());
     }
     let Some(lease) = lease else {
-        return;
+        return Ok(());
     };
 
     let mut remaining = bytes;
@@ -676,6 +702,54 @@ async fn wait_for_traffic_budget(
             wait_ms,
         );
     }
+
+    Ok(())
+}
+
+async fn wait_for_traffic_budget_or_cancel(
+    lease: Option<&Arc<TrafficLease>>,
+    direction: RateDirection,
+    bytes: u64,
+    cancel: &CancellationToken,
+    stats: &Stats,
+) -> Result<()> {
+    if bytes == 0 {
+        return Ok(());
+    }
+    let Some(lease) = lease else {
+        return Ok(());
+    };
+
+    let mut remaining = bytes;
+    while remaining > 0 {
+        let consume = lease.try_consume(direction, remaining);
+        if consume.granted > 0 {
+            remaining = remaining.saturating_sub(consume.granted);
+            continue;
+        }
+
+        let wait_started_at = Instant::now();
+        tokio::select! {
+            _ = tokio::time::sleep(next_refill_delay()) => {}
+            _ = cancel.cancelled() => {
+                stats.increment_flow_wait_middle_rate_limit_cancelled_total();
+                return Err(ProxyError::Proxy("traffic budget wait cancelled".into()));
+            }
+        }
+        let wait_ms = wait_started_at
+            .elapsed()
+            .as_millis()
+            .min(u128::from(u64::MAX)) as u64;
+        lease.observe_wait_ms(
+            direction,
+            consume.blocked_user,
+            consume.blocked_cidr,
+            wait_ms,
+        );
+        stats.observe_flow_wait_middle_rate_limit_ms(wait_ms);
+    }
+
+    Ok(())
 }
 
 fn classify_me_d2c_flush_reason(
@@ -1169,7 +1243,7 @@ where
     let c2me_byte_semaphore = Arc::new(Semaphore::new(c2me_byte_budget));
     let (c2me_tx, mut c2me_rx) = mpsc::channel::<C2MeCommand>(c2me_channel_capacity);
     let me_pool_c2me = me_pool.clone();
-    let c2me_sender = tokio::spawn(async move {
+    let mut c2me_sender = tokio::spawn(async move {
         let mut sent_since_yield = 0usize;
         while let Some(cmd) = c2me_rx.recv().await {
             match cmd {
@@ -1205,16 +1279,18 @@ where
     });
 
     let (stop_tx, mut stop_rx) = oneshot::channel::<()>();
+    let flow_cancel = CancellationToken::new();
     let mut me_rx_task = me_rx;
     let stats_clone = stats.clone();
     let rng_clone = rng.clone();
     let user_clone = user.clone();
     let quota_user_stats_me_writer = quota_user_stats.clone();
     let traffic_lease_me_writer = traffic_lease.clone();
+    let flow_cancel_me_writer = flow_cancel.clone();
     let last_downstream_activity_ms_clone = last_downstream_activity_ms.clone();
     let bytes_me2c_clone = bytes_me2c.clone();
     let d2c_flush_policy = MeD2cFlushPolicy::from_config(&config);
-    let me_writer = tokio::spawn(async move {
+    let mut me_writer = tokio::spawn(async move {
         let mut writer = crypto_writer;
         let mut frame_buf = Vec::with_capacity(16 * 1024);
         let shrink_threshold = d2c_flush_policy.frame_buf_shrink_threshold_bytes;
@@ -1234,7 +1310,7 @@ where
                     let Some(first) = msg else {
                         debug!(conn_id, "ME channel closed");
                         shrink_session_vec(&mut frame_buf, shrink_threshold);
-                        return Err(ProxyError::Proxy("ME connection lost".into()));
+                        return Err(ProxyError::MiddleConnectionLost);
                     };
 
                     let mut batch_frames = 0usize;
@@ -1256,6 +1332,7 @@ where
                         quota_limit,
                         d2c_flush_policy.quota_soft_overshoot_bytes,
                         traffic_lease_me_writer.as_ref(),
+                        &flow_cancel_me_writer,
                         bytes_me2c_clone.as_ref(),
                         conn_id,
                         d2c_flush_policy.ack_flush_immediate,
@@ -1317,6 +1394,7 @@ where
                             quota_limit,
                             d2c_flush_policy.quota_soft_overshoot_bytes,
                             traffic_lease_me_writer.as_ref(),
+                            &flow_cancel_me_writer,
                             bytes_me2c_clone.as_ref(),
                             conn_id,
                             d2c_flush_policy.ack_flush_immediate,
@@ -1381,6 +1459,7 @@ where
                                     quota_limit,
                                     d2c_flush_policy.quota_soft_overshoot_bytes,
                                     traffic_lease_me_writer.as_ref(),
+                                    &flow_cancel_me_writer,
                                     bytes_me2c_clone.as_ref(),
                                     conn_id,
                                     d2c_flush_policy.ack_flush_immediate,
@@ -1447,6 +1526,7 @@ where
                                         quota_limit,
                                         d2c_flush_policy.quota_soft_overshoot_bytes,
                                         traffic_lease_me_writer.as_ref(),
+                                        &flow_cancel_me_writer,
                                         bytes_me2c_clone.as_ref(),
                                         conn_id,
                                         d2c_flush_policy.ack_flush_immediate,
@@ -1495,7 +1575,7 @@ where
                             Ok(None) => {
                                 debug!(conn_id, "ME channel closed");
                                 shrink_session_vec(&mut frame_buf, shrink_threshold);
-                                return Err(ProxyError::Proxy("ME connection lost".into()));
+                                return Err(ProxyError::MiddleConnectionLost);
                             }
                             Err(_) => {
                                 max_delay_fired = true;
@@ -1642,25 +1722,41 @@ where
                             RateDirection::Up,
                             payload.len() as u64,
                         )
-                        .await;
+                        .await?;
                         forensics.bytes_c2me = forensics
                             .bytes_c2me
                             .saturating_add(payload.len() as u64);
                         if let (Some(limit), Some(user_stats)) =
                             (quota_limit, quota_user_stats.as_deref())
                         {
-                            if reserve_user_quota_with_yield(
+                            match reserve_user_quota_with_yield(
                                 user_stats,
                                 payload.len() as u64,
                                 limit,
+                                stats.as_ref(),
+                                &flow_cancel,
                             )
                             .await
-                            .is_err()
                             {
-                                main_result = Err(ProxyError::DataQuotaExceeded {
-                                    user: user.clone(),
-                                });
-                                break;
+                                Ok(_) => {}
+                                Err(MiddleQuotaReserveError::LimitExceeded) => {
+                                    main_result = Err(ProxyError::DataQuotaExceeded {
+                                        user: user.clone(),
+                                    });
+                                    break;
+                                }
+                                Err(MiddleQuotaReserveError::Contended) => {
+                                    main_result = Err(ProxyError::Proxy(
+                                        "ME C->ME quota reservation contended".into(),
+                                    ));
+                                    break;
+                                }
+                                Err(MiddleQuotaReserveError::Cancelled) => {
+                                    main_result = Err(ProxyError::Proxy(
+                                        "ME C->ME quota reservation cancelled".into(),
+                                    ));
+                                    break;
+                                }
                             }
                             stats.add_user_octets_from_handle(user_stats, payload.len() as u64);
                         } else {
@@ -1729,21 +1825,35 @@ where
     }
 
     drop(c2me_tx);
-    let c2me_result = c2me_sender
-        .await
-        .unwrap_or_else(|e| Err(ProxyError::Proxy(format!("ME sender join error: {e}"))));
+    let c2me_result = match timeout(ME_CHILD_JOIN_TIMEOUT, &mut c2me_sender).await {
+        Ok(joined) => {
+            joined.unwrap_or_else(|e| Err(ProxyError::Proxy(format!("ME sender join error: {e}"))))
+        }
+        Err(_) => {
+            stats.increment_me_child_join_timeout_total();
+            stats.increment_me_child_abort_total();
+            c2me_sender.abort();
+            Err(ProxyError::Proxy("ME sender join timeout".into()))
+        }
+    };
 
+    flow_cancel.cancel();
     let _ = stop_tx.send(());
-    let mut writer_result = me_writer
-        .await
-        .unwrap_or_else(|e| Err(ProxyError::Proxy(format!("ME writer join error: {e}"))));
+    let mut writer_result = match timeout(ME_CHILD_JOIN_TIMEOUT, &mut me_writer).await {
+        Ok(joined) => {
+            joined.unwrap_or_else(|e| Err(ProxyError::Proxy(format!("ME writer join error: {e}"))))
+        }
+        Err(_) => {
+            stats.increment_me_child_join_timeout_total();
+            stats.increment_me_child_abort_total();
+            me_writer.abort();
+            Err(ProxyError::Proxy("ME writer join timeout".into()))
+        }
+    };
 
     // When client closes, but ME channel stopped as unregistered - it isnt error
     if client_closed
-        && matches!(
-            writer_result,
-            Err(ProxyError::Proxy(ref msg)) if msg == "ME connection lost"
-        )
+        && matches!(writer_result, Err(ProxyError::MiddleConnectionLost))
     {
         writer_result = Ok(());
     }
@@ -2300,6 +2410,7 @@ where
         quota_limit,
         quota_soft_overshoot_bytes,
         None,
+        &CancellationToken::new(),
         bytes_me2c,
         conn_id,
         ack_flush_immediate,
@@ -2320,6 +2431,7 @@ async fn process_me_writer_response_with_traffic_lease<W>(
     quota_limit: Option<u64>,
     quota_soft_overshoot_bytes: u64,
     traffic_lease: Option<&Arc<TrafficLease>>,
+    cancel: &CancellationToken,
     bytes_me2c: &AtomicU64,
     conn_id: u64,
     ack_flush_immediate: bool,
@@ -2338,17 +2450,36 @@ where
             let data_len = data.len() as u64;
             if let (Some(limit), Some(user_stats)) = (quota_limit, quota_user_stats) {
                 let soft_limit = quota_soft_cap(limit, quota_soft_overshoot_bytes);
-                if reserve_user_quota_with_yield(user_stats, data_len, soft_limit)
+                match reserve_user_quota_with_yield(user_stats, data_len, soft_limit, stats, cancel)
                     .await
-                    .is_err()
                 {
-                    stats.increment_me_d2c_quota_reject_total(MeD2cQuotaRejectStage::PreWrite);
-                    return Err(ProxyError::DataQuotaExceeded {
-                        user: user.to_string(),
-                    });
+                    Ok(_) => {}
+                    Err(MiddleQuotaReserveError::LimitExceeded) => {
+                        stats.increment_me_d2c_quota_reject_total(MeD2cQuotaRejectStage::PreWrite);
+                        return Err(ProxyError::DataQuotaExceeded {
+                            user: user.to_string(),
+                        });
+                    }
+                    Err(MiddleQuotaReserveError::Contended) => {
+                        return Err(ProxyError::Proxy(
+                            "ME D->C quota reservation contended".into(),
+                        ));
+                    }
+                    Err(MiddleQuotaReserveError::Cancelled) => {
+                        return Err(ProxyError::Proxy(
+                            "ME D->C quota reservation cancelled".into(),
+                        ));
+                    }
                 }
             }
-            wait_for_traffic_budget(traffic_lease, RateDirection::Down, data_len).await;
+            wait_for_traffic_budget_or_cancel(
+                traffic_lease,
+                RateDirection::Down,
+                data_len,
+                cancel,
+                stats,
+            )
+            .await?;
 
             let write_mode =
                 match write_client_payload(client_writer, proto_tag, flags, &data, rng, frame_buf)
@@ -2386,7 +2517,8 @@ where
             } else {
                 trace!(conn_id, confirm, "ME->C quickack");
             }
-            wait_for_traffic_budget(traffic_lease, RateDirection::Down, 4).await;
+            wait_for_traffic_budget_or_cancel(traffic_lease, RateDirection::Down, 4, cancel, stats)
+                .await?;
             write_client_ack(client_writer, proto_tag, confirm).await?;
             stats.increment_me_d2c_ack_frames_total();
 

src/proxy/relay.rs

@@ -215,6 +215,7 @@ struct StatsIo<S> {
     c2s_rate_debt_bytes: u64,
     c2s_wait: RateWaitState,
     s2c_wait: RateWaitState,
+    quota_wait: RateWaitState,
     quota_limit: Option<u64>,
     quota_exceeded: Arc<AtomicBool>,
     quota_bytes_since_check: u64,
@@ -275,6 +276,7 @@ impl<S> StatsIo<S> {
             c2s_rate_debt_bytes: 0,
             c2s_wait: RateWaitState::default(),
             s2c_wait: RateWaitState::default(),
+            quota_wait: RateWaitState::default(),
             quota_limit,
             quota_exceeded,
             quota_bytes_since_check: 0,
@@ -353,6 +355,11 @@ impl<S> StatsIo<S> {
 
         Poll::Ready(())
     }
+
+    fn arm_quota_wait(&mut self, cx: &mut Context<'_>) -> Poll<()> {
+        Self::arm_wait(&mut self.quota_wait, false, false);
+        Self::poll_wait(&mut self.quota_wait, cx, None, RateDirection::Up)
+    }
 }
 
 #[derive(Debug)]
@@ -430,8 +437,13 @@ impl<S: AsyncRead + Unpin> AsyncRead for StatsIo<S> {
         if this.settle_c2s_rate_debt(cx).is_pending() {
             return Poll::Pending;
         }
+        if buf.remaining() == 0 {
+            return Pin::new(&mut this.inner).poll_read(cx, buf);
+        }
 
         let mut remaining_before = None;
+        let mut reserved_read_bytes = 0u64;
+        let mut read_limit = buf.remaining();
         if let Some(limit) = this.quota_limit {
             let used_before = this.user_stats.quota_used();
             let remaining = limit.saturating_sub(used_before);
@@ -440,50 +452,82 @@ impl<S: AsyncRead + Unpin> AsyncRead for StatsIo<S> {
                 return Poll::Ready(Err(quota_io_error()));
             }
             remaining_before = Some(remaining);
+            read_limit = read_limit.min(remaining as usize);
+            if read_limit == 0 {
+                this.quota_exceeded.store(true, Ordering::Release);
+                return Poll::Ready(Err(quota_io_error()));
+            }
+
+            let desired = read_limit as u64;
+            let mut reserve_rounds = 0usize;
+            while reserved_read_bytes == 0 {
+                for _ in 0..QUOTA_RESERVE_SPIN_RETRIES {
+                    match this.user_stats.quota_try_reserve(desired, limit) {
+                        Ok(_) => {
+                            reserved_read_bytes = desired;
+                            break;
+                        }
+                        Err(crate::stats::QuotaReserveError::LimitExceeded) => {
+                            this.quota_exceeded.store(true, Ordering::Release);
+                            return Poll::Ready(Err(quota_io_error()));
+                        }
+                        Err(crate::stats::QuotaReserveError::Contended) => {
+                            this.stats.increment_quota_contention_total();
+                        }
+                    }
+                }
+
+                if reserved_read_bytes == 0 {
+                    reserve_rounds = reserve_rounds.saturating_add(1);
+                    if reserve_rounds >= QUOTA_RESERVE_MAX_ROUNDS {
+                        this.stats.increment_quota_contention_timeout_total();
+                        if this.arm_quota_wait(cx).is_pending() {
+                            return Poll::Pending;
+                        }
+                        reserve_rounds = 0;
+                    }
+                }
+            }
         }
 
-        let before = buf.filled().len();
+        let limited_read = read_limit < buf.remaining();
+        let read_result = if limited_read {
+            let mut limited_buf = ReadBuf::new(buf.initialize_unfilled_to(read_limit));
+            match Pin::new(&mut this.inner).poll_read(cx, &mut limited_buf) {
+                Poll::Ready(Ok(())) => {
+                    let n = limited_buf.filled().len();
+                    buf.advance(n);
+                    Poll::Ready(Ok(n))
+                }
+                Poll::Ready(Err(err)) => Poll::Ready(Err(err)),
+                Poll::Pending => Poll::Pending,
+            }
+        } else {
+            let before = buf.filled().len();
+            match Pin::new(&mut this.inner).poll_read(cx, buf) {
+                Poll::Ready(Ok(())) => {
+                    let n = buf.filled().len() - before;
+                    Poll::Ready(Ok(n))
+                }
+                Poll::Ready(Err(err)) => Poll::Ready(Err(err)),
+                Poll::Pending => Poll::Pending,
+            }
+        };
 
-        match Pin::new(&mut this.inner).poll_read(cx, buf) {
-            Poll::Ready(Ok(())) => {
-                let n = buf.filled().len() - before;
+        match read_result {
+            Poll::Ready(Ok(n)) => {
+                if reserved_read_bytes > n as u64 {
+                    let refund_bytes = reserved_read_bytes - n as u64;
+                    refund_reserved_quota_bytes(
+                        this.user_stats.as_ref(),
+                        refund_bytes,
+                    );
+                    this.stats.add_quota_refund_bytes_total(refund_bytes);
+                }
                 if n > 0 {
                     let n_to_charge = n as u64;
 
-                    if let (Some(limit), Some(remaining)) = (this.quota_limit, remaining_before) {
-                        let mut reserved_total = None;
-                        let mut reserve_rounds = 0usize;
-                        while reserved_total.is_none() {
-                            let mut saw_contention = false;
-                            for _ in 0..QUOTA_RESERVE_SPIN_RETRIES {
-                                match this.user_stats.quota_try_reserve(n_to_charge, limit) {
-                                    Ok(total) => {
-                                        reserved_total = Some(total);
-                                        break;
-                                    }
-                                    Err(crate::stats::QuotaReserveError::LimitExceeded) => {
-                                        this.quota_exceeded.store(true, Ordering::Release);
-                                        buf.set_filled(before);
-                                        return Poll::Ready(Err(quota_io_error()));
-                                    }
-                                    Err(crate::stats::QuotaReserveError::Contended) => {
-                                        saw_contention = true;
-                                    }
-                                }
-                            }
-                            if reserved_total.is_none() {
-                                reserve_rounds = reserve_rounds.saturating_add(1);
-                                if reserve_rounds >= QUOTA_RESERVE_MAX_ROUNDS {
-                                    this.quota_exceeded.store(true, Ordering::Release);
-                                    buf.set_filled(before);
-                                    return Poll::Ready(Err(quota_io_error()));
-                                }
-                                if saw_contention {
-                                    std::thread::yield_now();
-                                }
-                            }
-                        }
-
+                    if let Some(remaining) = remaining_before {
                         if should_immediate_quota_check(remaining, n_to_charge) {
                             this.quota_bytes_since_check = 0;
                         } else {
@@ -495,9 +539,11 @@ impl<S: AsyncRead + Unpin> AsyncRead for StatsIo<S> {
                             }
                         }
 
-                        if reserved_total.unwrap_or(0) >= limit {
-                            this.quota_exceeded.store(true, Ordering::Release);
-                        }
+                    }
+                    if let Some(limit) = this.quota_limit
+                        && this.user_stats.quota_used() >= limit
+                    {
+                        this.quota_exceeded.store(true, Ordering::Release);
                     }
 
                     // C→S: client sent data
@@ -521,7 +567,20 @@ impl<S: AsyncRead + Unpin> AsyncRead for StatsIo<S> {
                 }
                 Poll::Ready(Ok(()))
             }
-            other => other,
+            Poll::Pending => {
+                if reserved_read_bytes > 0 {
+                    refund_reserved_quota_bytes(this.user_stats.as_ref(), reserved_read_bytes);
+                    this.stats.add_quota_refund_bytes_total(reserved_read_bytes);
+                }
+                Poll::Pending
+            }
+            Poll::Ready(Err(err)) => {
+                if reserved_read_bytes > 0 {
+                    refund_reserved_quota_bytes(this.user_stats.as_ref(), reserved_read_bytes);
+                    this.stats.add_quota_refund_bytes_total(reserved_read_bytes);
+                }
+                Poll::Ready(Err(err))
+            }
         }
     }
 }
@@ -603,6 +662,7 @@ impl<S: AsyncWrite + Unpin> AsyncWrite for StatsIo<S> {
                                 break;
                             }
                             Err(crate::stats::QuotaReserveError::Contended) => {
+                                this.stats.increment_quota_contention_total();
                                 saw_contention = true;
                             }
                         }
@@ -611,14 +671,14 @@ impl<S: AsyncWrite + Unpin> AsyncWrite for StatsIo<S> {
                     if reserved_bytes == 0 {
                         reserve_rounds = reserve_rounds.saturating_add(1);
                         if reserve_rounds >= QUOTA_RESERVE_MAX_ROUNDS {
+                            this.stats.increment_quota_contention_timeout_total();
                             if let Some(lease) = this.traffic_lease.as_ref() {
                                 lease.refund(RateDirection::Down, shaper_reserved_bytes);
                             }
-                            this.quota_exceeded.store(true, Ordering::Release);
-                            return Poll::Ready(Err(quota_io_error()));
-                        }
-                        if saw_contention {
-                            std::thread::yield_now();
+                            let _ = this.arm_quota_wait(cx);
+                            return Poll::Pending;
+                        } else if saw_contention {
+                            std::hint::spin_loop();
                         }
                     }
                 }
@@ -639,10 +699,12 @@ impl<S: AsyncWrite + Unpin> AsyncWrite for StatsIo<S> {
         match Pin::new(&mut this.inner).poll_write(cx, write_buf) {
             Poll::Ready(Ok(n)) => {
                 if reserved_bytes > n as u64 {
+                    let refund_bytes = reserved_bytes - n as u64;
                     refund_reserved_quota_bytes(
                         this.user_stats.as_ref(),
-                        reserved_bytes - n as u64,
+                        refund_bytes,
                     );
+                    this.stats.add_quota_refund_bytes_total(refund_bytes);
                 }
                 if shaper_reserved_bytes > n as u64
                     && let Some(lease) = this.traffic_lease.as_ref()
@@ -693,6 +755,7 @@ impl<S: AsyncWrite + Unpin> AsyncWrite for StatsIo<S> {
             Poll::Ready(Err(err)) => {
                 if reserved_bytes > 0 {
                     refund_reserved_quota_bytes(this.user_stats.as_ref(), reserved_bytes);
+                    this.stats.add_quota_refund_bytes_total(reserved_bytes);
                 }
                 if shaper_reserved_bytes > 0
                     && let Some(lease) = this.traffic_lease.as_ref()
@@ -704,6 +767,7 @@ impl<S: AsyncWrite + Unpin> AsyncWrite for StatsIo<S> {
             Poll::Pending => {
                 if reserved_bytes > 0 {
                     refund_reserved_quota_bytes(this.user_stats.as_ref(), reserved_bytes);
+                    this.stats.add_quota_refund_bytes_total(reserved_bytes);
                 }
                 if shaper_reserved_bytes > 0
                     && let Some(lease) = this.traffic_lease.as_ref()

src/stats/mod.rs

@@ -274,11 +274,22 @@ pub struct Stats {
     me_inline_recovery_total: AtomicU64,
     ip_reservation_rollback_tcp_limit_total: AtomicU64,
     ip_reservation_rollback_quota_limit_total: AtomicU64,
+    quota_refund_bytes_total: AtomicU64,
+    quota_contention_total: AtomicU64,
+    quota_contention_timeout_total: AtomicU64,
+    quota_acquire_cancelled_total: AtomicU64,
     quota_write_fail_bytes_total: AtomicU64,
     quota_write_fail_events_total: AtomicU64,
+    me_child_join_timeout_total: AtomicU64,
+    me_child_abort_total: AtomicU64,
+    flow_wait_middle_rate_limit_total: AtomicU64,
+    flow_wait_middle_rate_limit_cancelled_total: AtomicU64,
+    flow_wait_middle_rate_limit_ms_total: AtomicU64,
+    session_drop_fallback_total: AtomicU64,
     telemetry_core_enabled: AtomicBool,
     telemetry_user_enabled: AtomicBool,
     telemetry_me_level: AtomicU8,
+    cached_epoch_secs: AtomicU64,
     user_stats: DashMap<String, Arc<UserStats>>,
     user_stats_last_cleanup_epoch_secs: AtomicU64,
     start_time: parking_lot::RwLock<Option<Instant>>,
@@ -348,6 +359,7 @@ impl Stats {
     pub fn new() -> Self {
         let stats = Self::default();
         stats.apply_telemetry_policy(TelemetryPolicy::default());
+        stats.refresh_cached_epoch_secs();
         *stats.start_time.write() = Some(Instant::now());
         stats
     }
@@ -397,33 +409,55 @@ impl Stats {
             .as_secs()
     }
 
-    fn touch_user_stats(stats: &UserStats) {
+    fn refresh_cached_epoch_secs(&self) -> u64 {
+        let now_epoch_secs = Self::now_epoch_secs();
+        self.cached_epoch_secs
+            .store(now_epoch_secs, Ordering::Relaxed);
+        now_epoch_secs
+    }
+
+    fn cached_epoch_secs(&self) -> u64 {
+        let cached = self.cached_epoch_secs.load(Ordering::Relaxed);
+        if cached != 0 {
+            return cached;
+        }
+        self.refresh_cached_epoch_secs()
+    }
+
+    fn touch_user_stats(&self, stats: &UserStats) {
         stats
             .last_seen_epoch_secs
-            .store(Self::now_epoch_secs(), Ordering::Relaxed);
+            .store(self.cached_epoch_secs(), Ordering::Relaxed);
     }
 
     pub(crate) fn get_or_create_user_stats_handle(&self, user: &str) -> Arc<UserStats> {
-        self.maybe_cleanup_user_stats();
         if let Some(existing) = self.user_stats.get(user) {
             let handle = Arc::clone(existing.value());
-            Self::touch_user_stats(handle.as_ref());
+            self.touch_user_stats(handle.as_ref());
             return handle;
         }
 
         let entry = self.user_stats.entry(user.to_string()).or_default();
         if entry.last_seen_epoch_secs.load(Ordering::Relaxed) == 0 {
-            Self::touch_user_stats(entry.value().as_ref());
+            self.touch_user_stats(entry.value().as_ref());
         }
         Arc::clone(entry.value())
     }
 
+    pub(crate) async fn run_periodic_user_stats_maintenance(self: Arc<Self>) {
+        let mut interval = tokio::time::interval(Duration::from_secs(60));
+        loop {
+            interval.tick().await;
+            self.maybe_cleanup_user_stats();
+        }
+    }
+
     #[inline]
     pub(crate) fn add_user_octets_from_handle(&self, user_stats: &UserStats, bytes: u64) {
         if !self.telemetry_user_enabled() {
             return;
         }
-        Self::touch_user_stats(user_stats);
+        self.touch_user_stats(user_stats);
         user_stats
             .octets_from_client
             .fetch_add(bytes, Ordering::Relaxed);
@@ -434,7 +468,7 @@ impl Stats {
         if !self.telemetry_user_enabled() {
             return;
         }
-        Self::touch_user_stats(user_stats);
+        self.touch_user_stats(user_stats);
         user_stats
             .octets_to_client
             .fetch_add(bytes, Ordering::Relaxed);
@@ -445,7 +479,7 @@ impl Stats {
         if !self.telemetry_user_enabled() {
             return;
         }
-        Self::touch_user_stats(user_stats);
+        self.touch_user_stats(user_stats);
         user_stats.msgs_from_client.fetch_add(1, Ordering::Relaxed);
     }
 
@@ -454,7 +488,7 @@ impl Stats {
         if !self.telemetry_user_enabled() {
             return;
         }
-        Self::touch_user_stats(user_stats);
+        self.touch_user_stats(user_stats);
         user_stats.msgs_to_client.fetch_add(1, Ordering::Relaxed);
     }
 
@@ -464,7 +498,7 @@ impl Stats {
     /// mixing reserve and post-charge on a single I/O event.
     #[inline]
     pub(crate) fn quota_charge_post_write(&self, user_stats: &UserStats, bytes: u64) -> u64 {
-        Self::touch_user_stats(user_stats);
+        self.touch_user_stats(user_stats);
         user_stats
             .quota_used
             .fetch_add(bytes, Ordering::Relaxed)
@@ -475,7 +509,7 @@ impl Stats {
         const USER_STATS_CLEANUP_INTERVAL_SECS: u64 = 60;
         const USER_STATS_IDLE_TTL_SECS: u64 = 24 * 60 * 60;
 
-        let now_epoch_secs = Self::now_epoch_secs();
+        let now_epoch_secs = self.refresh_cached_epoch_secs();
         let last_cleanup_epoch_secs = self
             .user_stats_last_cleanup_epoch_secs
             .load(Ordering::Relaxed);
@@ -1437,6 +1471,29 @@ impl Stats {
                 .fetch_add(1, Ordering::Relaxed);
         }
     }
+    pub fn add_quota_refund_bytes_total(&self, bytes: u64) {
+        if self.telemetry_core_enabled() {
+            self.quota_refund_bytes_total
+                .fetch_add(bytes, Ordering::Relaxed);
+        }
+    }
+    pub fn increment_quota_contention_total(&self) {
+        if self.telemetry_core_enabled() {
+            self.quota_contention_total.fetch_add(1, Ordering::Relaxed);
+        }
+    }
+    pub fn increment_quota_contention_timeout_total(&self) {
+        if self.telemetry_core_enabled() {
+            self.quota_contention_timeout_total
+                .fetch_add(1, Ordering::Relaxed);
+        }
+    }
+    pub fn increment_quota_acquire_cancelled_total(&self) {
+        if self.telemetry_core_enabled() {
+            self.quota_acquire_cancelled_total
+                .fetch_add(1, Ordering::Relaxed);
+        }
+    }
     pub fn add_quota_write_fail_bytes_total(&self, bytes: u64) {
         if self.telemetry_core_enabled() {
             self.quota_write_fail_bytes_total
@@ -1449,6 +1506,37 @@ impl Stats {
                 .fetch_add(1, Ordering::Relaxed);
         }
     }
+    pub fn increment_me_child_join_timeout_total(&self) {
+        if self.telemetry_core_enabled() {
+            self.me_child_join_timeout_total
+                .fetch_add(1, Ordering::Relaxed);
+        }
+    }
+    pub fn increment_me_child_abort_total(&self) {
+        if self.telemetry_core_enabled() {
+            self.me_child_abort_total.fetch_add(1, Ordering::Relaxed);
+        }
+    }
+    pub fn observe_flow_wait_middle_rate_limit_ms(&self, wait_ms: u64) {
+        if self.telemetry_core_enabled() {
+            self.flow_wait_middle_rate_limit_total
+                .fetch_add(1, Ordering::Relaxed);
+            self.flow_wait_middle_rate_limit_ms_total
+                .fetch_add(wait_ms, Ordering::Relaxed);
+        }
+    }
+    pub fn increment_flow_wait_middle_rate_limit_cancelled_total(&self) {
+        if self.telemetry_core_enabled() {
+            self.flow_wait_middle_rate_limit_cancelled_total
+                .fetch_add(1, Ordering::Relaxed);
+        }
+    }
+    pub fn increment_session_drop_fallback_total(&self) {
+        if self.telemetry_core_enabled() {
+            self.session_drop_fallback_total
+                .fetch_add(1, Ordering::Relaxed);
+        }
+    }
     pub fn increment_me_endpoint_quarantine_total(&self) {
         if self.telemetry_me_allows_normal() {
             self.me_endpoint_quarantine_total
@@ -2283,19 +2371,53 @@ impl Stats {
         self.ip_reservation_rollback_quota_limit_total
             .load(Ordering::Relaxed)
     }
+    pub fn get_quota_refund_bytes_total(&self) -> u64 {
+        self.quota_refund_bytes_total.load(Ordering::Relaxed)
+    }
+    pub fn get_quota_contention_total(&self) -> u64 {
+        self.quota_contention_total.load(Ordering::Relaxed)
+    }
+    pub fn get_quota_contention_timeout_total(&self) -> u64 {
+        self.quota_contention_timeout_total
+            .load(Ordering::Relaxed)
+    }
+    pub fn get_quota_acquire_cancelled_total(&self) -> u64 {
+        self.quota_acquire_cancelled_total.load(Ordering::Relaxed)
+    }
     pub fn get_quota_write_fail_bytes_total(&self) -> u64 {
         self.quota_write_fail_bytes_total.load(Ordering::Relaxed)
     }
     pub fn get_quota_write_fail_events_total(&self) -> u64 {
         self.quota_write_fail_events_total.load(Ordering::Relaxed)
     }
+    pub fn get_me_child_join_timeout_total(&self) -> u64 {
+        self.me_child_join_timeout_total.load(Ordering::Relaxed)
+    }
+    pub fn get_me_child_abort_total(&self) -> u64 {
+        self.me_child_abort_total.load(Ordering::Relaxed)
+    }
+    pub fn get_flow_wait_middle_rate_limit_total(&self) -> u64 {
+        self.flow_wait_middle_rate_limit_total
+            .load(Ordering::Relaxed)
+    }
+    pub fn get_flow_wait_middle_rate_limit_cancelled_total(&self) -> u64 {
+        self.flow_wait_middle_rate_limit_cancelled_total
+            .load(Ordering::Relaxed)
+    }
+    pub fn get_flow_wait_middle_rate_limit_ms_total(&self) -> u64 {
+        self.flow_wait_middle_rate_limit_ms_total
+            .load(Ordering::Relaxed)
+    }
+    pub fn get_session_drop_fallback_total(&self) -> u64 {
+        self.session_drop_fallback_total.load(Ordering::Relaxed)
+    }
 
     pub fn increment_user_connects(&self, user: &str) {
         if !self.telemetry_user_enabled() {
             return;
         }
         let stats = self.get_or_create_user_stats_handle(user);
-        Self::touch_user_stats(stats.as_ref());
+        self.touch_user_stats(stats.as_ref());
         stats.connects.fetch_add(1, Ordering::Relaxed);
     }
 
@@ -2304,7 +2426,7 @@ impl Stats {
             return;
         }
         let stats = self.get_or_create_user_stats_handle(user);
-        Self::touch_user_stats(stats.as_ref());
+        self.touch_user_stats(stats.as_ref());
         stats.curr_connects.fetch_add(1, Ordering::Relaxed);
     }
 
@@ -2314,7 +2436,7 @@ impl Stats {
         }
 
         let stats = self.get_or_create_user_stats_handle(user);
-        Self::touch_user_stats(stats.as_ref());
+        self.touch_user_stats(stats.as_ref());
 
         let counter = &stats.curr_connects;
         let mut current = counter.load(Ordering::Relaxed);
@@ -2337,9 +2459,8 @@ impl Stats {
     }
 
     pub fn decrement_user_curr_connects(&self, user: &str) {
-        self.maybe_cleanup_user_stats();
         if let Some(stats) = self.user_stats.get(user) {
-            Self::touch_user_stats(stats.value().as_ref());
+            self.touch_user_stats(stats.value().as_ref());
             let counter = &stats.curr_connects;
             let mut current = counter.load(Ordering::Relaxed);
             loop {

src/transport/proxy_protocol.rs

@@ -18,6 +18,9 @@ const PROXY_V1_MIN_LEN: usize = 6;
 /// Minimum length for v2 header
 const PROXY_V2_MIN_LEN: usize = 16;
 
+/// Maximum accepted PROXY v2 address and TLV payload.
+const PROXY_V2_MAX_ADDR_LEN: usize = 216;
+
 /// Address families for v2
 mod address_family {
     pub const UNSPEC: u8 = 0x0;
@@ -169,6 +172,9 @@ async fn parse_v2<R: AsyncRead + Unpin>(
 
     let family_protocol = header[13];
     let addr_len = u16::from_be_bytes([header[14], header[15]]) as usize;
+    if addr_len > PROXY_V2_MAX_ADDR_LEN {
+        return Err(ProxyError::InvalidProxyProtocol);
+    }
 
     // Read address data
     let mut addr_data = vec![0u8; addr_len];