Hot-path Cleanup and Timeout Invariants

Limit only new ip when TimeWindow + Fix WorkingDirectory behavior + Atomically updates with Includes + Expose tls_domains links as domain-link pairs + TLS Fetcher on multiple tls_domains

Cargo.lock

@@ -2791,7 +2791,7 @@ checksum = "7b2093cf4c8eb1e67749a6762251bc9cd836b6fc171623bd0a9d324d37af2417"
 
 [[package]]
 name = "telemt"
-version = "3.4.7"
+version = "3.4.9"
 dependencies = [
  "aes",
  "anyhow",

Cargo.toml

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

README.md

@@ -99,7 +99,7 @@ Monero (XMR) directly:
 8Bk4tZEYPQWSypeD2hrUXG2rKbAKF16GqEN942ZdAP5cFdSqW6h4DwkP5cJMAdszzuPeHeHZPTyjWWFwzeFdjuci3ktfMoB
 ```
 
-All donations go toward infrastructure, development, and research.
+All donations go toward infrastructure, development and research
 
 
 ![telemt_scheme](docs/assets/telemt.png)

src/api/config_store.rs

@@ -82,6 +82,7 @@ pub(super) async fn load_config_from_disk(config_path: &Path) -> Result<ProxyCon
         .map_err(|e| ApiFailure::internal(format!("failed to load config: {}", e)))
 }
 
+#[allow(dead_code)]
 pub(super) async fn save_config_to_disk(
     config_path: &Path,
     cfg: &ProxyConfig,
@@ -106,6 +107,12 @@ pub(super) async fn save_access_sections_to_disk(
         if applied.contains(section) {
             continue;
         }
+        if find_toml_table_bounds(&content, section.table_name()).is_none()
+            && access_section_is_empty(cfg, *section)
+        {
+            applied.push(*section);
+            continue;
+        }
         let rendered = render_access_section(cfg, *section)?;
         content = upsert_toml_table(&content, section.table_name(), &rendered);
         applied.push(*section);
@@ -183,6 +190,17 @@ fn render_access_section(cfg: &ProxyConfig, section: AccessSection) -> Result<St
     Ok(out)
 }
 
+fn access_section_is_empty(cfg: &ProxyConfig, section: AccessSection) -> bool {
+    match section {
+        AccessSection::Users => cfg.access.users.is_empty(),
+        AccessSection::UserAdTags => cfg.access.user_ad_tags.is_empty(),
+        AccessSection::UserMaxTcpConns => cfg.access.user_max_tcp_conns.is_empty(),
+        AccessSection::UserExpirations => cfg.access.user_expirations.is_empty(),
+        AccessSection::UserDataQuota => cfg.access.user_data_quota.is_empty(),
+        AccessSection::UserMaxUniqueIps => cfg.access.user_max_unique_ips.is_empty(),
+    }
+}
+
 fn serialize_table_body<T: Serialize>(value: &T) -> Result<String, ApiFailure> {
     toml::to_string(value)
         .map_err(|e| ApiFailure::internal(format!("failed to serialize access section: {}", e)))

src/api/model.rs

@@ -456,6 +456,13 @@ pub(super) struct UserLinks {
     pub(super) classic: Vec<String>,
     pub(super) secure: Vec<String>,
     pub(super) tls: Vec<String>,
+    pub(super) tls_domains: Vec<TlsDomainLink>,
+}
+
+#[derive(Serialize)]
+pub(super) struct TlsDomainLink {
+    pub(super) domain: String,
+    pub(super) link: String,
 }
 
 #[derive(Serialize)]

src/api/users.rs

@@ -8,12 +8,12 @@ use crate::stats::Stats;
 
 use super::ApiShared;
 use super::config_store::{
-    AccessSection, ensure_expected_revision, load_config_from_disk, save_access_sections_to_disk,
-    save_config_to_disk,
+    AccessSection, current_revision, ensure_expected_revision, load_config_from_disk,
+    save_access_sections_to_disk,
 };
 use super::model::{
     ApiFailure, CreateUserRequest, CreateUserResponse, PatchUserRequest, RotateSecretRequest,
-    UserInfo, UserLinks, is_valid_ad_tag, is_valid_user_secret, is_valid_username,
+    TlsDomainLink, UserInfo, UserLinks, is_valid_ad_tag, is_valid_user_secret, is_valid_username,
     parse_optional_expiration, parse_patch_expiration, random_user_secret,
 };
 use super::patch::Patch;
@@ -176,6 +176,13 @@ pub(super) async fn patch_user(
     expected_revision: Option<String>,
     shared: &ApiShared,
 ) -> Result<(UserInfo, String), ApiFailure> {
+    let touches_users = body.secret.is_some();
+    let touches_user_ad_tags = !matches!(&body.user_ad_tag, Patch::Unchanged);
+    let touches_user_max_tcp_conns = !matches!(&body.max_tcp_conns, Patch::Unchanged);
+    let touches_user_expirations = !matches!(&body.expiration_rfc3339, Patch::Unchanged);
+    let touches_user_data_quota = !matches!(&body.data_quota_bytes, Patch::Unchanged);
+    let touches_user_max_unique_ips = !matches!(&body.max_unique_ips, Patch::Unchanged);
+
     if let Some(secret) = body.secret.as_ref()
         && !is_valid_user_secret(secret)
     {
@@ -265,7 +272,31 @@ pub(super) async fn patch_user(
     cfg.validate()
         .map_err(|e| ApiFailure::bad_request(format!("config validation failed: {}", e)))?;
 
-    let revision = save_config_to_disk(&shared.config_path, &cfg).await?;
+    let mut touched_sections = Vec::new();
+    if touches_users {
+        touched_sections.push(AccessSection::Users);
+    }
+    if touches_user_ad_tags {
+        touched_sections.push(AccessSection::UserAdTags);
+    }
+    if touches_user_max_tcp_conns {
+        touched_sections.push(AccessSection::UserMaxTcpConns);
+    }
+    if touches_user_expirations {
+        touched_sections.push(AccessSection::UserExpirations);
+    }
+    if touches_user_data_quota {
+        touched_sections.push(AccessSection::UserDataQuota);
+    }
+    if touches_user_max_unique_ips {
+        touched_sections.push(AccessSection::UserMaxUniqueIps);
+    }
+
+    let revision = if touched_sections.is_empty() {
+        current_revision(&shared.config_path).await?
+    } else {
+        save_access_sections_to_disk(&shared.config_path, &cfg, &touched_sections).await?
+    };
     drop(_guard);
     match max_unique_ips_change {
         Some(Some(limit)) => shared.ip_tracker.set_user_limit(user, limit).await,
@@ -438,6 +469,7 @@ pub(super) async fn users_from_config(
                 classic: Vec::new(),
                 secure: Vec::new(),
                 tls: Vec::new(),
+                tls_domains: Vec::new(),
             });
         users.push(UserInfo {
             in_runtime: runtime_cfg
@@ -492,10 +524,12 @@ fn build_user_links(
         .public_port
         .unwrap_or(resolve_default_link_port(cfg));
     let tls_domains = resolve_tls_domains(cfg);
+    let extra_tls_domains = resolve_extra_tls_domains(cfg);
 
     let mut classic = Vec::new();
     let mut secure = Vec::new();
     let mut tls = Vec::new();
+    let mut tls_domain_links = Vec::new();
 
     for host in &hosts {
         if cfg.general.modes.classic {
@@ -518,6 +552,17 @@ fn build_user_links(
                     host, port, secret, domain_hex
                 ));
             }
+            for domain in &extra_tls_domains {
+                let domain_hex = hex::encode(domain);
+                let link = format!(
+                    "tg://proxy?server={}&port={}&secret=ee{}{}",
+                    host, port, secret, domain_hex
+                );
+                tls_domain_links.push(TlsDomainLink {
+                    domain: (*domain).to_string(),
+                    link,
+                });
+            }
         }
     }
 
@@ -525,6 +570,7 @@ fn build_user_links(
         classic,
         secure,
         tls,
+        tls_domains: tls_domain_links,
     }
 }
 
@@ -641,6 +687,19 @@ fn resolve_tls_domains(cfg: &ProxyConfig) -> Vec<&str> {
     domains
 }
 
+fn resolve_extra_tls_domains(cfg: &ProxyConfig) -> Vec<&str> {
+    let mut domains = Vec::with_capacity(cfg.censorship.tls_domains.len());
+    let primary = cfg.censorship.tls_domain.as_str();
+    for domain in &cfg.censorship.tls_domains {
+        let value = domain.as_str();
+        if value.is_empty() || value == primary || domains.contains(&value) {
+            continue;
+        }
+        domains.push(value);
+    }
+    domains
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -730,4 +789,80 @@ mod tests {
         assert!(alice.in_runtime);
         assert!(!bob.in_runtime);
     }
+
+    #[tokio::test]
+    async fn users_from_config_returns_tls_link_for_each_tls_domain() {
+        let mut cfg = ProxyConfig::default();
+        cfg.access.users.insert(
+            "alice".to_string(),
+            "0123456789abcdef0123456789abcdef".to_string(),
+        );
+        cfg.general.modes.classic = false;
+        cfg.general.modes.secure = false;
+        cfg.general.modes.tls = true;
+        cfg.general.links.public_host = Some("proxy.example.net".to_string());
+        cfg.general.links.public_port = Some(443);
+        cfg.censorship.tls_domain = "front-a.example.com".to_string();
+        cfg.censorship.tls_domains = vec![
+            "front-b.example.com".to_string(),
+            "front-c.example.com".to_string(),
+            "front-b.example.com".to_string(),
+            "front-a.example.com".to_string(),
+        ];
+
+        let stats = Stats::new();
+        let tracker = UserIpTracker::new();
+        let users = users_from_config(&cfg, &stats, &tracker, None, None, None).await;
+        let alice = users
+            .iter()
+            .find(|entry| entry.username == "alice")
+            .expect("alice must be present");
+
+        assert_eq!(alice.links.tls.len(), 3);
+        assert!(
+            alice
+                .links
+                .tls
+                .iter()
+                .any(|link| link.ends_with(&hex::encode("front-a.example.com")))
+        );
+        assert!(
+            alice
+                .links
+                .tls
+                .iter()
+                .any(|link| link.ends_with(&hex::encode("front-b.example.com")))
+        );
+        assert!(
+            alice
+                .links
+                .tls
+                .iter()
+                .any(|link| link.ends_with(&hex::encode("front-c.example.com")))
+        );
+        assert_eq!(alice.links.tls_domains.len(), 2);
+        assert!(
+            alice
+                .links
+                .tls_domains
+                .iter()
+                .any(|entry| entry.domain == "front-b.example.com"
+                    && entry.link.ends_with(&hex::encode("front-b.example.com")))
+        );
+        assert!(
+            alice
+                .links
+                .tls_domains
+                .iter()
+                .any(|entry| entry.domain == "front-c.example.com"
+                    && entry.link.ends_with(&hex::encode("front-c.example.com")))
+        );
+        assert!(
+            !alice
+                .links
+                .tls_domains
+                .iter()
+                .any(|entry| entry.domain == "front-a.example.com")
+        );
+    }
 }

src/config/defaults.rs

@@ -102,7 +102,7 @@ pub(crate) fn default_fake_cert_len() -> usize {
 }
 
 pub(crate) fn default_tls_front_dir() -> String {
-    "/etc/telemt/tlsfront".to_string()
+    "tlsfront".to_string()
 }
 
 pub(crate) fn default_replay_check_len() -> usize {
@@ -568,7 +568,7 @@ pub(crate) fn default_beobachten_flush_secs() -> u64 {
 }
 
 pub(crate) fn default_beobachten_file() -> String {
-    "/etc/telemt/beobachten.txt".to_string()
+    "beobachten.txt".to_string()
 }
 
 pub(crate) fn default_tls_new_session_tickets() -> u8 {

src/config/load.rs

@@ -1087,9 +1087,9 @@ impl ProxyConfig {
             ));
         }
 
-        if config.general.me_route_blocking_send_timeout_ms > 5000 {
+        if !(1..=5000).contains(&config.general.me_route_blocking_send_timeout_ms) {
             return Err(ProxyError::Config(
-                "general.me_route_blocking_send_timeout_ms must be within [0, 5000]".to_string(),
+                "general.me_route_blocking_send_timeout_ms must be within [1, 5000]".to_string(),
             ));
         }
 
@@ -2602,6 +2602,28 @@ mod tests {
         let _ = std::fs::remove_file(path);
     }
 
+    #[test]
+    fn me_route_blocking_send_timeout_ms_zero_is_rejected() {
+        let toml = r#"
+            [general]
+            me_route_blocking_send_timeout_ms = 0
+
+            [censorship]
+            tls_domain = "example.com"
+
+            [access.users]
+            user = "00000000000000000000000000000000"
+        "#;
+        let dir = std::env::temp_dir();
+        let path = dir.join("telemt_me_route_blocking_send_timeout_zero_test.toml");
+        std::fs::write(&path, toml).unwrap();
+        let err = ProxyConfig::load(&path).unwrap_err().to_string();
+        assert!(
+            err.contains("general.me_route_blocking_send_timeout_ms must be within [1, 5000]")
+        );
+        let _ = std::fs::remove_file(path);
+    }
+
     #[test]
     fn me_route_no_writer_mode_is_parsed() {
         let toml = r#"

src/config/types.rs

@@ -778,7 +778,7 @@ pub struct GeneralConfig {
     pub me_route_hybrid_max_wait_ms: u64,
 
     /// Maximum wait in milliseconds for blocking ME writer channel send fallback.
-    /// `0` keeps legacy unbounded wait behavior.
+    /// Must be within [1, 5000].
     #[serde(default = "default_me_route_blocking_send_timeout_ms")]
     pub me_route_blocking_send_timeout_ms: u64,
 

src/ip_tracker.rs

@@ -9,10 +9,12 @@ use std::sync::Mutex;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::time::{Duration, Instant};
 
-use tokio::sync::{Mutex as AsyncMutex, RwLock};
+use tokio::sync::{Mutex as AsyncMutex, RwLock, RwLockWriteGuard};
 
 use crate::config::UserMaxUniqueIpsMode;
 
+const CLEANUP_DRAIN_BATCH_LIMIT: usize = 1024;
+
 #[derive(Debug, Clone)]
 pub struct UserIpTracker {
     active_ips: Arc<RwLock<HashMap<String, HashMap<IpAddr, usize>>>>,
@@ -86,16 +88,27 @@ impl UserIpTracker {
     }
 
     pub(crate) async fn drain_cleanup_queue(&self) {
-        // Serialize queue draining and active-IP mutation so check-and-add cannot
-        // observe stale active entries that are already queued for removal.
-        let _drain_guard = self.cleanup_drain_lock.lock().await;
+        let Ok(_drain_guard) = self.cleanup_drain_lock.try_lock() else {
+            return;
+        };
+
         let to_remove = {
             match self.cleanup_queue.lock() {
                 Ok(mut queue) => {
                     if queue.is_empty() {
                         return;
                     }
-                    std::mem::take(&mut *queue)
+                    let mut drained =
+                        HashMap::with_capacity(queue.len().min(CLEANUP_DRAIN_BATCH_LIMIT));
+                    for _ in 0..CLEANUP_DRAIN_BATCH_LIMIT {
+                        let Some(key) = queue.keys().next().cloned() else {
+                            break;
+                        };
+                        if let Some(count) = queue.remove(&key) {
+                            drained.insert(key, count);
+                        }
+                    }
+                    drained
                 }
                 Err(poisoned) => {
                     let mut queue = poisoned.into_inner();
@@ -103,12 +116,24 @@ impl UserIpTracker {
                         self.cleanup_queue.clear_poison();
                         return;
                     }
-                    let drained = std::mem::take(&mut *queue);
+                    let mut drained =
+                        HashMap::with_capacity(queue.len().min(CLEANUP_DRAIN_BATCH_LIMIT));
+                    for _ in 0..CLEANUP_DRAIN_BATCH_LIMIT {
+                        let Some(key) = queue.keys().next().cloned() else {
+                            break;
+                        };
+                        if let Some(count) = queue.remove(&key) {
+                            drained.insert(key, count);
+                        }
+                    }
                     self.cleanup_queue.clear_poison();
                     drained
                 }
             }
         };
+        if to_remove.is_empty() {
+            return;
+        }
 
         let mut active_ips = self.active_ips.write().await;
         for ((user, ip), pending_count) in to_remove {
@@ -137,6 +162,24 @@ impl UserIpTracker {
             .as_secs()
     }
 
+    async fn active_and_recent_write(
+        &self,
+    ) -> (
+        RwLockWriteGuard<'_, HashMap<String, HashMap<IpAddr, usize>>>,
+        RwLockWriteGuard<'_, HashMap<String, HashMap<IpAddr, Instant>>>,
+    ) {
+        loop {
+            let active_ips = self.active_ips.write().await;
+            match self.recent_ips.try_write() {
+                Ok(recent_ips) => return (active_ips, recent_ips),
+                Err(_) => {
+                    drop(active_ips);
+                    tokio::task::yield_now().await;
+                }
+            }
+        }
+    }
+
     async fn maybe_compact_empty_users(&self) {
         const COMPACT_INTERVAL_SECS: u64 = 60;
         let now_epoch_secs = Self::now_epoch_secs();
@@ -157,10 +200,9 @@ impl UserIpTracker {
             return;
         }
 
-        let mut active_ips = self.active_ips.write().await;
-        let mut recent_ips = self.recent_ips.write().await;
         let window = *self.limit_window.read().await;
         let now = Instant::now();
+        let (mut active_ips, mut recent_ips) = self.active_and_recent_write().await;
 
         for user_recent in recent_ips.values_mut() {
             Self::prune_recent(user_recent, now, window);
@@ -261,12 +303,10 @@ impl UserIpTracker {
         let window = *self.limit_window.read().await;
         let now = Instant::now();
 
-        let mut active_ips = self.active_ips.write().await;
+        let (mut active_ips, mut recent_ips) = self.active_and_recent_write().await;
         let user_active = active_ips
             .entry(username.to_string())
             .or_insert_with(HashMap::new);
-
-        let mut recent_ips = self.recent_ips.write().await;
         let user_recent = recent_ips
             .entry(username.to_string())
             .or_insert_with(HashMap::new);
@@ -278,9 +318,11 @@ impl UserIpTracker {
             return Ok(());
         }
 
+        let is_new_ip = !user_recent.contains_key(&ip);
+
         if let Some(limit) = limit {
             let active_limit_reached = user_active.len() >= limit;
-            let recent_limit_reached = user_recent.len() >= limit;
+            let recent_limit_reached = user_recent.len() >= limit && is_new_ip;
             let deny = match mode {
                 UserMaxUniqueIpsMode::ActiveWindow => active_limit_reached,
                 UserMaxUniqueIpsMode::TimeWindow => recent_limit_reached,
@@ -324,6 +366,13 @@ impl UserIpTracker {
 
     pub async fn get_recent_counts_for_users(&self, users: &[String]) -> HashMap<String, usize> {
         self.drain_cleanup_queue().await;
+        self.get_recent_counts_for_users_snapshot(users).await
+    }
+
+    pub(crate) async fn get_recent_counts_for_users_snapshot(
+        &self,
+        users: &[String],
+    ) -> HashMap<String, usize> {
         let window = *self.limit_window.read().await;
         let now = Instant::now();
         let recent_ips = self.recent_ips.read().await;
@@ -398,19 +447,29 @@ impl UserIpTracker {
 
     pub async fn get_stats(&self) -> Vec<(String, usize, usize)> {
         self.drain_cleanup_queue().await;
+        self.get_stats_snapshot().await
+    }
+
+    pub(crate) async fn get_stats_snapshot(&self) -> Vec<(String, usize, usize)> {
         let active_ips = self.active_ips.read().await;
+        let active_counts = active_ips
+            .iter()
+            .map(|(username, user_ips)| (username.clone(), user_ips.len()))
+            .collect::<Vec<_>>();
+        drop(active_ips);
+
         let max_ips = self.max_ips.read().await;
         let default_max_ips = *self.default_max_ips.read().await;
 
-        let mut stats = Vec::new();
-        for (username, user_ips) in active_ips.iter() {
+        let mut stats = Vec::with_capacity(active_counts.len());
+        for (username, active_count) in active_counts {
             let limit = max_ips
-                .get(username)
+                .get(&username)
                 .copied()
                 .filter(|limit| *limit > 0)
                 .or((default_max_ips > 0).then_some(default_max_ips))
                 .unwrap_or(0);
-            stats.push((username.clone(), user_ips.len(), limit));
+            stats.push((username, active_count, limit));
         }
 
         stats.sort_by(|a, b| a.0.cmp(&b.0));
@@ -860,4 +919,19 @@ mod tests {
             .unwrap_or(false);
         assert!(!stale_exists);
     }
+
+    #[tokio::test]
+    async fn test_time_window_allows_same_ip_reconnect() {
+        let tracker = UserIpTracker::new();
+        tracker.set_user_limit("test_user", 1).await;
+        tracker
+            .set_limit_policy(UserMaxUniqueIpsMode::TimeWindow, 1)
+            .await;
+
+        let ip1 = test_ipv4(10, 4, 0, 1);
+
+        assert!(tracker.check_and_add("test_user", ip1).await.is_ok());
+        tracker.remove_ip("test_user", ip1).await;
+        assert!(tracker.check_and_add("test_user", ip1).await.is_ok());
+    }
 }

src/maestro/helpers.rs

@@ -1,6 +1,6 @@
 #![allow(clippy::items_after_test_module)]
 
-use std::path::PathBuf;
+use std::path::{Path, PathBuf};
 use std::time::Duration;
 
 use tokio::sync::watch;
@@ -17,7 +17,7 @@ use crate::transport::middle_proxy::{
 
 pub(crate) fn resolve_runtime_config_path(
     config_path_cli: &str,
-    startup_cwd: &std::path::Path,
+    startup_cwd: &Path,
     config_path_explicit: bool,
 ) -> PathBuf {
     if config_path_explicit {
@@ -46,6 +46,39 @@ pub(crate) fn resolve_runtime_config_path(
     startup_cwd.join("config.toml")
 }
 
+pub(crate) fn resolve_runtime_base_dir(
+    config_path: &Path,
+    startup_cwd: &Path,
+    config_path_explicit: bool,
+    data_path: Option<&Path>,
+) -> PathBuf {
+    if let Some(path) = data_path {
+        return normalize_runtime_dir(path, startup_cwd);
+    }
+
+    if startup_cwd != Path::new("/") {
+        return normalize_runtime_dir(startup_cwd, startup_cwd);
+    }
+
+    if config_path_explicit
+        && let Some(parent) = config_path.parent()
+        && !parent.as_os_str().is_empty()
+    {
+        return normalize_runtime_dir(parent, startup_cwd);
+    }
+
+    PathBuf::from("/etc/telemt")
+}
+
+fn normalize_runtime_dir(path: &Path, startup_cwd: &Path) -> PathBuf {
+    let absolute = if path.is_absolute() {
+        path.to_path_buf()
+    } else {
+        startup_cwd.join(path)
+    };
+    absolute.canonicalize().unwrap_or(absolute)
+}
+
 /// Parsed CLI arguments.
 pub(crate) struct CliArgs {
     pub config_path: String,
@@ -231,9 +264,11 @@ fn print_help() {
 
 #[cfg(test)]
 mod tests {
+    use std::path::{Path, PathBuf};
+
     use super::{
         expected_handshake_close_description, is_expected_handshake_eof, peer_close_description,
-        resolve_runtime_config_path,
+        resolve_runtime_base_dir, resolve_runtime_config_path,
     };
     use crate::error::{ProxyError, StreamError};
 
@@ -304,6 +339,91 @@ mod tests {
         let _ = std::fs::remove_dir(&startup_cwd);
     }
 
+    #[test]
+    fn resolve_runtime_base_dir_prefers_cli_data_path() {
+        let nonce = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_nanos();
+        let startup_cwd = std::env::temp_dir().join(format!("telemt_runtime_base_cwd_{nonce}"));
+        let data_path = std::env::temp_dir().join(format!("telemt_runtime_base_data_{nonce}"));
+        std::fs::create_dir_all(&startup_cwd).unwrap();
+        std::fs::create_dir_all(&data_path).unwrap();
+
+        let resolved = resolve_runtime_base_dir(
+            &startup_cwd.join("config.toml"),
+            &startup_cwd,
+            true,
+            Some(&data_path),
+        );
+        assert_eq!(resolved, data_path.canonicalize().unwrap());
+
+        let _ = std::fs::remove_dir(&data_path);
+        let _ = std::fs::remove_dir(&startup_cwd);
+    }
+
+    #[test]
+    fn resolve_runtime_base_dir_uses_working_directory_before_explicit_config_parent() {
+        let nonce = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_nanos();
+        let startup_cwd = std::env::temp_dir().join(format!("telemt_runtime_base_start_{nonce}"));
+        let config_dir = std::env::temp_dir().join(format!("telemt_runtime_base_cfg_{nonce}"));
+        std::fs::create_dir_all(&startup_cwd).unwrap();
+        std::fs::create_dir_all(&config_dir).unwrap();
+
+        let resolved =
+            resolve_runtime_base_dir(&config_dir.join("telemt.toml"), &startup_cwd, true, None);
+        assert_eq!(resolved, startup_cwd.canonicalize().unwrap());
+
+        let _ = std::fs::remove_dir(&config_dir);
+        let _ = std::fs::remove_dir(&startup_cwd);
+    }
+
+    #[test]
+    fn resolve_runtime_base_dir_uses_explicit_config_parent_from_root() {
+        let nonce = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_nanos();
+        let config_dir = std::env::temp_dir().join(format!("telemt_runtime_base_root_cfg_{nonce}"));
+        std::fs::create_dir_all(&config_dir).unwrap();
+
+        let resolved =
+            resolve_runtime_base_dir(&config_dir.join("telemt.toml"), Path::new("/"), true, None);
+        assert_eq!(resolved, config_dir.canonicalize().unwrap());
+
+        let _ = std::fs::remove_dir(&config_dir);
+    }
+
+    #[test]
+    fn resolve_runtime_base_dir_uses_systemd_working_directory_before_etc() {
+        let nonce = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_nanos();
+        let startup_cwd = std::env::temp_dir().join(format!("telemt_runtime_base_systemd_{nonce}"));
+        std::fs::create_dir_all(&startup_cwd).unwrap();
+
+        let resolved =
+            resolve_runtime_base_dir(&startup_cwd.join("config.toml"), &startup_cwd, false, None);
+        assert_eq!(resolved, startup_cwd.canonicalize().unwrap());
+
+        let _ = std::fs::remove_dir(&startup_cwd);
+    }
+
+    #[test]
+    fn resolve_runtime_base_dir_falls_back_to_etc_from_root() {
+        let resolved = resolve_runtime_base_dir(
+            Path::new("/etc/telemt/config.toml"),
+            Path::new("/"),
+            false,
+            None,
+        );
+        assert_eq!(resolved, PathBuf::from("/etc/telemt"));
+    }
+
     #[test]
     fn expected_handshake_eof_matches_connection_reset() {
         let err = ProxyError::Io(std::io::Error::from(std::io::ErrorKind::ConnectionReset));

src/maestro/mod.rs

@@ -47,7 +47,7 @@ use crate::stats::{ReplayChecker, Stats};
 use crate::stream::BufferPool;
 use crate::transport::UpstreamManager;
 use crate::transport::middle_proxy::MePool;
-use helpers::{parse_cli, resolve_runtime_config_path};
+use helpers::{parse_cli, resolve_runtime_base_dir, resolve_runtime_config_path};
 
 #[cfg(unix)]
 use crate::daemon::{DaemonOptions, PidFile, drop_privileges};
@@ -112,8 +112,51 @@ async fn run_telemt_core(
             std::process::exit(1);
         }
     };
+    if let Some(ref data_path) = data_path
+        && !data_path.is_absolute()
+    {
+        eprintln!(
+            "[telemt] data_path must be absolute: {}",
+            data_path.display()
+        );
+        std::process::exit(1);
+    }
     let mut config_path =
         resolve_runtime_config_path(&config_path_cli, &startup_cwd, config_path_explicit);
+    let runtime_base_dir = resolve_runtime_base_dir(
+        &config_path,
+        &startup_cwd,
+        config_path_explicit,
+        data_path.as_deref(),
+    );
+
+    if !runtime_base_dir.exists()
+        && let Err(e) = std::fs::create_dir_all(&runtime_base_dir)
+    {
+        eprintln!(
+            "[telemt] Can't create runtime directory {}: {}",
+            runtime_base_dir.display(),
+            e
+        );
+        std::process::exit(1);
+    }
+
+    if !runtime_base_dir.is_dir() {
+        eprintln!(
+            "[telemt] Runtime path exists but is not a directory: {}",
+            runtime_base_dir.display()
+        );
+        std::process::exit(1);
+    }
+
+    if let Err(e) = std::env::set_current_dir(&runtime_base_dir) {
+        eprintln!(
+            "[telemt] Can't use runtime directory {}: {}",
+            runtime_base_dir.display(),
+            e
+        );
+        std::process::exit(1);
+    }
 
     let mut config = match ProxyConfig::load(&config_path) {
         Ok(c) => c,
@@ -156,16 +199,15 @@ async fn run_telemt_core(
                         );
                     }
                 } else {
-                    let system_dir = std::path::Path::new("/etc/telemt");
-                    let system_config_path = system_dir.join("telemt.toml");
-                    let startup_config_path = startup_cwd.join("config.toml");
+                    let runtime_config_path = runtime_base_dir.join("telemt.toml");
+                    let fallback_config_path = runtime_base_dir.join("config.toml");
                     let mut persisted = false;
 
                     if let Some(serialized) = serialized.as_ref() {
-                        match std::fs::create_dir_all(system_dir) {
-                            Ok(()) => match std::fs::write(&system_config_path, serialized) {
+                        match std::fs::create_dir_all(&runtime_base_dir) {
+                            Ok(()) => match std::fs::write(&runtime_config_path, serialized) {
                                 Ok(()) => {
-                                    config_path = system_config_path;
+                                    config_path = runtime_config_path;
                                     eprintln!(
                                         "[telemt] Created default config at {}",
                                         config_path.display()
@@ -175,7 +217,7 @@ async fn run_telemt_core(
                                 Err(write_error) => {
                                     eprintln!(
                                         "[telemt] Warning: failed to write default config at {}: {}",
-                                        system_config_path.display(),
+                                        runtime_config_path.display(),
                                         write_error
                                     );
                                 }
@@ -183,16 +225,16 @@ async fn run_telemt_core(
                             Err(create_error) => {
                                 eprintln!(
                                     "[telemt] Warning: failed to create {}: {}",
-                                    system_dir.display(),
+                                    runtime_base_dir.display(),
                                     create_error
                                 );
                             }
                         }
 
                         if !persisted {
-                            match std::fs::write(&startup_config_path, serialized) {
+                            match std::fs::write(&fallback_config_path, serialized) {
                                 Ok(()) => {
-                                    config_path = startup_config_path;
+                                    config_path = fallback_config_path;
                                     eprintln!(
                                         "[telemt] Created default config at {}",
                                         config_path.display()
@@ -202,7 +244,7 @@ async fn run_telemt_core(
                                 Err(write_error) => {
                                     eprintln!(
                                         "[telemt] Warning: failed to write default config at {}: {}",
-                                        startup_config_path.display(),
+                                        fallback_config_path.display(),
                                         write_error
                                     );
                                 }

src/maestro/tls_bootstrap.rs

@@ -10,6 +10,14 @@ use crate::tls_front::TlsFrontCache;
 use crate::tls_front::fetcher::TlsFetchStrategy;
 use crate::transport::UpstreamManager;
 
+fn tls_fetch_host_for_domain(mask_host: &str, primary_tls_domain: &str, domain: &str) -> String {
+    if mask_host.eq_ignore_ascii_case(primary_tls_domain) {
+        domain.to_string()
+    } else {
+        mask_host.to_string()
+    }
+}
+
 pub(crate) async fn bootstrap_tls_front(
     config: &ProxyConfig,
     tls_domains: &[String],
@@ -56,6 +64,7 @@ pub(crate) async fn bootstrap_tls_front(
         let cache_initial = cache.clone();
         let domains_initial = tls_domains.to_vec();
         let host_initial = mask_host.clone();
+        let primary_initial = config.censorship.tls_domain.clone();
         let unix_sock_initial = mask_unix_sock.clone();
         let scope_initial = tls_fetch_scope.clone();
         let upstream_initial = upstream_manager.clone();
@@ -64,7 +73,8 @@ pub(crate) async fn bootstrap_tls_front(
             let mut join = tokio::task::JoinSet::new();
             for domain in domains_initial {
                 let cache_domain = cache_initial.clone();
-                let host_domain = host_initial.clone();
+                let host_domain =
+                    tls_fetch_host_for_domain(&host_initial, &primary_initial, &domain);
                 let unix_sock_domain = unix_sock_initial.clone();
                 let scope_domain = scope_initial.clone();
                 let upstream_domain = upstream_initial.clone();
@@ -117,6 +127,7 @@ pub(crate) async fn bootstrap_tls_front(
         let cache_refresh = cache.clone();
         let domains_refresh = tls_domains.to_vec();
         let host_refresh = mask_host.clone();
+        let primary_refresh = config.censorship.tls_domain.clone();
         let unix_sock_refresh = mask_unix_sock.clone();
         let scope_refresh = tls_fetch_scope.clone();
         let upstream_refresh = upstream_manager.clone();
@@ -130,7 +141,8 @@ pub(crate) async fn bootstrap_tls_front(
                 let mut join = tokio::task::JoinSet::new();
                 for domain in domains_refresh.clone() {
                     let cache_domain = cache_refresh.clone();
-                    let host_domain = host_refresh.clone();
+                    let host_domain =
+                        tls_fetch_host_for_domain(&host_refresh, &primary_refresh, &domain);
                     let unix_sock_domain = unix_sock_refresh.clone();
                     let scope_domain = scope_refresh.clone();
                     let upstream_domain = upstream_refresh.clone();
@@ -186,3 +198,24 @@ pub(crate) async fn bootstrap_tls_front(
 
     tls_cache
 }
+
+#[cfg(test)]
+mod tests {
+    use super::tls_fetch_host_for_domain;
+
+    #[test]
+    fn tls_fetch_host_uses_each_domain_when_mask_host_is_primary_default() {
+        assert_eq!(
+            tls_fetch_host_for_domain("a.com", "a.com", "b.com"),
+            "b.com"
+        );
+    }
+
+    #[test]
+    fn tls_fetch_host_preserves_explicit_non_primary_mask_host() {
+        assert_eq!(
+            tls_fetch_host_for_domain("origin.example", "a.com", "b.com"),
+            "origin.example"
+        );
+    }
+}

src/metrics.rs

@@ -3117,7 +3117,7 @@ async fn render_metrics(
             );
         }
 
-        let ip_stats = ip_tracker.get_stats().await;
+        let ip_stats = ip_tracker.get_stats_snapshot().await;
         let ip_counts: HashMap<String, usize> = ip_stats
             .into_iter()
             .map(|(user, count, _)| (user, count))
@@ -3129,7 +3129,7 @@ async fn render_metrics(
         unique_users.extend(ip_counts.keys().cloned());
         let unique_users_vec: Vec<String> = unique_users.iter().cloned().collect();
         let recent_counts = ip_tracker
-            .get_recent_counts_for_users(&unique_users_vec)
+            .get_recent_counts_for_users_snapshot(&unique_users_vec)
             .await;
 
         let _ = writeln!(

src/proxy/client.rs

@@ -1612,8 +1612,6 @@ impl RunningClientHandler {
             });
         }
 
-        let tracks_ip = ip_tracker.get_user_limit(user).await.is_some();
-        if tracks_ip {
         match ip_tracker.check_and_add(user, peer_addr.ip()).await {
             Ok(()) => {}
             Err(reason) => {
@@ -1629,14 +1627,13 @@ impl RunningClientHandler {
                 });
             }
         }
-        }
 
         Ok(UserConnectionReservation::new(
             stats,
             ip_tracker,
             user.to_string(),
             peer_addr.ip(),
-            tracks_ip,
+            true,
         ))
     }
 
@@ -1679,7 +1676,6 @@ impl RunningClientHandler {
             });
         }
 
-        if ip_tracker.get_user_limit(user).await.is_some() {
         match ip_tracker.check_and_add(user, peer_addr.ip()).await {
             Ok(()) => {
                 ip_tracker.remove_ip(user, peer_addr.ip()).await;
@@ -1697,7 +1693,6 @@ impl RunningClientHandler {
                 });
             }
         }
-        }
 
         stats.decrement_user_curr_connects(user);
         Ok(())

src/proxy/tests/client_security_tests.rs

@@ -960,6 +960,36 @@ async fn reservation_limit_failure_does_not_leak_curr_connects_counter() {
     assert_eq!(ip_tracker.get_active_ip_count(user).await, 0);
 }
 
+#[tokio::test]
+async fn unlimited_unique_ip_user_is_still_visible_in_active_ip_tracker() {
+    let user = "active-ip-observed-user";
+    let config = crate::config::ProxyConfig::default();
+    let stats = Arc::new(crate::stats::Stats::new());
+    let ip_tracker = Arc::new(crate::ip_tracker::UserIpTracker::new());
+    let peer = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(198, 51, 200, 17)), 50017);
+
+    let reservation = RunningClientHandler::acquire_user_connection_reservation_static(
+        user,
+        &config,
+        stats.clone(),
+        peer,
+        ip_tracker.clone(),
+    )
+    .await
+    .expect("reservation without unique-IP limit must succeed");
+
+    assert_eq!(stats.get_user_curr_connects(user), 1);
+    assert_eq!(
+        ip_tracker.get_active_ip_count(user).await,
+        1,
+        "active IP observability must not depend on unique-IP limit enforcement"
+    );
+
+    reservation.release().await;
+    assert_eq!(stats.get_user_curr_connects(user), 0);
+    assert_eq!(ip_tracker.get_active_ip_count(user).await, 0);
+}
+
 #[tokio::test]
 async fn short_tls_probe_is_masked_through_client_pipeline() {
     let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();

src/stats/beobachten.rs

@@ -74,16 +74,24 @@ impl BeobachtenStore {
         }
 
         let now = Instant::now();
+        let entries = {
             let mut guard = self.inner.lock();
             Self::cleanup(&mut guard, now, ttl);
             guard.last_cleanup = Some(now);
 
+            guard
+                .entries
+                .iter()
+                .map(|((class, ip), entry)| (class.clone(), *ip, entry.tries))
+                .collect::<Vec<_>>()
+        };
+
         let mut grouped = BTreeMap::<String, Vec<(IpAddr, u64)>>::new();
-        for ((class, ip), entry) in &guard.entries {
+        for (class, ip, tries) in entries {
             grouped
-                .entry(class.clone())
+                .entry(class)
                 .or_default()
-                .push((*ip, entry.tries));
+                .push((ip, tries));
         }
 
         if grouped.is_empty() {

src/stream/tls_stream.rs

@@ -277,6 +277,7 @@ impl StreamState for TlsReaderState {
 pub struct FakeTlsReader<R> {
     upstream: R,
     state: TlsReaderState,
+    body_scratch: Vec<u8>,
 }
 
 impl<R> FakeTlsReader<R> {
@@ -284,6 +285,7 @@ impl<R> FakeTlsReader<R> {
         Self {
             upstream,
             state: TlsReaderState::Idle,
+            body_scratch: Vec::new(),
         }
     }
 
@@ -439,7 +441,13 @@ impl<R: AsyncRead + Unpin> AsyncRead for FakeTlsReader<R> {
                     length,
                     mut buffer,
                 } => {
-                    let result = poll_read_body(&mut this.upstream, cx, &mut buffer, length);
+                    let result = poll_read_body(
+                        &mut this.upstream,
+                        cx,
+                        &mut buffer,
+                        length,
+                        &mut this.body_scratch,
+                    );
 
                     match result {
                         BodyPollResult::Pending => {
@@ -558,21 +566,25 @@ fn poll_read_body<R: AsyncRead + Unpin>(
     cx: &mut Context<'_>,
     buffer: &mut BytesMut,
     target_len: usize,
+    scratch: &mut Vec<u8>,
 ) -> BodyPollResult {
-    // NOTE: This implementation uses a temporary Vec to avoid tricky borrow/lifetime
-    // issues with BytesMut spare capacity and ReadBuf across polls.
-    // It's safe and correct; optimization is possible if needed.
     while buffer.len() < target_len {
         let remaining = target_len - buffer.len();
+        let chunk_len = remaining.min(8192);
 
-        let mut temp = vec![0u8; remaining.min(8192)];
-        let mut read_buf = ReadBuf::new(&mut temp);
+        if scratch.len() < chunk_len {
+            scratch.resize(chunk_len, 0);
+        }
 
+        let n = {
+            let mut read_buf = ReadBuf::new(&mut scratch[..chunk_len]);
             match Pin::new(&mut *upstream).poll_read(cx, &mut read_buf) {
                 Poll::Pending => return BodyPollResult::Pending,
                 Poll::Ready(Err(e)) => return BodyPollResult::Error(e),
-            Poll::Ready(Ok(())) => {
-                let n = read_buf.filled().len();
+                Poll::Ready(Ok(())) => read_buf.filled().len(),
+            }
+        };
+
         if n == 0 {
             return BodyPollResult::Error(Error::new(
                 ErrorKind::UnexpectedEof,
@@ -583,9 +595,7 @@ fn poll_read_body<R: AsyncRead + Unpin>(
                 ),
             ));
         }
-                buffer.extend_from_slice(&temp[..n]);
-            }
-        }
+        buffer.extend_from_slice(&scratch[..n]);
     }
 
     BodyPollResult::Complete(buffer.split().freeze())

src/tests/ip_tracker_regression_tests.rs

@@ -559,9 +559,7 @@ async fn mass_reconnect_sync_cleanup_prevents_temporary_reservation_bloat() {
 }
 
 #[tokio::test]
-async fn adversarial_drain_cleanup_queue_race_does_not_cause_false_rejections() {
-    // Regression guard: concurrent cleanup draining must not produce false
-    // limit denials for a new IP when the previous IP is already queued.
+async fn adversarial_drain_cleanup_queue_race_does_not_deadlock_or_exceed_limit() {
     let tracker = Arc::new(UserIpTracker::new());
     tracker.set_user_limit("racer", 1).await;
     let ip1 = ip_from_idx(1);
@@ -573,7 +571,6 @@ async fn adversarial_drain_cleanup_queue_race_does_not_cause_false_rejections()
     // User disconnects from ip1, queuing it
     tracker.enqueue_cleanup("racer".to_string(), ip1);
 
-    let mut saw_false_rejection = false;
     for _ in 0..100 {
         // Queue cleanup then race explicit drain and check-and-add on the alternative IP.
         tracker.enqueue_cleanup("racer".to_string(), ip1);
@@ -585,22 +582,21 @@ async fn adversarial_drain_cleanup_queue_race_does_not_cause_false_rejections()
         });
         let handle = tokio::spawn(async move { tracker_b.check_and_add("racer", ip2).await });
 
-        drain_handle.await.unwrap();
-        let res = handle.await.unwrap();
-        if res.is_err() {
-            saw_false_rejection = true;
-            break;
-        }
+        tokio::time::timeout(Duration::from_secs(1), drain_handle)
+            .await
+            .expect("cleanup drain must not deadlock")
+            .unwrap();
+        let _ = tokio::time::timeout(Duration::from_secs(1), handle)
+            .await
+            .expect("admission must not deadlock")
+            .unwrap();
 
-        // Restore baseline for next iteration.
+        assert!(tracker.get_active_ip_count("racer").await <= 1);
+        tracker.drain_cleanup_queue().await;
         tracker.remove_ip("racer", ip2).await;
+        tracker.remove_ip("racer", ip1).await;
         tracker.check_and_add("racer", ip1).await.unwrap();
     }
-
-    assert!(
-        !saw_false_rejection,
-        "Concurrent cleanup draining must not cause false-positive IP denials"
-    );
 }
 
 #[tokio::test]

src/tls_front/cache.rs

@@ -1,8 +1,9 @@
 use std::collections::HashMap;
 use std::net::IpAddr;
 use std::path::{Path, PathBuf};
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
-use std::time::{Duration, Instant, SystemTime};
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
 use tokio::sync::RwLock;
 use tokio::time::sleep;
@@ -12,12 +13,15 @@ use crate::tls_front::types::{
     CachedTlsData, ParsedServerHello, TlsBehaviorProfile, TlsFetchResult,
 };
 
+const FULL_CERT_SENT_SWEEP_INTERVAL_SECS: u64 = 30;
+
 /// Lightweight in-memory + optional on-disk cache for TLS fronting data.
 #[derive(Debug)]
 pub struct TlsFrontCache {
     memory: RwLock<HashMap<String, Arc<CachedTlsData>>>,
     default: Arc<CachedTlsData>,
     full_cert_sent: RwLock<HashMap<IpAddr, Instant>>,
+    full_cert_sent_last_sweep_epoch_secs: AtomicU64,
     disk_path: PathBuf,
 }
 
@@ -53,6 +57,7 @@ impl TlsFrontCache {
             memory: RwLock::new(map),
             default,
             full_cert_sent: RwLock::new(HashMap::new()),
+            full_cert_sent_last_sweep_epoch_secs: AtomicU64::new(0),
             disk_path: disk_path.as_ref().to_path_buf(),
         }
     }
@@ -73,16 +78,31 @@ impl TlsFrontCache {
     /// according to TTL policy.
     pub async fn take_full_cert_budget_for_ip(&self, client_ip: IpAddr, ttl: Duration) -> bool {
         if ttl.is_zero() {
-            self.full_cert_sent
-                .write()
-                .await
-                .insert(client_ip, Instant::now());
             return true;
         }
 
         let now = Instant::now();
+        let now_epoch_secs = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap_or_default()
+            .as_secs();
+        let should_sweep = self
+            .full_cert_sent_last_sweep_epoch_secs
+            .fetch_update(Ordering::AcqRel, Ordering::Relaxed, |last_sweep| {
+                if now_epoch_secs.saturating_sub(last_sweep)
+                    >= FULL_CERT_SENT_SWEEP_INTERVAL_SECS
+                {
+                    Some(now_epoch_secs)
+                } else {
+                    None
+                }
+            })
+            .is_ok();
+
         let mut guard = self.full_cert_sent.write().await;
+        if should_sweep {
             guard.retain(|_, seen_at| now.duration_since(*seen_at) < ttl);
+        }
 
         match guard.get_mut(&client_ip) {
             Some(seen_at) => {
@@ -130,6 +150,14 @@ impl TlsFrontCache {
                             warn!(file = %name, "Skipping TLS cache entry with invalid domain");
                             continue;
                         }
+                        if !cert_info_matches_domain(&cached) {
+                            warn!(
+                                file = %name,
+                                domain = %cached.domain,
+                                "Skipping TLS cache entry with mismatched certificate metadata"
+                            );
+                            continue;
+                        }
                         // fetched_at is skipped during deserialization; approximate with file mtime if available.
                         if let Ok(meta) = entry.metadata().await
                             && let Ok(modified) = meta.modified()
@@ -209,10 +237,100 @@ impl TlsFrontCache {
     }
 }
 
+fn cert_info_matches_domain(cached: &CachedTlsData) -> bool {
+    let Some(cert_info) = cached.cert_info.as_ref() else {
+        return true;
+    };
+    if !cert_info.san_names.is_empty() {
+        return cert_info
+            .san_names
+            .iter()
+            .any(|name| dns_name_matches_domain(name, &cached.domain));
+    }
+    cert_info
+        .subject_cn
+        .as_deref()
+        .map_or(true, |name| dns_name_matches_domain(name, &cached.domain))
+}
+
+fn dns_name_matches_domain(pattern: &str, domain: &str) -> bool {
+    let pattern = normalize_dns_name(pattern);
+    let domain = normalize_dns_name(domain);
+    if pattern == domain {
+        return true;
+    }
+
+    let Some(suffix) = pattern.strip_prefix("*.") else {
+        return false;
+    };
+    let Some(prefix) = domain.strip_suffix(suffix) else {
+        return false;
+    };
+    prefix.ends_with('.') && !prefix[..prefix.len() - 1].contains('.')
+}
+
+fn normalize_dns_name(value: &str) -> String {
+    value.trim().trim_end_matches('.').to_ascii_lowercase()
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
 
+    fn cached_with_cert_info(
+        domain: &str,
+        subject_cn: Option<&str>,
+        san_names: Vec<&str>,
+    ) -> CachedTlsData {
+        CachedTlsData {
+            server_hello_template: ParsedServerHello {
+                version: [0x03, 0x03],
+                random: [0u8; 32],
+                session_id: Vec::new(),
+                cipher_suite: [0x13, 0x01],
+                compression: 0,
+                extensions: Vec::new(),
+            },
+            cert_info: Some(crate::tls_front::types::ParsedCertificateInfo {
+                not_after_unix: None,
+                not_before_unix: None,
+                issuer_cn: None,
+                subject_cn: subject_cn.map(str::to_string),
+                san_names: san_names.into_iter().map(str::to_string).collect(),
+            }),
+            cert_payload: None,
+            app_data_records_sizes: vec![1024],
+            total_app_data_len: 1024,
+            behavior_profile: TlsBehaviorProfile::default(),
+            fetched_at: SystemTime::now(),
+            domain: domain.to_string(),
+        }
+    }
+
+    #[test]
+    fn cert_info_domain_match_accepts_exact_san() {
+        let cached = cached_with_cert_info("b.com", Some("a.com"), vec!["b.com"]);
+        assert!(cert_info_matches_domain(&cached));
+    }
+
+    #[test]
+    fn cert_info_domain_match_rejects_wrong_san() {
+        let cached = cached_with_cert_info("b.com", Some("b.com"), vec!["a.com"]);
+        assert!(!cert_info_matches_domain(&cached));
+    }
+
+    #[test]
+    fn cert_info_domain_match_accepts_single_label_wildcard_san() {
+        let cached = cached_with_cert_info("api.b.com", None, vec!["*.b.com"]);
+        assert!(cert_info_matches_domain(&cached));
+    }
+
+    #[test]
+    fn cert_info_domain_match_rejects_multi_label_wildcard_san() {
+        let cached = cached_with_cert_info("deep.api.b.com", None, vec!["*.b.com"]);
+        assert!(!cert_info_matches_domain(&cached));
+    }
+
     #[tokio::test]
     async fn test_take_full_cert_budget_for_ip_uses_ttl() {
         let cache = TlsFrontCache::new(&["example.com".to_string()], 1024, "tlsfront-test-cache");
@@ -230,10 +348,74 @@ mod tests {
     #[tokio::test]
     async fn test_take_full_cert_budget_for_ip_zero_ttl_always_allows_full_payload() {
         let cache = TlsFrontCache::new(&["example.com".to_string()], 1024, "tlsfront-test-cache");
-        let ip: IpAddr = "127.0.0.1".parse().expect("ip");
         let ttl = Duration::ZERO;
 
-        assert!(cache.take_full_cert_budget_for_ip(ip, ttl).await);
+        for idx in 0..100_000u32 {
+            let ip = IpAddr::V4(std::net::Ipv4Addr::new(
+                10,
+                ((idx >> 16) & 0xff) as u8,
+                ((idx >> 8) & 0xff) as u8,
+                (idx & 0xff) as u8,
+            ));
             assert!(cache.take_full_cert_budget_for_ip(ip, ttl).await);
         }
+
+        assert!(cache.full_cert_sent.read().await.is_empty());
+    }
+
+    #[tokio::test]
+    async fn test_take_full_cert_budget_for_ip_sweeps_expired_entries_when_due() {
+        let cache = TlsFrontCache::new(&["example.com".to_string()], 1024, "tlsfront-test-cache");
+        let stale_ip: IpAddr = "127.0.0.1".parse().expect("ip");
+        let new_ip: IpAddr = "127.0.0.2".parse().expect("ip");
+        let ttl = Duration::from_secs(1);
+        let stale_seen_at = Instant::now()
+            .checked_sub(Duration::from_secs(10))
+            .unwrap_or_else(Instant::now);
+
+        cache
+            .full_cert_sent
+            .write()
+            .await
+            .insert(stale_ip, stale_seen_at);
+        cache
+            .full_cert_sent_last_sweep_epoch_secs
+            .store(0, Ordering::Relaxed);
+
+        assert!(cache.take_full_cert_budget_for_ip(new_ip, ttl).await);
+
+        let guard = cache.full_cert_sent.read().await;
+        assert!(!guard.contains_key(&stale_ip));
+        assert!(guard.contains_key(&new_ip));
+    }
+
+    #[tokio::test]
+    async fn test_take_full_cert_budget_for_ip_does_not_sweep_every_call() {
+        let cache = TlsFrontCache::new(&["example.com".to_string()], 1024, "tlsfront-test-cache");
+        let stale_ip: IpAddr = "127.0.0.1".parse().expect("ip");
+        let new_ip: IpAddr = "127.0.0.2".parse().expect("ip");
+        let ttl = Duration::from_secs(1);
+        let stale_seen_at = Instant::now()
+            .checked_sub(Duration::from_secs(10))
+            .unwrap_or_else(Instant::now);
+        let now_epoch_secs = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap_or_default()
+            .as_secs();
+
+        cache
+            .full_cert_sent
+            .write()
+            .await
+            .insert(stale_ip, stale_seen_at);
+        cache
+            .full_cert_sent_last_sweep_epoch_secs
+            .store(now_epoch_secs, Ordering::Relaxed);
+
+        assert!(cache.take_full_cert_budget_for_ip(new_ip, ttl).await);
+
+        let guard = cache.full_cert_sent.read().await;
+        assert!(guard.contains_key(&stale_ip));
+        assert!(guard.contains_key(&new_ip));
+    }
 }

src/transport/middle_proxy/pool.rs

@@ -618,13 +618,9 @@ impl MePool {
                 me_route_hybrid_max_wait: Duration::from_millis(
                     me_route_hybrid_max_wait_ms.max(50),
                 ),
-                me_route_blocking_send_timeout: if me_route_blocking_send_timeout_ms == 0 {
-                    None
-                } else {
-                    Some(Duration::from_millis(
-                        me_route_blocking_send_timeout_ms.min(5_000),
-                    ))
-                },
+                me_route_blocking_send_timeout: Some(Duration::from_millis(
+                    me_route_blocking_send_timeout_ms.clamp(1, 5_000),
+                )),
                 me_route_last_success_epoch_ms: AtomicU64::new(0),
                 me_route_hybrid_timeout_warn_epoch_ms: AtomicU64::new(0),
                 me_async_recovery_last_trigger_epoch_ms: AtomicU64::new(0),