Merge pull request #745 from telemt/flow

API PATCH fixes + No IP tracking with disabled unique-IP limits + Bound hot-path pressure in ME Relay and Handshake + Bounded ME Route fairness and IP-Cleanup-Backlog + Bound relay queues by bytes

Cargo.lock

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

Cargo.toml

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

src/proxy/tests/client_security_tests.rs

@@ -281,13 +281,8 @@ async fn user_connection_reservation_drop_enqueues_cleanup_synchronously() {
     assert_eq!(ip_tracker.get_active_ip_count(&user).await, 1);
     assert_eq!(stats.get_user_curr_connects(&user), 1);
 
-    let reservation = UserConnectionReservation::new(
+    let reservation =
-        stats.clone(),
+        UserConnectionReservation::new(stats.clone(), ip_tracker.clone(), user.clone(), ip, true);
-        ip_tracker.clone(),
-        user.clone(),
-        ip,
-        true,
-    );
 
     // Drop the reservation synchronously without any tokio::spawn/await yielding!
     drop(reservation);

src/transport/middle_proxy/registry.rs

@@ -8,8 +8,8 @@ use dashmap::DashMap;
 use tokio::sync::mpsc::error::TrySendError;
 use tokio::sync::{Mutex, Semaphore, mpsc};
 
-use super::{MeResponse, RouteBytePermit};
 use super::codec::WriterCommand;
+use super::{MeResponse, RouteBytePermit};
 
 const ROUTE_BACKPRESSURE_BASE_TIMEOUT_MS: u64 = 25;
 const ROUTE_BACKPRESSURE_HIGH_TIMEOUT_MS: u64 = 120;
@@ -129,7 +129,10 @@ impl ConnRegistry {
         )
     }
 
-    fn with_route_limits(route_channel_capacity: usize, route_byte_permits_per_conn: usize) -> Self {
+    fn with_route_limits(
+        route_channel_capacity: usize,
+        route_byte_permits_per_conn: usize,
+    ) -> Self {
         let start = rand::random::<u64>() | 1;
         let route_channel_capacity = route_channel_capacity.max(1);
         Self {
@@ -209,9 +212,10 @@ impl ConnRegistry {
         let id = self.next_id.fetch_add(1, Ordering::Relaxed);
         let (tx, rx) = mpsc::channel(self.route_channel_capacity);
         self.routing.map.insert(id, tx);
-        self.routing
+        self.routing.byte_budget.insert(
-            .byte_budget
+            id,
-            .insert(id, Arc::new(Semaphore::new(self.route_byte_permits_per_conn)));
+            Arc::new(Semaphore::new(self.route_byte_permits_per_conn)),
+        );
         (id, rx)
     }
 
@@ -287,8 +291,7 @@ impl ConnRegistry {
                 .map_err(|_| RouteResult::QueueFullHigh)?,
             Some(timeout_ms) => {
                 let acquire = semaphore.acquire_many_owned(permits);
-                match tokio::time::timeout(Duration::from_millis(timeout_ms.max(1)), acquire)
+                match tokio::time::timeout(Duration::from_millis(timeout_ms.max(1)), acquire).await
-                    .await
                 {
                     Ok(Ok(permit)) => permit,
                     Ok(Err(_)) => return Err(RouteResult::ChannelClosed),