ME + Admission + Cleanup Correctness: merge pull request #779 from telemt/flow

ME + Admission + Cleanup Correctness

Cargo.lock

@@ -2404,9 +2404,9 @@ checksum = "f87165f0995f63a9fbeea62b64d10b4d9d8e78ec6d7d51fb2125fda7bb36788f"
 
 [[package]]
 name = "rustls-webpki"
-version = "0.103.12"
+version = "0.103.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8279bb85272c9f10811ae6a6c547ff594d6a7f3c6c6b02ee9726d1d0dcfcdd06"
+checksum = "61c429a8649f110dddef65e2a5ad240f747e85f7758a6bccc7e5777bd33f756e"
 dependencies = [
  "aws-lc-rs",
  "ring",
@@ -2791,7 +2791,7 @@ checksum = "7b2093cf4c8eb1e67749a6762251bc9cd836b6fc171623bd0a9d324d37af2417"
 
 [[package]]
 name = "telemt"
-version = "3.4.10"
+version = "3.4.11"
 dependencies = [
  "aes",
  "anyhow",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "telemt"
-version = "3.4.10"
+version = "3.4.11"
 edition = "2024"
 
 [features]

src/config/load.rs

@@ -621,10 +621,7 @@ fn collect_unknown_config_keys(parsed_toml: &toml::Value) -> Vec<UnknownConfigKe
         }
     }
 
-    if let Some(upstreams) = parsed_toml
+    if let Some(upstreams) = parsed_toml.get("upstreams").and_then(toml::Value::as_array) {
-        .get("upstreams")
-        .and_then(toml::Value::as_array)
-    {
         for (idx, upstream) in upstreams.iter().enumerate() {
             check_nested_table_value(
                 &mut unknown,

src/error.rs

@@ -225,6 +225,18 @@ pub enum ProxyError {
     #[error("ME connection lost")]
     MiddleConnectionLost,
 
+    #[error("Session terminated")]
+    RouteSwitched,
+
+    #[error("Traffic budget wait cancelled")]
+    TrafficBudgetWaitCancelled,
+
+    #[error("Traffic budget wait deadline exceeded")]
+    TrafficBudgetWaitDeadlineExceeded,
+
+    #[error("ME client writer cancelled")]
+    MiddleClientWriterCancelled,
+
     // ============= Config Errors =============
     #[error("Config error: {0}")]
     Config(String),

src/ip_tracker.rs

@@ -32,6 +32,7 @@ pub struct UserIpTracker {
     limit_mode: Arc<RwLock<UserMaxUniqueIpsMode>>,
     limit_window: Arc<RwLock<Duration>>,
     last_compact_epoch_secs: Arc<AtomicU64>,
+    cleanup_queue_len: Arc<AtomicU64>,
     cleanup_queue: Arc<Mutex<HashMap<(String, IpAddr), usize>>>,
     cleanup_drain_lock: Arc<AsyncMutex<()>>,
 }
@@ -72,6 +73,7 @@ impl UserIpTracker {
             limit_mode: Arc::new(RwLock::new(UserMaxUniqueIpsMode::ActiveWindow)),
             limit_window: Arc::new(RwLock::new(Duration::from_secs(30))),
             last_compact_epoch_secs: Arc::new(AtomicU64::new(0)),
+            cleanup_queue_len: Arc::new(AtomicU64::new(0)),
             cleanup_queue: Arc::new(Mutex::new(HashMap::new())),
             cleanup_drain_lock: Arc::new(AsyncMutex::new(())),
         }
@@ -120,6 +122,9 @@ impl UserIpTracker {
         match self.cleanup_queue.lock() {
             Ok(mut queue) => {
                 let count = queue.entry((user, ip)).or_insert(0);
+                if *count == 0 {
+                    self.cleanup_queue_len.fetch_add(1, Ordering::Relaxed);
+                }
                 *count = count.saturating_add(1);
                 self.cleanup_deferred_releases
                     .fetch_add(1, Ordering::Relaxed);
@@ -127,6 +132,9 @@ impl UserIpTracker {
             Err(poisoned) => {
                 let mut queue = poisoned.into_inner();
                 let count = queue.entry((user.clone(), ip)).or_insert(0);
+                if *count == 0 {
+                    self.cleanup_queue_len.fetch_add(1, Ordering::Relaxed);
+                }
                 *count = count.saturating_add(1);
                 self.cleanup_deferred_releases
                     .fetch_add(1, Ordering::Relaxed);
@@ -156,6 +164,9 @@ impl UserIpTracker {
     }
 
     pub(crate) async fn drain_cleanup_queue(&self) {
+        if self.cleanup_queue_len.load(Ordering::Relaxed) == 0 {
+            return;
+        }
         let Ok(_drain_guard) = self.cleanup_drain_lock.try_lock() else {
             return;
         };
@@ -173,6 +184,7 @@ impl UserIpTracker {
                             break;
                         };
                         if let Some(count) = queue.remove(&key) {
+                            self.cleanup_queue_len.fetch_sub(1, Ordering::Relaxed);
                             drained.insert(key, count);
                         }
                     }
@@ -191,6 +203,7 @@ impl UserIpTracker {
                             break;
                         };
                         if let Some(count) = queue.remove(&key) {
+                            self.cleanup_queue_len.fetch_sub(1, Ordering::Relaxed);
                             drained.insert(key, count);
                         }
                     }
@@ -294,12 +307,17 @@ impl UserIpTracker {
         }
     }
 
+    pub async fn run_periodic_maintenance(self: Arc<Self>) {
+        let mut interval = tokio::time::interval(Duration::from_secs(1));
+        loop {
+            interval.tick().await;
+            self.drain_cleanup_queue().await;
+            self.maybe_compact_empty_users().await;
+        }
+    }
+
     pub async fn memory_stats(&self) -> UserIpTrackerMemoryStats {
-        let cleanup_queue_len = self
+        let cleanup_queue_len = self.cleanup_queue_len.load(Ordering::Relaxed) as usize;
-            .cleanup_queue
-            .lock()
-            .unwrap_or_else(|poisoned| poisoned.into_inner())
-            .len();
         let active_ips = self.active_ips.read().await;
         let recent_ips = self.recent_ips.read().await;
         let active_entries = active_ips.values().map(HashMap::len).sum();

src/maestro/listeners.rs

@@ -13,7 +13,7 @@ use crate::config::{ProxyConfig, RstOnCloseMode};
 use crate::crypto::SecureRandom;
 use crate::ip_tracker::UserIpTracker;
 use crate::proxy::ClientHandler;
-use crate::proxy::route_mode::{ROUTE_SWITCH_ERROR_MSG, RouteRuntimeController};
+use crate::proxy::route_mode::RouteRuntimeController;
 use crate::proxy::shared_state::ProxySharedState;
 use crate::startup::{COMPONENT_LISTENERS_BIND, StartupTracker};
 use crate::stats::beobachten::BeobachtenStore;
@@ -492,14 +492,10 @@ pub(crate) fn spawn_tcp_accept_loops(
                                 let handshake_close_reason =
                                     expected_handshake_close_description(&e);
 
-                                let me_closed = matches!(
+                                let me_closed =
-                                    &e,
+                                    matches!(&e, crate::error::ProxyError::MiddleConnectionLost);
-                                    crate::error::ProxyError::MiddleConnectionLost
+                                let route_switched =
-                                );
+                                    matches!(&e, crate::error::ProxyError::RouteSwitched);
-                                let route_switched = matches!(
-                                    &e,
-                                    crate::error::ProxyError::Proxy(msg) if msg == ROUTE_SWITCH_ERROR_MSG
-                                );
 
                                 match (peer_close_reason, me_closed) {
                                     (Some(reason), _) => {

src/maestro/runtime_tasks.rs

@@ -75,7 +75,14 @@ pub(crate) async fn spawn_runtime_tasks(
 
     let stats_maintenance = stats.clone();
     tokio::spawn(async move {
-        stats_maintenance.run_periodic_user_stats_maintenance().await;
+        stats_maintenance
+            .run_periodic_user_stats_maintenance()
+            .await;
+    });
+
+    let ip_tracker_maintenance = ip_tracker.clone();
+    tokio::spawn(async move {
+        ip_tracker_maintenance.run_periodic_maintenance().await;
     });
 
     let detected_ip_v4: Option<IpAddr> = probe.detected_ipv4.map(IpAddr::V4);

src/maestro/shutdown.rs

@@ -8,8 +8,8 @@
 //!
 //! SIGHUP is handled separately in config/hot_reload.rs for config reload.
 
-use std::sync::Arc;
 use std::path::PathBuf;
+use std::sync::Arc;
 use std::time::{Duration, Instant};
 
 #[cfg(not(unix))]
@@ -52,7 +52,14 @@ pub(crate) async fn wait_for_shutdown(
     quota_state_path: PathBuf,
 ) {
     let signal = wait_for_shutdown_signal().await;
-    perform_shutdown(signal, process_started_at, me_pool, &stats, quota_state_path).await;
+    perform_shutdown(
+        signal,
+        process_started_at,
+        me_pool,
+        &stats,
+        quota_state_path,
+    )
+    .await;
 }
 
 /// Waits for any shutdown signal (SIGINT, SIGTERM, SIGQUIT).

src/metrics.rs

@@ -374,10 +374,7 @@ async fn render_tls_front_profile_health(
         out,
         "# HELP telemt_tls_front_profile_age_seconds Age of cached TLS front profile data per configured domain"
     );
-    let _ = writeln!(
+    let _ = writeln!(out, "# TYPE telemt_tls_front_profile_age_seconds gauge");
-        out,
-        "# TYPE telemt_tls_front_profile_age_seconds gauge"
-    );
     let _ = writeln!(
         out,
         "# HELP telemt_tls_front_profile_app_data_records TLS front cached app-data record count per configured domain"
@@ -390,10 +387,7 @@ async fn render_tls_front_profile_health(
         out,
         "# HELP telemt_tls_front_profile_ticket_records TLS front cached ticket-like tail record count per configured domain"
     );
-    let _ = writeln!(
+    let _ = writeln!(out, "# TYPE telemt_tls_front_profile_ticket_records gauge");
-        out,
-        "# TYPE telemt_tls_front_profile_ticket_records gauge"
-    );
     let _ = writeln!(
         out,
         "# HELP telemt_tls_front_profile_change_cipher_spec_records TLS front cached ChangeCipherSpec record count per configured domain"
@@ -406,21 +400,14 @@ async fn render_tls_front_profile_health(
         out,
         "# HELP telemt_tls_front_profile_app_data_bytes TLS front cached total app-data bytes per configured domain"
     );
-    let _ = writeln!(
+    let _ = writeln!(out, "# TYPE telemt_tls_front_profile_app_data_bytes gauge");
-        out,
-        "# TYPE telemt_tls_front_profile_app_data_bytes gauge"
-    );
 
     for item in health {
         let domain = prometheus_label_value(&item.domain);
         let _ = writeln!(
             out,
             "telemt_tls_front_profile_info{{domain=\"{}\",source=\"{}\",is_default=\"{}\",has_cert_info=\"{}\",has_cert_payload=\"{}\"}} 1",
-            domain,
+            domain, item.source, item.is_default, item.has_cert_info, item.has_cert_payload
-            item.source,
-            item.is_default,
-            item.has_cert_info,
-            item.has_cert_payload
         );
         let _ = writeln!(
             out,
@@ -771,10 +758,7 @@ async fn render_metrics(
         out,
         "# HELP telemt_quota_contention_timeout_total Quota reservations that hit the bounded contention budget"
     );
-    let _ = writeln!(
+    let _ = writeln!(out, "# TYPE telemt_quota_contention_timeout_total counter");
-        out,
-        "# TYPE telemt_quota_contention_timeout_total counter"
-    );
     let _ = writeln!(
         out,
         "telemt_quota_contention_timeout_total {}",
@@ -917,10 +901,7 @@ async fn render_metrics(
         out,
         "# HELP telemt_rate_limiter_burst_bound_bytes Configured upper bound for one direct relay rate-limit burst"
     );
-    let _ = writeln!(
+    let _ = writeln!(out, "# TYPE telemt_rate_limiter_burst_bound_bytes gauge");
-        out,
-        "# TYPE telemt_rate_limiter_burst_bound_bytes gauge"
-    );
     let _ = writeln!(
         out,
         "telemt_rate_limiter_burst_bound_bytes{{direction=\"up\"}} {}",
@@ -3793,8 +3774,9 @@ mod tests {
         )));
         assert!(output.contains("telemt_connections_total 2"));
         assert!(output.contains("telemt_connections_bad_total 1"));
-        assert!(output
+        assert!(output.contains(
-            .contains("telemt_connections_bad_by_class_total{class=\"tls_handshake_bad_client\"} 1"));
+            "telemt_connections_bad_by_class_total{class=\"tls_handshake_bad_client\"} 1"
+        ));
         assert!(output.contains("telemt_handshake_timeouts_total 1"));
         assert!(output.contains("telemt_handshake_failures_by_class_total{class=\"timeout\"} 1"));
         assert!(output.contains("telemt_auth_expensive_checks_total 9"));
@@ -3910,16 +3892,21 @@ mod tests {
             output.contains("telemt_tls_front_profile_info{domain=\"fallback.example\",source=\"default\",is_default=\"true\",has_cert_info=\"false\",has_cert_payload=\"false\"} 1")
         );
         assert!(
-            output.contains("telemt_tls_front_profile_app_data_records{domain=\"primary.example\"} 2")
+            output.contains(
+                "telemt_tls_front_profile_app_data_records{domain=\"primary.example\"} 2"
+            )
         );
         assert!(
-            output.contains("telemt_tls_front_profile_ticket_records{domain=\"primary.example\"} 1")
+            output
+                .contains("telemt_tls_front_profile_ticket_records{domain=\"primary.example\"} 1")
         );
+        assert!(output.contains(
+            "telemt_tls_front_profile_change_cipher_spec_records{domain=\"primary.example\"} 1"
+        ));
         assert!(
-            output.contains("telemt_tls_front_profile_change_cipher_spec_records{domain=\"primary.example\"} 1")
+            output.contains(
-        );
+                "telemt_tls_front_profile_app_data_bytes{domain=\"primary.example\"} 1536"
-        assert!(
+            )
-            output.contains("telemt_tls_front_profile_app_data_bytes{domain=\"primary.example\"} 1536")
         );
     }
 

src/proxy/client.rs

@@ -59,24 +59,30 @@ impl UserConnectionReservation {
         }
     }
 
-    async fn release(mut self) {
+    fn mark_released(&mut self) -> bool {
         if self.state != SessionReservationState::Active {
+            return false;
+        }
+        self.state = SessionReservationState::Released;
+        true
+    }
+
+    async fn release(mut self) {
+        if !self.mark_released() {
             return;
         }
         if self.tracks_ip {
             self.ip_tracker.remove_ip(&self.user, self.ip).await;
         }
-        self.state = SessionReservationState::Released;
         self.stats.decrement_user_curr_connects(&self.user);
     }
 }
 
 impl Drop for UserConnectionReservation {
     fn drop(&mut self) {
-        if self.state != SessionReservationState::Active {
+        if !self.mark_released() {
             return;
         }
-        self.state = SessionReservationState::Released;
         self.stats.increment_session_drop_fallback_total();
         self.stats.decrement_user_curr_connects(&self.user);
         if self.tracks_ip {
@@ -996,12 +1002,7 @@ impl RunningClientHandler {
                     trusted = ?self.config.server.proxy_protocol_trusted_cidrs,
                     "Rejecting PROXY protocol header from untrusted source"
                 );
-                record_beobachten_class(
+                record_beobachten_class(&self.beobachten, &self.config, self.peer.ip(), "other");
-                    &self.beobachten,
-                    &self.config,
-                    self.peer.ip(),
-                    "other",
-                );
                 return Err(ProxyError::InvalidProxyProtocol);
             }
 

src/proxy/direct_relay.rs

@@ -18,8 +18,7 @@ use crate::error::{ProxyError, Result};
 use crate::protocol::constants::*;
 use crate::proxy::handshake::{HandshakeSuccess, encrypt_tg_nonce_with_ciphers, generate_tg_nonce};
 use crate::proxy::route_mode::{
-    ROUTE_SWITCH_ERROR_MSG, RelayRouteMode, RouteCutoverState, affected_cutover_state,
+    RelayRouteMode, RouteCutoverState, affected_cutover_state, cutover_stagger_delay,
-    cutover_stagger_delay,
 };
 use crate::proxy::shared_state::{
     ConntrackCloseEvent, ConntrackClosePublishResult, ConntrackCloseReason, ProxySharedState,
@@ -360,7 +359,7 @@ where
                 "Cutover affected direct session, closing client connection"
             );
             tokio::time::sleep(delay).await;
-            break Err(ProxyError::Proxy(ROUTE_SWITCH_ERROR_MSG.to_string()));
+            break Err(ProxyError::RouteSwitched);
         }
         tokio::select! {
             result = &mut relay_result => {

src/proxy/handshake.rs

@@ -1901,7 +1901,10 @@ where
                 .auth_expensive_checks_total
                 .fetch_add(validation_checks as u64, Ordering::Relaxed);
 
-            if config.access.is_user_source_ip_denied(user.as_str(), peer.ip()) {
+            if config
+                .access
+                .is_user_source_ip_denied(user.as_str(), peer.ip())
+            {
                 auth_probe_record_failure_in(shared, peer.ip(), Instant::now());
                 maybe_apply_server_hello_delay(config).await;
                 warn!(

src/proxy/middle_relay.rs

@@ -23,8 +23,7 @@ use crate::error::{ProxyError, Result};
 use crate::protocol::constants::{secure_padding_len, *};
 use crate::proxy::handshake::HandshakeSuccess;
 use crate::proxy::route_mode::{
-    ROUTE_SWITCH_ERROR_MSG, RelayRouteMode, RouteCutoverState, affected_cutover_state,
+    RelayRouteMode, RouteCutoverState, affected_cutover_state, cutover_stagger_delay,
-    cutover_stagger_delay,
 };
 use crate::proxy::shared_state::{
     ConntrackCloseEvent, ConntrackClosePublishResult, ConntrackCloseReason, ProxySharedState,
@@ -73,6 +72,7 @@ enum MiddleQuotaReserveError {
     LimitExceeded,
     Contended,
     Cancelled,
+    DeadlineExceeded,
 }
 
 #[derive(Default)]
@@ -633,6 +633,7 @@ async fn reserve_user_quota_with_yield(
     limit: u64,
     stats: &Stats,
     cancel: &CancellationToken,
+    deadline: Option<Instant>,
 ) -> std::result::Result<u64, MiddleQuotaReserveError> {
     let mut backoff_ms = QUOTA_RESERVE_BACKOFF_MIN_MS;
     let mut backoff_rounds = 0usize;
@@ -651,6 +652,10 @@ async fn reserve_user_quota_with_yield(
         }
 
         tokio::task::yield_now().await;
+        if deadline.is_some_and(|deadline| Instant::now() >= deadline) {
+            stats.increment_quota_contention_timeout_total();
+            return Err(MiddleQuotaReserveError::DeadlineExceeded);
+        }
         tokio::select! {
             _ = tokio::time::sleep(Duration::from_millis(backoff_ms)) => {}
             _ = cancel.cancelled() => {
@@ -673,6 +678,7 @@ async fn wait_for_traffic_budget(
     lease: Option<&Arc<TrafficLease>>,
     direction: RateDirection,
     bytes: u64,
+    deadline: Option<Instant>,
 ) -> Result<()> {
     if bytes == 0 {
         return Ok(());
@@ -690,6 +696,9 @@ async fn wait_for_traffic_budget(
         }
 
         let wait_started_at = Instant::now();
+        if deadline.is_some_and(|deadline| wait_started_at >= deadline) {
+            return Err(ProxyError::TrafficBudgetWaitDeadlineExceeded);
+        }
         tokio::time::sleep(next_refill_delay()).await;
         let wait_ms = wait_started_at
             .elapsed()
@@ -712,6 +721,7 @@ async fn wait_for_traffic_budget_or_cancel(
     bytes: u64,
     cancel: &CancellationToken,
     stats: &Stats,
+    deadline: Option<Instant>,
 ) -> Result<()> {
     if bytes == 0 {
         return Ok(());
@@ -729,11 +739,15 @@ async fn wait_for_traffic_budget_or_cancel(
         }
 
         let wait_started_at = Instant::now();
+        if deadline.is_some_and(|deadline| wait_started_at >= deadline) {
+            stats.increment_flow_wait_middle_rate_limit_cancelled_total();
+            return Err(ProxyError::TrafficBudgetWaitDeadlineExceeded);
+        }
         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()));
+                return Err(ProxyError::TrafficBudgetWaitCancelled);
             }
         }
         let wait_ms = wait_started_at
@@ -1188,7 +1202,7 @@ where
         tokio::time::sleep(delay).await;
         let _ = me_pool.send_close(conn_id).await;
         me_pool.registry().unregister(conn_id).await;
-        return Err(ProxyError::Proxy(ROUTE_SWITCH_ERROR_MSG.to_string()));
+        return Err(ProxyError::RouteSwitched);
     }
 
     // Per-user ad_tag from access.user_ad_tags; fallback to general.ad_tag (hot-reloadable)
@@ -1353,7 +1367,7 @@ where
                             } else {
                                 None
                             };
-                            let _ = writer.flush().await;
+                            let _ = flush_client_or_cancel(&mut writer, &flow_cancel_me_writer).await;
                             let flush_duration_us = flush_started_at.map(|started| {
                                 started
                                     .elapsed()
@@ -1416,7 +1430,8 @@ where
                                     } else {
                                         None
                                     };
-                                let _ = writer.flush().await;
+                                let _ =
+                                    flush_client_or_cancel(&mut writer, &flow_cancel_me_writer).await;
                                 let flush_duration_us = flush_started_at.map(|started| {
                                     started
                                         .elapsed()
@@ -1484,7 +1499,11 @@ where
                                         } else {
                                             None
                                         };
-                                        let _ = writer.flush().await;
+                                        let _ = flush_client_or_cancel(
+                                            &mut writer,
+                                            &flow_cancel_me_writer,
+                                        )
+                                        .await;
                                         let flush_duration_us = flush_started_at.map(|started| {
                                             started
                                                 .elapsed()
@@ -1551,7 +1570,11 @@ where
                                             } else {
                                                 None
                                             };
-                                            let _ = writer.flush().await;
+                                            let _ = flush_client_or_cancel(
+                                                &mut writer,
+                                                &flow_cancel_me_writer,
+                                            )
+                                            .await;
                                             let flush_duration_us = flush_started_at.map(|started| {
                                                 started
                                                     .elapsed()
@@ -1597,7 +1620,7 @@ where
                     } else {
                         None
                     };
-                    writer.flush().await.map_err(ProxyError::Io)?;
+                    flush_client_or_cancel(&mut writer, &flow_cancel_me_writer).await?;
                     let flush_duration_us = flush_started_at.map(|started| {
                         started
                             .elapsed()
@@ -1690,7 +1713,7 @@ where
                 stats.as_ref(),
             )
             .await;
-            main_result = Err(ProxyError::Proxy(ROUTE_SWITCH_ERROR_MSG.to_string()));
+            main_result = Err(ProxyError::RouteSwitched);
             break;
         }
 
@@ -1721,6 +1744,7 @@ where
                             traffic_lease.as_ref(),
                             RateDirection::Up,
                             payload.len() as u64,
+                            None,
                         )
                         .await?;
                         forensics.bytes_c2me = forensics
@@ -1735,6 +1759,7 @@ where
                                 limit,
                                 stats.as_ref(),
                                 &flow_cancel,
+                                None,
                             )
                             .await
                             {
@@ -1757,6 +1782,12 @@ where
                                     ));
                                     break;
                                 }
+                                Err(MiddleQuotaReserveError::DeadlineExceeded) => {
+                                    main_result = Err(ProxyError::Proxy(
+                                        "ME C->ME quota reservation deadline exceeded".into(),
+                                    ));
+                                    break;
+                                }
                             }
                             stats.add_user_octets_from_handle(user_stats, payload.len() as u64);
                         } else {
@@ -1852,9 +1883,7 @@ where
     };
 
     // When client closes, but ME channel stopped as unregistered - it isnt error
-    if client_closed
+    if client_closed && matches!(writer_result, Err(ProxyError::MiddleConnectionLost)) {
-        && matches!(writer_result, Err(ProxyError::MiddleConnectionLost))
-    {
         writer_result = Ok(());
     }
 
@@ -2450,8 +2479,10 @@ 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);
-                match reserve_user_quota_with_yield(user_stats, data_len, soft_limit, stats, cancel)
+                match reserve_user_quota_with_yield(
-                    .await
+                    user_stats, data_len, soft_limit, stats, cancel, None,
+                )
+                .await
                 {
                     Ok(_) => {}
                     Err(MiddleQuotaReserveError::LimitExceeded) => {
@@ -2470,6 +2501,11 @@ where
                             "ME D->C quota reservation cancelled".into(),
                         ));
                     }
+                    Err(MiddleQuotaReserveError::DeadlineExceeded) => {
+                        return Err(ProxyError::Proxy(
+                            "ME D->C quota reservation deadline exceeded".into(),
+                        ));
+                    }
                 }
             }
             wait_for_traffic_budget_or_cancel(
@@ -2478,22 +2514,30 @@ where
                 data_len,
                 cancel,
                 stats,
+                None,
             )
             .await?;
 
-            let write_mode =
+            let write_mode = match write_client_payload(
-                match write_client_payload(client_writer, proto_tag, flags, &data, rng, frame_buf)
+                client_writer,
-                    .await
+                proto_tag,
-                {
+                flags,
-                    Ok(mode) => mode,
+                &data,
-                    Err(err) => {
+                rng,
-                        if quota_limit.is_some() {
+                frame_buf,
-                            stats.add_quota_write_fail_bytes_total(data_len);
+                cancel,
-                            stats.increment_quota_write_fail_events_total();
+            )
-                        }
+            .await
-                        return Err(err);
+            {
+                Ok(mode) => mode,
+                Err(err) => {
+                    if quota_limit.is_some() {
+                        stats.add_quota_write_fail_bytes_total(data_len);
+                        stats.increment_quota_write_fail_events_total();
                     }
-                };
+                    return Err(err);
+                }
+            };
 
             bytes_me2c.fetch_add(data_len, Ordering::Relaxed);
             if let Some(user_stats) = quota_user_stats {
@@ -2517,9 +2561,16 @@ where
             } else {
                 trace!(conn_id, confirm, "ME->C quickack");
             }
-            wait_for_traffic_budget_or_cancel(traffic_lease, RateDirection::Down, 4, cancel, stats)
+            wait_for_traffic_budget_or_cancel(
-                .await?;
+                traffic_lease,
-            write_client_ack(client_writer, proto_tag, confirm).await?;
+                RateDirection::Down,
+                4,
+                cancel,
+                stats,
+                None,
+            )
+            .await?;
+            write_client_ack(client_writer, proto_tag, confirm, cancel).await?;
             stats.increment_me_d2c_ack_frames_total();
 
             Ok(MeWriterResponseOutcome::Continue {
@@ -2571,6 +2622,7 @@ async fn write_client_payload<W>(
     data: &[u8],
     rng: &SecureRandom,
     frame_buf: &mut Vec<u8>,
+    cancel: &CancellationToken,
 ) -> Result<MeD2cWriteMode>
 where
     W: AsyncWrite + Unpin + Send + 'static,
@@ -2598,21 +2650,12 @@ where
                     frame_buf.reserve(wire_len);
                     frame_buf.push(first);
                     frame_buf.extend_from_slice(data);
-                    client_writer
+                    write_all_client_or_cancel(client_writer, frame_buf.as_slice(), cancel).await?;
-                        .write_all(frame_buf.as_slice())
-                        .await
-                        .map_err(ProxyError::Io)?;
                     MeD2cWriteMode::Coalesced
                 } else {
                     let header = [first];
-                    client_writer
+                    write_all_client_or_cancel(client_writer, &header, cancel).await?;
-                        .write_all(&header)
+                    write_all_client_or_cancel(client_writer, data, cancel).await?;
-                        .await
-                        .map_err(ProxyError::Io)?;
-                    client_writer
-                        .write_all(data)
-                        .await
-                        .map_err(ProxyError::Io)?;
                     MeD2cWriteMode::Split
                 }
             } else if len_words < (1 << 24) {
@@ -2627,21 +2670,12 @@ where
                     frame_buf.reserve(wire_len);
                     frame_buf.extend_from_slice(&[first, lw[0], lw[1], lw[2]]);
                     frame_buf.extend_from_slice(data);
-                    client_writer
+                    write_all_client_or_cancel(client_writer, frame_buf.as_slice(), cancel).await?;
-                        .write_all(frame_buf.as_slice())
-                        .await
-                        .map_err(ProxyError::Io)?;
                     MeD2cWriteMode::Coalesced
                 } else {
                     let header = [first, lw[0], lw[1], lw[2]];
-                    client_writer
+                    write_all_client_or_cancel(client_writer, &header, cancel).await?;
-                        .write_all(&header)
+                    write_all_client_or_cancel(client_writer, data, cancel).await?;
-                        .await
-                        .map_err(ProxyError::Io)?;
-                    client_writer
-                        .write_all(data)
-                        .await
-                        .map_err(ProxyError::Io)?;
                     MeD2cWriteMode::Split
                 }
             } else {
@@ -2676,21 +2710,12 @@ where
                     frame_buf.resize(start + padding_len, 0);
                     rng.fill(&mut frame_buf[start..]);
                 }
-                client_writer
+                write_all_client_or_cancel(client_writer, frame_buf.as_slice(), cancel).await?;
-                    .write_all(frame_buf.as_slice())
-                    .await
-                    .map_err(ProxyError::Io)?;
                 MeD2cWriteMode::Coalesced
             } else {
                 let header = len_val.to_le_bytes();
-                client_writer
+                write_all_client_or_cancel(client_writer, &header, cancel).await?;
-                    .write_all(&header)
+                write_all_client_or_cancel(client_writer, data, cancel).await?;
-                    .await
-                    .map_err(ProxyError::Io)?;
-                client_writer
-                    .write_all(data)
-                    .await
-                    .map_err(ProxyError::Io)?;
                 if padding_len > 0 {
                     frame_buf.clear();
                     if frame_buf.capacity() < padding_len {
@@ -2698,10 +2723,7 @@ where
                     }
                     frame_buf.resize(padding_len, 0);
                     rng.fill(frame_buf.as_mut_slice());
-                    client_writer
+                    write_all_client_or_cancel(client_writer, frame_buf.as_slice(), cancel).await?;
-                        .write_all(frame_buf.as_slice())
-                        .await
-                        .map_err(ProxyError::Io)?;
                 }
                 MeD2cWriteMode::Split
             }
@@ -2715,6 +2737,7 @@ async fn write_client_ack<W>(
     client_writer: &mut CryptoWriter<W>,
     proto_tag: ProtoTag,
     confirm: u32,
+    cancel: &CancellationToken,
 ) -> Result<()>
 where
     W: AsyncWrite + Unpin + Send + 'static,
@@ -2724,10 +2747,34 @@ where
     } else {
         confirm.to_le_bytes()
     };
-    client_writer
+    write_all_client_or_cancel(client_writer, &bytes, cancel).await
-        .write_all(&bytes)
+}
-        .await
+
-        .map_err(ProxyError::Io)
+async fn write_all_client_or_cancel<W>(
+    client_writer: &mut CryptoWriter<W>,
+    bytes: &[u8],
+    cancel: &CancellationToken,
+) -> Result<()>
+where
+    W: AsyncWrite + Unpin + Send + 'static,
+{
+    tokio::select! {
+        result = client_writer.write_all(bytes) => result.map_err(ProxyError::Io),
+        _ = cancel.cancelled() => Err(ProxyError::MiddleClientWriterCancelled),
+    }
+}
+
+async fn flush_client_or_cancel<W>(
+    client_writer: &mut CryptoWriter<W>,
+    cancel: &CancellationToken,
+) -> Result<()>
+where
+    W: AsyncWrite + Unpin + Send + 'static,
+{
+    tokio::select! {
+        result = client_writer.flush() => result.map_err(ProxyError::Io),
+        _ = cancel.cancelled() => Err(ProxyError::MiddleClientWriterCancelled),
+    }
 }
 
 #[cfg(test)]

src/proxy/relay.rs

@@ -518,10 +518,7 @@ impl<S: AsyncRead + Unpin> AsyncRead for StatsIo<S> {
             Poll::Ready(Ok(n)) => {
                 if reserved_read_bytes > n as u64 {
                     let refund_bytes = reserved_read_bytes - n as u64;
-                    refund_reserved_quota_bytes(
+                    refund_reserved_quota_bytes(this.user_stats.as_ref(), refund_bytes);
-                        this.user_stats.as_ref(),
-                        refund_bytes,
-                    );
                     this.stats.add_quota_refund_bytes_total(refund_bytes);
                 }
                 if n > 0 {
@@ -538,7 +535,6 @@ impl<S: AsyncRead + Unpin> AsyncRead for StatsIo<S> {
                                 this.quota_bytes_since_check = 0;
                             }
                         }
-
                     }
                     if let Some(limit) = this.quota_limit
                         && this.user_stats.quota_used() >= limit
@@ -700,10 +696,7 @@ impl<S: AsyncWrite + Unpin> AsyncWrite for StatsIo<S> {
             Poll::Ready(Ok(n)) => {
                 if reserved_bytes > n as u64 {
                     let refund_bytes = reserved_bytes - n as u64;
-                    refund_reserved_quota_bytes(
+                    refund_reserved_quota_bytes(this.user_stats.as_ref(), refund_bytes);
-                        this.user_stats.as_ref(),
-                        refund_bytes,
-                    );
                     this.stats.add_quota_refund_bytes_total(refund_bytes);
                 }
                 if shaper_reserved_bytes > n as u64

src/proxy/route_mode.rs

@@ -4,8 +4,6 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 use tokio::sync::watch;
 
-pub(crate) const ROUTE_SWITCH_ERROR_MSG: &str = "Session terminated";
-
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 #[repr(u8)]
 pub(crate) enum RelayRouteMode {

src/proxy/tests/client_security_tests.rs

@@ -661,7 +661,7 @@ async fn integration_route_cutover_and_quota_overlap_fails_closed_and_releases_s
 
     assert!(
         matches!(relay_result, Err(ProxyError::DataQuotaExceeded { .. }))
-            || matches!(relay_result, Err(ProxyError::Proxy(ref msg)) if msg == crate::proxy::route_mode::ROUTE_SWITCH_ERROR_MSG),
+            || matches!(relay_result, Err(ProxyError::RouteSwitched)),
         "overlap race must fail closed via quota enforcement or generic cutover termination"
     );
 

src/proxy/tests/direct_relay_security_tests.rs

@@ -637,6 +637,22 @@ fn unknown_dc_log_path_revalidation_rejects_parent_swapped_to_symlink() {
             "telemt-unknown-dc-parent-swap-{}",
             std::process::id()
         ));
+    if let Ok(meta) = fs::symlink_metadata(&parent) {
+        if meta.file_type().is_symlink() || meta.is_file() {
+            fs::remove_file(&parent).expect("stale parent-swap path must be removable");
+        } else {
+            fs::remove_dir_all(&parent).expect("stale parent-swap directory must be removable");
+        }
+    }
+    let moved = parent.with_extension("bak");
+    if let Ok(meta) = fs::symlink_metadata(&moved) {
+        if meta.file_type().is_symlink() || meta.is_file() {
+            fs::remove_file(&moved).expect("stale parent-swap backup path must be removable");
+        } else {
+            fs::remove_dir_all(&moved)
+                .expect("stale parent-swap backup directory must be removable");
+        }
+    }
     fs::create_dir_all(&parent).expect("parent-swap test parent must be creatable");
 
     let rel_candidate = format!(
@@ -646,8 +662,6 @@ fn unknown_dc_log_path_revalidation_rejects_parent_swapped_to_symlink() {
     let sanitized = sanitize_unknown_dc_log_path(&rel_candidate)
         .expect("candidate must sanitize before parent swap");
 
-    let moved = parent.with_extension("bak");
-    let _ = fs::remove_dir_all(&moved);
     fs::rename(&parent, &moved).expect("parent must be movable for swap simulation");
     symlink("/tmp", &parent).expect("symlink replacement for parent must be creatable");
 
@@ -720,6 +734,24 @@ fn adversarial_parent_swap_after_check_is_blocked_by_anchored_open() {
             "telemt-unknown-dc-parent-swap-openat-{}",
             std::process::id()
         ));
+    if let Ok(meta) = fs::symlink_metadata(&base) {
+        if meta.file_type().is_symlink() || meta.is_file() {
+            fs::remove_file(&base).expect("stale parent-swap-openat path must be removable");
+        } else {
+            fs::remove_dir_all(&base)
+                .expect("stale parent-swap-openat directory must be removable");
+        }
+    }
+    let moved = base.with_extension("bak");
+    if let Ok(meta) = fs::symlink_metadata(&moved) {
+        if meta.file_type().is_symlink() || meta.is_file() {
+            fs::remove_file(&moved)
+                .expect("stale parent-swap-openat backup path must be removable");
+        } else {
+            fs::remove_dir_all(&moved)
+                .expect("stale parent-swap-openat backup directory must be removable");
+        }
+    }
     fs::create_dir_all(&base).expect("parent-swap-openat base must be creatable");
 
     let rel_candidate = format!(
@@ -743,8 +775,6 @@ fn adversarial_parent_swap_after_check_is_blocked_by_anchored_open() {
     let outside_target = outside_parent.join("unknown-dc.log");
     let _ = fs::remove_file(&outside_target);
 
-    let moved = base.with_extension("bak");
-    let _ = fs::remove_dir_all(&moved);
     fs::rename(&base, &moved).expect("base parent must be movable for swap simulation");
     symlink(&outside_parent, &base).expect("base parent symlink replacement must be creatable");
 
@@ -1489,10 +1519,7 @@ async fn direct_relay_cutover_midflight_releases_route_gauge() {
         "cutover should terminate direct relay session"
     );
     assert!(
-        matches!(
+        matches!(relay_result, Err(ProxyError::RouteSwitched)),
-            relay_result,
-            Err(ProxyError::Proxy(ref msg)) if msg == ROUTE_SWITCH_ERROR_MSG
-        ),
         "client-visible cutover error must stay generic and avoid route-internal metadata"
     );
 
@@ -1629,10 +1656,7 @@ async fn direct_relay_cutover_storm_multi_session_keeps_generic_errors_and_relea
             .expect("direct relay task must not panic");
 
         assert!(
-            matches!(
+            matches!(relay_result, Err(ProxyError::RouteSwitched)),
-                relay_result,
-                Err(ProxyError::Proxy(ref msg)) if msg == ROUTE_SWITCH_ERROR_MSG
-            ),
             "storm-cutover termination must remain generic for all direct sessions"
         );
     }
@@ -1935,10 +1959,7 @@ async fn adversarial_direct_relay_cutover_integrity() {
         .expect("Session must not panic");
 
     assert!(
-        matches!(
+        matches!(result, Err(ProxyError::RouteSwitched)),
-            result,
-            Err(ProxyError::Proxy(ref msg)) if msg == ROUTE_SWITCH_ERROR_MSG
-        ),
         "Session must terminate with route switch error on cutover"
     );
 }

src/proxy/tests/middle_relay_atomic_quota_invariant_tests.rs

@@ -13,6 +13,8 @@ struct CountedWriter {
     fail_writes: bool,
 }
 
+struct StalledWriter;
+
 impl CountedWriter {
     fn new(write_calls: Arc<AtomicUsize>, fail_writes: bool) -> Self {
         Self {
@@ -49,12 +51,36 @@ impl AsyncWrite for CountedWriter {
     }
 }
 
+impl AsyncWrite for StalledWriter {
+    fn poll_write(
+        self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+        _buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        Poll::Pending
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Poll::Pending
+    }
+
+    fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Poll::Pending
+    }
+}
+
 fn make_crypto_writer(inner: CountedWriter) -> CryptoWriter<CountedWriter> {
     let key = [0u8; 32];
     let iv = 0u128;
     CryptoWriter::new(inner, AesCtr::new(&key, iv), 8 * 1024)
 }
 
+fn make_stalled_crypto_writer() -> CryptoWriter<StalledWriter> {
+    let key = [0u8; 32];
+    let iv = 0u128;
+    CryptoWriter::new(StalledWriter, AesCtr::new(&key, iv), 8 * 1024)
+}
+
 #[tokio::test]
 async fn me_writer_write_fail_keeps_reserved_quota_and_tracks_fail_metrics() {
     let stats = Stats::new();
@@ -189,3 +215,53 @@ async fn me_writer_pre_write_quota_reject_happens_before_writer_poll() {
     );
     assert_eq!(bytes_me2c.load(Ordering::Relaxed), 0);
 }
+
+#[tokio::test]
+async fn me_writer_data_write_obeys_flow_cancellation() {
+    let stats = Stats::new();
+    let user = "middle-me-writer-cancel-user";
+    let mut writer = make_stalled_crypto_writer();
+    let mut frame_buf = Vec::new();
+    let bytes_me2c = AtomicU64::new(0);
+    let cancel = CancellationToken::new();
+    cancel.cancel();
+
+    let result = process_me_writer_response_with_traffic_lease(
+        MeResponse::Data {
+            flags: 0,
+            data: Bytes::from_static(&[0x31, 0x32, 0x33, 0x34]),
+            route_permit: None,
+        },
+        &mut writer,
+        ProtoTag::Intermediate,
+        &SecureRandom::new(),
+        &mut frame_buf,
+        &stats,
+        user,
+        None,
+        None,
+        0,
+        None,
+        &cancel,
+        &bytes_me2c,
+        13,
+        true,
+        false,
+    )
+    .await;
+
+    assert!(
+        matches!(result, Err(ProxyError::MiddleClientWriterCancelled)),
+        "cancelled middle writer must return a bounded cancellation error"
+    );
+    assert_eq!(
+        bytes_me2c.load(Ordering::Relaxed),
+        0,
+        "cancelled write must not advance committed ME->C bytes"
+    );
+    assert_eq!(
+        stats.get_user_total_octets(user),
+        0,
+        "cancelled write must not advance user output telemetry"
+    );
+}

src/proxy/tests/relay_atomic_quota_invariant_tests.rs

@@ -4,10 +4,67 @@ use std::io;
 use std::pin::Pin;
 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::sync::{Arc, Mutex};
-use std::task::{Context, Poll};
+use std::task::{Context, Poll, Wake};
-use tokio::io::{AsyncWrite, AsyncWriteExt};
+use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt, ReadBuf};
 use tokio::time::Instant;
 
+enum ReadStep {
+    Data(Vec<u8>),
+    Pending,
+    Eof,
+    Error,
+}
+
+struct ScriptedReader {
+    scripted_reads: Arc<Mutex<VecDeque<ReadStep>>>,
+    read_calls: Arc<AtomicUsize>,
+}
+
+impl ScriptedReader {
+    fn new(script: Vec<ReadStep>, read_calls: Arc<AtomicUsize>) -> Self {
+        Self {
+            scripted_reads: Arc::new(Mutex::new(script.into())),
+            read_calls,
+        }
+    }
+}
+
+impl AsyncRead for ScriptedReader {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
+        let this = self.get_mut();
+        this.read_calls.fetch_add(1, Ordering::Relaxed);
+        let step = this
+            .scripted_reads
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+            .pop_front()
+            .unwrap_or(ReadStep::Eof);
+        match step {
+            ReadStep::Data(data) => {
+                let n = data.len().min(buf.remaining());
+                buf.put_slice(&data[..n]);
+                Poll::Ready(Ok(()))
+            }
+            ReadStep::Pending => Poll::Pending,
+            ReadStep::Eof => Poll::Ready(Ok(())),
+            ReadStep::Error => Poll::Ready(Err(io::Error::new(
+                io::ErrorKind::BrokenPipe,
+                "forced read failure",
+            ))),
+        }
+    }
+}
+
+struct NoopWake;
+
+impl Wake for NoopWake {
+    fn wake(self: Arc<Self>) {}
+}
+
 struct ScriptedWriter {
     scripted_writes: Arc<Mutex<VecDeque<usize>>>,
     write_calls: Arc<AtomicUsize>,
@@ -80,6 +137,127 @@ fn make_stats_io_with_script(
     (io, stats, write_calls, quota_exceeded)
 }
 
+fn make_stats_io_with_read_script(
+    user: &str,
+    quota_limit: u64,
+    precharged_quota: u64,
+    script: Vec<ReadStep>,
+) -> (
+    StatsIo<ScriptedReader>,
+    Arc<Stats>,
+    Arc<AtomicUsize>,
+    Arc<AtomicBool>,
+) {
+    let stats = Arc::new(Stats::new());
+    if precharged_quota > 0 {
+        let user_stats = stats.get_or_create_user_stats_handle(user);
+        stats.quota_charge_post_write(user_stats.as_ref(), precharged_quota);
+    }
+
+    let read_calls = Arc::new(AtomicUsize::new(0));
+    let quota_exceeded = Arc::new(AtomicBool::new(false));
+    let io = StatsIo::new(
+        ScriptedReader::new(script, read_calls.clone()),
+        Arc::new(SharedCounters::new()),
+        stats.clone(),
+        user.to_string(),
+        Some(quota_limit),
+        quota_exceeded.clone(),
+        Instant::now(),
+    );
+
+    (io, stats, read_calls, quota_exceeded)
+}
+
+fn poll_read_once<R: AsyncRead + Unpin>(
+    io: &mut StatsIo<R>,
+    storage: &mut [u8],
+) -> Poll<io::Result<usize>> {
+    let waker = Arc::new(NoopWake).into();
+    let mut cx = Context::from_waker(&waker);
+    let mut read_buf = ReadBuf::new(storage);
+    let before = read_buf.filled().len();
+    match Pin::new(io).poll_read(&mut cx, &mut read_buf) {
+        Poll::Ready(Ok(())) => Poll::Ready(Ok(read_buf.filled().len() - before)),
+        Poll::Ready(Err(error)) => Poll::Ready(Err(error)),
+        Poll::Pending => Poll::Pending,
+    }
+}
+
+#[test]
+fn direct_c2s_quota_refunds_unused_on_short_read() {
+    let user = "direct-c2s-short-read-refund-user";
+    let (mut io, stats, read_calls, quota_exceeded) =
+        make_stats_io_with_read_script(user, 64, 0, vec![ReadStep::Data(vec![0x11; 5])]);
+    let mut storage = [0u8; 16];
+
+    let n = match poll_read_once(&mut io, &mut storage) {
+        Poll::Ready(Ok(n)) => n,
+        other => panic!("short read must complete, got {other:?}"),
+    };
+
+    assert_eq!(n, 5);
+    assert_eq!(read_calls.load(Ordering::Relaxed), 1);
+    assert_eq!(stats.get_user_quota_used(user), 5);
+    assert_eq!(stats.get_quota_refund_bytes_total(), 11);
+    assert!(!quota_exceeded.load(Ordering::Acquire));
+}
+
+#[test]
+fn direct_c2s_quota_refunds_full_reservation_on_pending() {
+    let user = "direct-c2s-pending-refund-user";
+    let (mut io, stats, read_calls, quota_exceeded) =
+        make_stats_io_with_read_script(user, 64, 0, vec![ReadStep::Pending]);
+    let mut storage = [0u8; 16];
+
+    assert!(matches!(
+        poll_read_once(&mut io, &mut storage),
+        Poll::Pending
+    ));
+    assert_eq!(read_calls.load(Ordering::Relaxed), 1);
+    assert_eq!(stats.get_user_quota_used(user), 0);
+    assert_eq!(stats.get_quota_refund_bytes_total(), 16);
+    assert!(!quota_exceeded.load(Ordering::Acquire));
+}
+
+#[test]
+fn direct_c2s_quota_refunds_full_reservation_on_eof() {
+    let user = "direct-c2s-eof-refund-user";
+    let (mut io, stats, read_calls, quota_exceeded) =
+        make_stats_io_with_read_script(user, 64, 0, vec![ReadStep::Eof]);
+    let mut storage = [0u8; 16];
+
+    let n = match poll_read_once(&mut io, &mut storage) {
+        Poll::Ready(Ok(n)) => n,
+        other => panic!("EOF read must complete with zero bytes, got {other:?}"),
+    };
+
+    assert_eq!(n, 0);
+    assert_eq!(read_calls.load(Ordering::Relaxed), 1);
+    assert_eq!(stats.get_user_quota_used(user), 0);
+    assert_eq!(stats.get_quota_refund_bytes_total(), 16);
+    assert!(!quota_exceeded.load(Ordering::Acquire));
+}
+
+#[test]
+fn direct_c2s_quota_refunds_full_reservation_on_error() {
+    let user = "direct-c2s-error-refund-user";
+    let (mut io, stats, read_calls, quota_exceeded) =
+        make_stats_io_with_read_script(user, 64, 0, vec![ReadStep::Error]);
+    let mut storage = [0u8; 16];
+
+    let error = match poll_read_once(&mut io, &mut storage) {
+        Poll::Ready(Err(error)) => error,
+        other => panic!("error read must return error, got {other:?}"),
+    };
+
+    assert_eq!(error.kind(), io::ErrorKind::BrokenPipe);
+    assert_eq!(read_calls.load(Ordering::Relaxed), 1);
+    assert_eq!(stats.get_user_quota_used(user), 0);
+    assert_eq!(stats.get_quota_refund_bytes_total(), 16);
+    assert!(!quota_exceeded.load(Ordering::Acquire));
+}
+
 #[tokio::test]
 async fn direct_partial_write_charges_only_committed_bytes_without_double_charge() {
     let user = "direct-partial-charge-user";

src/stats/mod.rs

@@ -8,8 +8,8 @@ pub mod telemetry;
 use dashmap::DashMap;
 use lru::LruCache;
 use parking_lot::Mutex;
-use std::collections::{HashMap, VecDeque};
 use std::collections::hash_map::DefaultHasher;
+use std::collections::{HashMap, VecDeque};
 use std::hash::{Hash, Hasher};
 use std::num::NonZeroUsize;
 use std::sync::Arc;
@@ -2378,8 +2378,7 @@ impl Stats {
         self.quota_contention_total.load(Ordering::Relaxed)
     }
     pub fn get_quota_contention_timeout_total(&self) -> u64 {
-        self.quota_contention_timeout_total
+        self.quota_contention_timeout_total.load(Ordering::Relaxed)
-            .load(Ordering::Relaxed)
     }
     pub fn get_quota_acquire_cancelled_total(&self) -> u64 {
         self.quota_acquire_cancelled_total.load(Ordering::Relaxed)
@@ -2536,12 +2535,7 @@ impl Stats {
             .unwrap_or(0)
     }
 
-    pub fn load_user_quota_state(
+    pub fn load_user_quota_state(&self, user: &str, used_bytes: u64, last_reset_epoch_secs: u64) {
-        &self,
-        user: &str,
-        used_bytes: u64,
-        last_reset_epoch_secs: u64,
-    ) {
         let stats = self.get_or_create_user_stats_handle(user);
         stats.quota_used.store(used_bytes, Ordering::Relaxed);
         stats

src/tls_front/cache.rs

@@ -150,9 +150,10 @@ impl TlsFrontCache {
                 is_default: cached.domain == "default",
                 has_cert_info: cached.cert_info.is_some(),
                 has_cert_payload: cached.cert_payload.is_some(),
-                app_data_records: cached.app_data_records_sizes.len().max(
+                app_data_records: cached
-                    behavior.app_data_record_sizes.len(),
+                    .app_data_records_sizes
-                ),
+                    .len()
+                    .max(behavior.app_data_record_sizes.len()),
                 ticket_records: behavior.ticket_record_sizes.len(),
                 change_cipher_spec_count: behavior.change_cipher_spec_count,
                 total_app_data_len: cached.total_app_data_len,