Update API.md

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]

docs/Architecture/API/API.md

@@ -13,7 +13,7 @@ API runtime is configured in `[server.api]`.
 | `listen` | `string` (`IP:PORT`) | `0.0.0.0:9091` | API bind address. |
 | `whitelist` | `CIDR[]` | `127.0.0.0/8` | Source IP allowlist. Empty list means allow all. |
 | `auth_header` | `string` | `""` | Exact value for `Authorization` header. Empty disables header auth. |
-| `request_body_limit_bytes` | `usize` | `65536` | Maximum request body size. Must be `> 0`. |
+| `request_body_limit_bytes` | `usize` | `65536` | Maximum request body size. Must be within `[1, 1048576]`. |
 | `minimal_runtime_enabled` | `bool` | `true` | Enables runtime snapshot endpoints requiring ME pool read-lock aggregation. |
 | `minimal_runtime_cache_ttl_ms` | `u64` | `1000` | Cache TTL for minimal snapshots. `0` disables cache; valid range is `[0, 60000]`. |
 | `runtime_edge_enabled` | `bool` | `false` | Enables runtime edge endpoints with cached aggregation payloads. |
@@ -26,7 +26,7 @@ API runtime is configured in `[server.api]`.
 
 Runtime validation for API config:
 - `server.api.listen` must be a valid `IP:PORT`.
-- `server.api.request_body_limit_bytes` must be `> 0`.
+- `server.api.request_body_limit_bytes` must be within `[1, 1048576]`.
 - `server.api.minimal_runtime_cache_ttl_ms` must be within `[0, 60000]`.
 - `server.api.runtime_edge_cache_ttl_ms` must be within `[0, 60000]`.
 - `server.api.runtime_edge_top_n` must be within `[1, 1000]`.
@@ -76,13 +76,14 @@ Requests are processed in this order:
 
 Notes:
 - Whitelist is evaluated against the direct TCP peer IP (`SocketAddr::ip`), without `X-Forwarded-For` support.
-- `Authorization` check is exact string equality against configured `auth_header`.
+- `Authorization` check is exact constant-time byte equality against configured `auth_header`.
 
 ## Endpoint Matrix
 
 | Method | Path | Body | Success | `data` contract |
 | --- | --- | --- | --- | --- |
 | `GET` | `/v1/health` | none | `200` | `HealthData` |
+| `GET` | `/v1/health/ready` | none | `200` or `503` | `HealthReadyData` |
 | `GET` | `/v1/system/info` | none | `200` | `SystemInfoData` |
 | `GET` | `/v1/runtime/gates` | none | `200` | `RuntimeGatesData` |
 | `GET` | `/v1/runtime/initialization` | none | `200` | `RuntimeInitializationData` |
@@ -102,13 +103,50 @@ Notes:
 | `GET` | `/v1/runtime/me-selftest` | none | `200` | `RuntimeMeSelftestData` |
 | `GET` | `/v1/runtime/connections/summary` | none | `200` | `RuntimeEdgeConnectionsSummaryData` |
 | `GET` | `/v1/runtime/events/recent` | none | `200` | `RuntimeEdgeEventsData` |
+| `GET` | `/v1/stats/users/active-ips` | none | `200` | `UserActiveIps[]` |
 | `GET` | `/v1/stats/users` | none | `200` | `UserInfo[]` |
 | `GET` | `/v1/users` | none | `200` | `UserInfo[]` |
-| `POST` | `/v1/users` | `CreateUserRequest` | `201` | `CreateUserResponse` |
+| `POST` | `/v1/users` | `CreateUserRequest` | `201` or `202` | `CreateUserResponse` |
 | `GET` | `/v1/users/{username}` | none | `200` | `UserInfo` |
-| `PATCH` | `/v1/users/{username}` | `PatchUserRequest` | `200` | `UserInfo` |
-| `DELETE` | `/v1/users/{username}` | none | `200` | `string` (deleted username) |
-| `POST` | `/v1/users/{username}/rotate-secret` | `RotateSecretRequest` or empty body | `404` | `ErrorResponse` (`not_found`, current runtime behavior) |
+| `PATCH` | `/v1/users/{username}` | `PatchUserRequest` | `200` or `202` | `UserInfo` |
+| `DELETE` | `/v1/users/{username}` | none | `200` or `202` | `DeleteUserResponse` |
+| `POST` | `/v1/users/{username}/rotate-secret` | `RotateSecretRequest` or empty body | `200` or `202` | `CreateUserResponse` |
+| `POST` | `/v1/users/{username}/reset-quota` | empty body | `200` | `ResetUserQuotaResponse` |
+
+## Endpoint Behavior
+
+| Endpoint | Function |
+| --- | --- |
+| `GET /v1/health` | Returns basic API liveness and current `read_only` flag. |
+| `GET /v1/health/ready` | Returns readiness based on admission state and upstream health; returns `503` when not ready. |
+| `GET /v1/system/info` | Returns binary/build metadata, process uptime, config path/hash, and reload counters. |
+| `GET /v1/runtime/gates` | Returns admission, ME readiness, fallback/reroute, and startup gate state. |
+| `GET /v1/runtime/initialization` | Returns startup progress, ME initialization status, and per-component timeline. |
+| `GET /v1/limits/effective` | Returns effective timeout, upstream, ME, unique-IP, and TCP policy values after config defaults/resolution. |
+| `GET /v1/security/posture` | Returns current API/security/telemetry posture flags. |
+| `GET /v1/security/whitelist` | Returns configured API whitelist CIDRs. |
+| `GET /v1/stats/summary` | Returns compact core counters and classed failure counters. |
+| `GET /v1/stats/zero/all` | Returns zero-cost core, upstream, ME, pool, and desync counters. |
+| `GET /v1/stats/upstreams` | Returns upstream zero counters and, when enabled/available, runtime upstream health rows. |
+| `GET /v1/stats/minimal/all` | Returns cached minimal ME writer/DC/runtime/network-path snapshot. |
+| `GET /v1/stats/me-writers` | Returns cached ME writer coverage and per-writer status rows. |
+| `GET /v1/stats/dcs` | Returns cached per-DC endpoint/writer/load status rows. |
+| `GET /v1/runtime/me_pool_state` | Returns active/warm/pending/draining generation state, writer contour/health, and refill state. |
+| `GET /v1/runtime/me_quality` | Returns ME lifecycle counters, route-drop counters, family states, drain gate, and per-DC RTT/coverage. |
+| `GET /v1/runtime/upstream_quality` | Returns upstream policy/counters plus runtime upstream health rows when available. |
+| `GET /v1/runtime/nat_stun` | Returns NAT/STUN runtime flags, configured/live STUN servers, reflection cache, and backoff. |
+| `GET /v1/runtime/me-selftest` | Returns ME self-test state for KDF, time skew, IP family, PID, and SOCKS BND observations. |
+| `GET /v1/runtime/connections/summary` | Returns runtime-edge connection totals and top-N users by connections/throughput. |
+| `GET /v1/runtime/events/recent` | Returns recent API/runtime event records with optional `limit` query. |
+| `GET /v1/stats/users/active-ips` | Returns users that currently have non-empty active source-IP lists. |
+| `GET /v1/stats/users` | Alias of `GET /v1/users`; returns disk-first user views with runtime lag flag. |
+| `GET /v1/users` | Returns disk-first user views sorted by username. |
+| `POST /v1/users` | Creates a user and returns the effective user view plus secret. |
+| `GET /v1/users/{username}` | Returns one disk-first user view or `404` when absent. |
+| `PATCH /v1/users/{username}` | Updates selected per-user fields with JSON Merge Patch semantics. |
+| `DELETE /v1/users/{username}` | Deletes one user and related per-user access-map entries. |
+| `POST /v1/users/{username}/rotate-secret` | Rotates one user's secret and returns the effective secret. |
+| `POST /v1/users/{username}/reset-quota` | Resets one user's runtime quota counter and persists quota state. |
 
 ## Common Error Codes
 
@@ -118,7 +156,7 @@ Notes:
 | `401` | `unauthorized` | Missing/invalid `Authorization` when `auth_header` is configured. |
 | `403` | `forbidden` | Source IP is not allowed by whitelist. |
 | `403` | `read_only` | Mutating endpoint called while `read_only=true`. |
-| `404` | `not_found` | Unknown route, unknown user, or unsupported sub-route (including current `rotate-secret` route). |
+| `404` | `not_found` | Unknown route, unknown user, or unsupported sub-route. |
 | `405` | `method_not_allowed` | Unsupported method for `/v1/users/{username}` route shape. |
 | `409` | `revision_conflict` | `If-Match` revision mismatch. |
 | `409` | `user_exists` | User already exists on create. |
@@ -132,11 +170,12 @@ Notes:
 | Case | Behavior |
 | --- | --- |
 | Path matching | Exact match on `req.uri().path()`. Query string does not affect route matching. |
-| Trailing slash | Not normalized. Example: `/v1/users/` is `404`. |
+| Trailing slash | Trimmed for route matching when path length is greater than 1. Example: `/v1/users/` matches `/v1/users`. |
 | Username route with extra slash | `/v1/users/{username}/...` is not treated as user route and returns `404`. |
 | `PUT /v1/users/{username}` | `405 method_not_allowed`. |
 | `POST /v1/users/{username}` | `404 not_found`. |
-| `POST /v1/users/{username}/rotate-secret` | `404 not_found` in current release due route matcher limitation. |
+| `POST /v1/users/{username}/rotate-secret/` | Trailing slash is trimmed and the route matches `rotate-secret`. |
+| `POST /v1/users/{username}/reset-quota/` | Trailing slash is trimmed and the route matches `reset-quota`. |
 
 ## Body and JSON Semantics
 
@@ -146,7 +185,7 @@ Notes:
 - Invalid JSON returns `400 bad_request` (`Invalid JSON body`).
 - `Content-Type` is not required for JSON parsing.
 - Unknown JSON fields are ignored by deserialization.
-- `PATCH` updates only provided fields and does not support explicit clearing of optional fields.
+- `PATCH` uses JSON Merge Patch semantics for optional per-user fields: omitted means unchanged, explicit `null` removes the config entry, and a non-null value sets it.
 - `If-Match` supports both quoted and unquoted values; surrounding whitespace is trimmed.
 
 ## Query Parameters
@@ -172,11 +211,11 @@ Notes:
 | Field | Type | Required | Description |
 | --- | --- | --- | --- |
 | `secret` | `string` | no | Exactly 32 hex chars. |
-| `user_ad_tag` | `string` | no | Exactly 32 hex chars. |
-| `max_tcp_conns` | `usize` | no | Per-user concurrent TCP limit. |
-| `expiration_rfc3339` | `string` | no | RFC3339 expiration timestamp. |
-| `data_quota_bytes` | `u64` | no | Per-user traffic quota. |
-| `max_unique_ips` | `usize` | no | Per-user unique source IP limit. |
+| `user_ad_tag` | `string|null` | no | Exactly 32 hex chars; `null` removes the per-user ad tag. |
+| `max_tcp_conns` | `usize|null` | no | Per-user concurrent TCP limit; `null` removes the per-user override. |
+| `expiration_rfc3339` | `string|null` | no | RFC3339 expiration timestamp; `null` removes the expiration. |
+| `data_quota_bytes` | `u64|null` | no | Per-user traffic quota; `null` removes the per-user quota. |
+| `max_unique_ips` | `usize|null` | no | Per-user unique source IP limit; `null` removes the per-user override. |
 
 ### `access.user_source_deny` via API
 - In current API surface, per-user deny-list is **not** exposed as a dedicated field in `CreateUserRequest` / `PatchUserRequest`.
@@ -198,7 +237,7 @@ bob = ["198.51.100.42/32"]
 | --- | --- | --- | --- |
 | `secret` | `string` | no | Exactly 32 hex chars. If missing, generated automatically. |
 
-Note: the request contract is defined, but the corresponding route currently returns `404` (see routing edge cases).
+An empty request body is accepted and generates a new secret automatically.
 
 ## Response Data Contracts
 
@@ -208,15 +247,33 @@ Note: the request contract is defined, but the corresponding route currently ret
 | `status` | `string` | Always `"ok"`. |
 | `read_only` | `bool` | Mirrors current API `read_only` mode. |
 
+### `HealthReadyData`
+| Field | Type | Description |
+| --- | --- | --- |
+| `ready` | `bool` | `true` when admission is open and at least one upstream is healthy. |
+| `status` | `string` | `"ready"` or `"not_ready"`. |
+| `reason` | `string?` | `admission_closed` or `no_healthy_upstreams` when not ready. |
+| `admission_open` | `bool` | Current admission-gate state. |
+| `healthy_upstreams` | `usize` | Number of healthy upstream entries. |
+| `total_upstreams` | `usize` | Number of configured upstream entries. |
+
 ### `SummaryData`
 | Field | Type | Description |
 | --- | --- | --- |
 | `uptime_seconds` | `f64` | Process uptime in seconds. |
 | `connections_total` | `u64` | Total accepted client connections. |
 | `connections_bad_total` | `u64` | Failed/invalid client connections. |
+| `connections_bad_by_class` | `ClassCount[]` | Failed/invalid connections grouped by class. |
+| `handshake_failures_by_class` | `ClassCount[]` | Handshake failures grouped by class. |
 | `handshake_timeouts_total` | `u64` | Handshake timeout count. |
 | `configured_users` | `usize` | Number of configured users in config. |
 
+#### `ClassCount`
+| Field | Type | Description |
+| --- | --- | --- |
+| `class` | `string` | Failure class label. |
+| `total` | `u64` | Counter value for this class. |
+
 ### `SystemInfoData`
 | Field | Type | Description |
 | --- | --- | --- |
@@ -241,7 +298,12 @@ Note: the request contract is defined, but the corresponding route currently ret
 | `conditional_cast_enabled` | `bool` | Whether conditional ME admission logic is enabled (`general.use_middle_proxy`). |
 | `me_runtime_ready` | `bool` | Current ME runtime readiness status used for conditional gate decisions. |
 | `me2dc_fallback_enabled` | `bool` | Whether ME -> direct fallback is enabled. |
+| `me2dc_fast_enabled` | `bool` | Whether fast ME -> direct fallback is enabled. |
 | `use_middle_proxy` | `bool` | Current transport mode preference. |
+| `route_mode` | `string` | Current route mode label from route runtime controller. |
+| `reroute_active` | `bool` | `true` when ME fallback currently routes new sessions to Direct-DC. |
+| `reroute_to_direct_at_epoch_secs` | `u64?` | Unix timestamp when current direct reroute began. |
+| `reroute_reason` | `string?` | `fast_not_ready_fallback` or `strict_grace_fallback` while reroute is active. |
 | `startup_status` | `string` | Startup status (`pending`, `initializing`, `ready`, `failed`, `skipped`). |
 | `startup_stage` | `string` | Current startup stage identifier. |
 | `startup_progress_pct` | `f64` | Startup progress percentage (`0..100`). |
@@ -292,11 +354,13 @@ Note: the request contract is defined, but the corresponding route currently ret
 | `upstream` | `EffectiveUpstreamLimits` | Effective upstream connect/retry limits. |
 | `middle_proxy` | `EffectiveMiddleProxyLimits` | Effective ME pool/floor/reconnect limits. |
 | `user_ip_policy` | `EffectiveUserIpPolicyLimits` | Effective unique-IP policy mode/window. |
+| `user_tcp_policy` | `EffectiveUserTcpPolicyLimits` | Effective per-user TCP connection policy. |
 
 #### `EffectiveTimeoutLimits`
 | Field | Type | Description |
 | --- | --- | --- |
 | `client_handshake_secs` | `u64` | Client handshake timeout. |
+| `client_first_byte_idle_secs` | `u64` | First-byte idle timeout before protocol classification. |
 | `tg_connect_secs` | `u64` | Upstream Telegram connect timeout. |
 | `client_keepalive_secs` | `u64` | Client keepalive interval. |
 | `client_ack_secs` | `u64` | ACK timeout. |
@@ -335,13 +399,20 @@ Note: the request contract is defined, but the corresponding route currently ret
 | `writer_pick_mode` | `string` | Writer picker mode (`sorted_rr`, `p2c`). |
 | `writer_pick_sample_size` | `u8` | Candidate sample size for `p2c` picker mode. |
 | `me2dc_fallback` | `bool` | Effective ME -> direct fallback flag. |
+| `me2dc_fast` | `bool` | Effective fast fallback flag. |
 
 #### `EffectiveUserIpPolicyLimits`
 | Field | Type | Description |
 | --- | --- | --- |
+| `global_each` | `usize` | Global per-user unique-IP limit applied when no per-user override exists. |
 | `mode` | `string` | Unique-IP policy mode (`active_window`, `time_window`, `combined`). |
 | `window_secs` | `u64` | Time window length used by unique-IP policy. |
 
+#### `EffectiveUserTcpPolicyLimits`
+| Field | Type | Description |
+| --- | --- | --- |
+| `global_each` | `usize` | Global per-user concurrent TCP limit applied when no per-user override exists. |
+
 ### `SecurityPostureData`
 | Field | Type | Description |
 | --- | --- | --- |
@@ -445,6 +516,8 @@ Note: the request contract is defined, but the corresponding route currently ret
 | --- | --- | --- |
 | `counters` | `RuntimeMeQualityCountersData` | Key ME lifecycle/error counters. |
 | `route_drops` | `RuntimeMeQualityRouteDropData` | Route drop counters by reason. |
+| `family_states` | `RuntimeMeQualityFamilyStateData[]` | Per-family ME route/recovery state rows. |
+| `drain_gate` | `RuntimeMeQualityDrainGateData` | Current ME drain-gate decision state. |
 | `dc_rtt` | `RuntimeMeQualityDcRttData[]` | Per-DC RTT and writer coverage rows. |
 
 #### `RuntimeMeQualityCountersData`
@@ -466,6 +539,24 @@ Note: the request contract is defined, but the corresponding route currently ret
 | `queue_full_base_total` | `u64` | Route drops in base-queue path. |
 | `queue_full_high_total` | `u64` | Route drops in high-priority queue path. |
 
+#### `RuntimeMeQualityFamilyStateData`
+| Field | Type | Description |
+| --- | --- | --- |
+| `family` | `string` | Address family label. |
+| `state` | `string` | Current family state label. |
+| `state_since_epoch_secs` | `u64` | Unix timestamp when current state began. |
+| `suppressed_until_epoch_secs` | `u64?` | Unix timestamp until suppression remains active. |
+| `fail_streak` | `u32` | Consecutive failure count. |
+| `recover_success_streak` | `u32` | Consecutive recovery success count. |
+
+#### `RuntimeMeQualityDrainGateData`
+| Field | Type | Description |
+| --- | --- | --- |
+| `route_quorum_ok` | `bool` | Whether route quorum condition allows drain. |
+| `redundancy_ok` | `bool` | Whether redundancy condition allows drain. |
+| `block_reason` | `string` | Current drain block reason label. |
+| `updated_at_epoch_secs` | `u64` | Unix timestamp of the latest gate update. |
+
 #### `RuntimeMeQualityDcRttData`
 | Field | Type | Description |
 | --- | --- | --- |
@@ -728,11 +819,24 @@ Note: the request contract is defined, but the corresponding route currently ret
 | `uptime_seconds` | `f64` | Process uptime. |
 | `connections_total` | `u64` | Total accepted connections. |
 | `connections_bad_total` | `u64` | Failed/invalid connections. |
+| `connections_bad_by_class` | `ClassCount[]` | Failed/invalid connections grouped by class. |
+| `handshake_failures_by_class` | `ClassCount[]` | Handshake failures grouped by class. |
 | `handshake_timeouts_total` | `u64` | Handshake timeouts. |
+| `accept_permit_timeout_total` | `u64` | Listener admission permit acquisition timeouts. |
 | `configured_users` | `usize` | Configured user count. |
 | `telemetry_core_enabled` | `bool` | Core telemetry toggle. |
 | `telemetry_user_enabled` | `bool` | User telemetry toggle. |
 | `telemetry_me_level` | `string` | ME telemetry level (`off|normal|verbose`). |
+| `conntrack_control_enabled` | `bool` | Whether conntrack control is enabled by policy. |
+| `conntrack_control_available` | `bool` | Whether conntrack control backend is currently available. |
+| `conntrack_pressure_active` | `bool` | Current conntrack pressure flag. |
+| `conntrack_event_queue_depth` | `u64` | Current conntrack close-event queue depth. |
+| `conntrack_rule_apply_ok` | `bool` | Last conntrack rule application state. |
+| `conntrack_delete_attempt_total` | `u64` | Conntrack delete attempts. |
+| `conntrack_delete_success_total` | `u64` | Successful conntrack deletes. |
+| `conntrack_delete_not_found_total` | `u64` | Conntrack delete misses. |
+| `conntrack_delete_error_total` | `u64` | Conntrack delete errors. |
+| `conntrack_close_event_drop_total` | `u64` | Dropped conntrack close events. |
 
 #### `ZeroUpstreamData`
 | Field | Type | Description |
@@ -819,6 +923,24 @@ Note: the request contract is defined, but the corresponding route currently ret
 | `route_drop_queue_full_total` | `u64` | Route drops due to full queue (total). |
 | `route_drop_queue_full_base_total` | `u64` | Route drops in base queue mode. |
 | `route_drop_queue_full_high_total` | `u64` | Route drops in high queue mode. |
+| `d2c_batches_total` | `u64` | ME D->C batch flushes. |
+| `d2c_batch_frames_total` | `u64` | ME D->C frames included in batches. |
+| `d2c_batch_bytes_total` | `u64` | ME D->C payload bytes included in batches. |
+| `d2c_flush_reason_queue_drain_total` | `u64` | Flushes caused by queue drain. |
+| `d2c_flush_reason_batch_frames_total` | `u64` | Flushes caused by frame-count batch limit. |
+| `d2c_flush_reason_batch_bytes_total` | `u64` | Flushes caused by byte-count batch limit. |
+| `d2c_flush_reason_max_delay_total` | `u64` | Flushes caused by max-delay budget. |
+| `d2c_flush_reason_ack_immediate_total` | `u64` | Flushes caused by immediate ACK policy. |
+| `d2c_flush_reason_close_total` | `u64` | Flushes caused by close path. |
+| `d2c_data_frames_total` | `u64` | ME D->C data frames. |
+| `d2c_ack_frames_total` | `u64` | ME D->C ACK frames. |
+| `d2c_payload_bytes_total` | `u64` | ME D->C payload bytes. |
+| `d2c_write_mode_coalesced_total` | `u64` | Coalesced D->C writes. |
+| `d2c_write_mode_split_total` | `u64` | Split D->C writes. |
+| `d2c_quota_reject_pre_write_total` | `u64` | D->C quota rejects before write. |
+| `d2c_quota_reject_post_write_total` | `u64` | D->C quota rejects after write. |
+| `d2c_frame_buf_shrink_total` | `u64` | D->C frame-buffer shrink operations. |
+| `d2c_frame_buf_shrink_bytes_total` | `u64` | Bytes released by D->C frame-buffer shrink operations. |
 | `socks_kdf_strict_reject_total` | `u64` | SOCKS KDF strict rejects. |
 | `socks_kdf_compat_fallback_total` | `u64` | SOCKS KDF compat fallbacks. |
 | `endpoint_quarantine_total` | `u64` | Endpoint quarantine activations. |
@@ -978,6 +1100,8 @@ Note: the request contract is defined, but the corresponding route currently ret
 | `required_writers` | `usize` | Required writers based on current floor policy. |
 | `alive_writers` | `usize` | Writers currently alive. |
 | `coverage_pct` | `f64` | `alive_writers / required_writers * 100`. |
+| `fresh_alive_writers` | `usize` | Alive writers that match freshness requirements. |
+| `fresh_coverage_pct` | `f64` | `fresh_alive_writers / required_writers * 100`. |
 
 #### `MeWriterStatus`
 | Field | Type | Description |
@@ -992,6 +1116,12 @@ Note: the request contract is defined, but the corresponding route currently ret
 | `bound_clients` | `usize` | Number of currently bound clients. |
 | `idle_for_secs` | `u64?` | Idle age in seconds if idle. |
 | `rtt_ema_ms` | `f64?` | RTT exponential moving average. |
+| `matches_active_generation` | `bool` | Whether this writer belongs to the active pool generation. |
+| `in_desired_map` | `bool` | Whether this writer's endpoint remains in desired topology. |
+| `allow_drain_fallback` | `bool` | Whether drain fallback is allowed for this writer. |
+| `drain_started_at_epoch_secs` | `u64?` | Unix timestamp when drain started. |
+| `drain_deadline_epoch_secs` | `u64?` | Unix timestamp of drain deadline. |
+| `drain_over_ttl` | `bool` | Whether drain has exceeded its TTL. |
 
 ### `DcStatusData`
 | Field | Type | Description |
@@ -1016,6 +1146,8 @@ Note: the request contract is defined, but the corresponding route currently ret
 | `floor_capped` | `bool` | `true` when computed floor target was capped by active limits. |
 | `alive_writers` | `usize` | Alive writers in this DC. |
 | `coverage_pct` | `f64` | `alive_writers / required_writers * 100`. |
+| `fresh_alive_writers` | `usize` | Fresh alive writers in this DC. |
+| `fresh_coverage_pct` | `f64` | `fresh_alive_writers / required_writers * 100`. |
 | `rtt_ms` | `f64?` | Aggregated RTT for DC. |
 | `load` | `usize` | Active client sessions bound to this DC. |
 
@@ -1029,6 +1161,7 @@ Note: the request contract is defined, but the corresponding route currently ret
 | Field | Type | Description |
 | --- | --- | --- |
 | `username` | `string` | Username. |
+| `in_runtime` | `bool` | Whether current runtime config already contains this user. |
 | `user_ad_tag` | `string?` | Optional ad tag (32 hex chars). |
 | `max_tcp_conns` | `usize?` | Optional max concurrent TCP limit. |
 | `expiration_rfc3339` | `string?` | Optional expiration timestamp. |
@@ -1042,12 +1175,25 @@ Note: the request contract is defined, but the corresponding route currently ret
 | `total_octets` | `u64` | Total traffic octets for this user. |
 | `links` | `UserLinks` | Active connection links derived from current config. |
 
+### `UserActiveIps`
+| Field | Type | Description |
+| --- | --- | --- |
+| `username` | `string` | Username with at least one active tracked source IP. |
+| `active_ips` | `ip[]` | Active source IPs for this user. |
+
 #### `UserLinks`
 | Field | Type | Description |
 | --- | --- | --- |
 | `classic` | `string[]` | Active `tg://proxy` links for classic mode. |
 | `secure` | `string[]` | Active `tg://proxy` links for secure/DD mode. |
 | `tls` | `string[]` | Active `tg://proxy` links for EE-TLS mode (for each host+TLS domain). |
+| `tls_domains` | `TlsDomainLink[]` | Extra TLS-domain links as explicit domain/link pairs for `censorship.tls_domains`. |
+
+#### `TlsDomainLink`
+| Field | Type | Description |
+| --- | --- | --- |
+| `domain` | `string` | TLS domain represented by the link. |
+| `link` | `string` | `tg://proxy` link for this domain. |
 
 Link generation uses active config and enabled modes:
 - Link port is `general.links.public_port` when configured; otherwise `server.port`.
@@ -1067,13 +1213,27 @@ Link generation uses active config and enabled modes:
 | `user` | `UserInfo` | Created or updated user view. |
 | `secret` | `string` | Effective user secret. |
 
+### `DeleteUserResponse`
+| Field | Type | Description |
+| --- | --- | --- |
+| `username` | `string` | Deleted username. |
+| `in_runtime` | `bool` | `true` when runtime config still contains the user and hot-reload has not applied deletion yet. |
+
+### `ResetUserQuotaResponse`
+| Field | Type | Description |
+| --- | --- | --- |
+| `username` | `string` | User whose runtime quota counter was reset. |
+| `used_bytes` | `u64` | Current used bytes after reset; always `0` on success. |
+| `last_reset_epoch_secs` | `u64` | Unix timestamp of the reset operation. |
+
 ## Mutation Semantics
 
 | Endpoint | Notes |
 | --- | --- |
 | `POST /v1/users` | Creates user, validates config, then atomically updates only affected `access.*` TOML tables (`access.users` always, plus optional per-user tables present in request). |
-| `PATCH /v1/users/{username}` | Partial update of provided fields only. Missing fields remain unchanged. Current implementation persists full config document on success. |
-| `POST /v1/users/{username}/rotate-secret` | Currently returns `404` in runtime route matcher; request schema is reserved for intended behavior. |
+| `PATCH /v1/users/{username}` | Partial update of provided fields only. Missing fields remain unchanged; explicit `null` removes optional per-user entries. The write path updates only affected `access.*` TOML tables. |
+| `POST /v1/users/{username}/rotate-secret` | Replaces the user's secret with a provided valid 32-hex value or a generated value, then returns the effective secret in `CreateUserResponse`. |
+| `POST /v1/users/{username}/reset-quota` | Resets the runtime quota counter for the route username, persists quota state to `general.quota_state_path`, and does not modify user config. |
 | `DELETE /v1/users/{username}` | Deletes only specified user, removes this user from related optional `access.user_*` maps, blocks last-user deletion, and atomically updates only related `access.*` TOML tables. |
 
 All mutating endpoints:
@@ -1148,5 +1308,4 @@ When `general.use_middle_proxy=true` and `general.me2dc_fallback=true`:
 
 ## Known Limitations (Current Release)
 
-- `POST /v1/users/{username}/rotate-secret` is currently unreachable in route matcher and returns `404`.
 - API runtime controls under `server.api` are documented as restart-required; hot-reload behavior for these fields is not strictly uniform in all change combinations.

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
-        .get("upstreams")
-        .and_then(toml::Value::as_array)
-    {
+    if let Some(upstreams) = parsed_toml.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
-            .cleanup_queue
-            .lock()
-            .unwrap_or_else(|poisoned| poisoned.into_inner())
-            .len();
+        let cleanup_queue_len = self.cleanup_queue_len.load(Ordering::Relaxed) as usize;
         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!(
-                                    &e,
-                                    crate::error::ProxyError::MiddleConnectionLost
-                                );
-                                let route_switched = matches!(
-                                    &e,
-                                    crate::error::ProxyError::Proxy(msg) if msg == ROUTE_SWITCH_ERROR_MSG
-                                );
+                                let me_closed =
+                                    matches!(&e, crate::error::ProxyError::MiddleConnectionLost);
+                                let route_switched =
+                                    matches!(&e, crate::error::ProxyError::RouteSwitched);
 
                                 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!(
-        out,
-        "# TYPE telemt_tls_front_profile_age_seconds gauge"
-    );
+    let _ = writeln!(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!(
-        out,
-        "# TYPE telemt_tls_front_profile_ticket_records gauge"
-    );
+    let _ = writeln!(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!(
-        out,
-        "# TYPE telemt_tls_front_profile_app_data_bytes gauge"
-    );
+    let _ = writeln!(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,
-            item.source,
-            item.is_default,
-            item.has_cert_info,
-            item.has_cert_payload
+            domain, 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!(
-        out,
-        "# TYPE telemt_quota_contention_timeout_total counter"
-    );
+    let _ = writeln!(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!(
-        out,
-        "# TYPE telemt_rate_limiter_burst_bound_bytes gauge"
-    );
+    let _ = writeln!(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
-            .contains("telemt_connections_bad_by_class_total{class=\"tls_handshake_bad_client\"} 1"));
+        assert!(output.contains(
+            "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")
-        );
-        assert!(
-            output.contains("telemt_tls_front_profile_app_data_bytes{domain=\"primary.example\"} 1536")
+            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(
-                    &self.beobachten,
-                    &self.config,
-                    self.peer.ip(),
-                    "other",
-                );
+                record_beobachten_class(&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,
-    cutover_stagger_delay,
+    RelayRouteMode, RouteCutoverState, affected_cutover_state, 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,
-    cutover_stagger_delay,
+    RelayRouteMode, RouteCutoverState, affected_cutover_state, 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
-        && matches!(writer_result, Err(ProxyError::MiddleConnectionLost))
-    {
+    if client_closed && 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)
-                    .await
+                match reserve_user_quota_with_yield(
+                    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 =
-                match write_client_payload(client_writer, proto_tag, flags, &data, rng, frame_buf)
-                    .await
-                {
-                    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);
+            let write_mode = match write_client_payload(
+                client_writer,
+                proto_tag,
+                flags,
+                &data,
+                rng,
+                frame_buf,
+                cancel,
+            )
+            .await
+            {
+                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)
-                .await?;
-            write_client_ack(client_writer, proto_tag, confirm).await?;
+            wait_for_traffic_budget_or_cancel(
+                traffic_lease,
+                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(frame_buf.as_slice())
-                        .await
-                        .map_err(ProxyError::Io)?;
+                    write_all_client_or_cancel(client_writer, frame_buf.as_slice(), cancel).await?;
                     MeD2cWriteMode::Coalesced
                 } else {
                     let header = [first];
-                    client_writer
-                        .write_all(&header)
-                        .await
-                        .map_err(ProxyError::Io)?;
-                    client_writer
-                        .write_all(data)
-                        .await
-                        .map_err(ProxyError::Io)?;
+                    write_all_client_or_cancel(client_writer, &header, cancel).await?;
+                    write_all_client_or_cancel(client_writer, data, cancel).await?;
                     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(frame_buf.as_slice())
-                        .await
-                        .map_err(ProxyError::Io)?;
+                    write_all_client_or_cancel(client_writer, frame_buf.as_slice(), cancel).await?;
                     MeD2cWriteMode::Coalesced
                 } else {
                     let header = [first, lw[0], lw[1], lw[2]];
-                    client_writer
-                        .write_all(&header)
-                        .await
-                        .map_err(ProxyError::Io)?;
-                    client_writer
-                        .write_all(data)
-                        .await
-                        .map_err(ProxyError::Io)?;
+                    write_all_client_or_cancel(client_writer, &header, cancel).await?;
+                    write_all_client_or_cancel(client_writer, data, cancel).await?;
                     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(frame_buf.as_slice())
-                    .await
-                    .map_err(ProxyError::Io)?;
+                write_all_client_or_cancel(client_writer, frame_buf.as_slice(), cancel).await?;
                 MeD2cWriteMode::Coalesced
             } else {
                 let header = len_val.to_le_bytes();
-                client_writer
-                    .write_all(&header)
-                    .await
-                    .map_err(ProxyError::Io)?;
-                client_writer
-                    .write_all(data)
-                    .await
-                    .map_err(ProxyError::Io)?;
+                write_all_client_or_cancel(client_writer, &header, cancel).await?;
+                write_all_client_or_cancel(client_writer, data, cancel).await?;
                 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(frame_buf.as_slice())
-                        .await
-                        .map_err(ProxyError::Io)?;
+                    write_all_client_or_cancel(client_writer, frame_buf.as_slice(), cancel).await?;
                 }
                 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(&bytes)
-        .await
-        .map_err(ProxyError::Io)
+    write_all_client_or_cancel(client_writer, &bytes, cancel).await
+}
+
+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(
-                        this.user_stats.as_ref(),
-                        refund_bytes,
-                    );
+                    refund_reserved_quota_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(
-                        this.user_stats.as_ref(),
-                        refund_bytes,
-                    );
+                    refund_reserved_quota_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!(
-            relay_result,
-            Err(ProxyError::Proxy(ref msg)) if msg == ROUTE_SWITCH_ERROR_MSG
-        ),
+        matches!(relay_result, Err(ProxyError::RouteSwitched)),
         "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!(
-                relay_result,
-                Err(ProxyError::Proxy(ref msg)) if msg == ROUTE_SWITCH_ERROR_MSG
-            ),
+            matches!(relay_result, Err(ProxyError::RouteSwitched)),
             "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!(
-            result,
-            Err(ProxyError::Proxy(ref msg)) if msg == ROUTE_SWITCH_ERROR_MSG
-        ),
+        matches!(result, Err(ProxyError::RouteSwitched)),
         "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 tokio::io::{AsyncWrite, AsyncWriteExt};
+use std::task::{Context, Poll, Wake};
+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
-            .load(Ordering::Relaxed)
+        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)
@@ -2536,12 +2535,7 @@ impl Stats {
             .unwrap_or(0)
     }
 
-    pub fn load_user_quota_state(
-        &self,
-        user: &str,
-        used_bytes: u64,
-        last_reset_epoch_secs: u64,
-    ) {
+    pub fn load_user_quota_state(&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(
-                    behavior.app_data_record_sizes.len(),
-                ),
+                app_data_records: cached
+                    .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,