DC to Client fine tuning

src/transport/middle_proxy/pool.rs

@@ -183,6 +183,7 @@ pub struct MePool {
     pub(super) me_writer_pick_mode: AtomicU8,
     pub(super) me_writer_pick_sample_size: AtomicU8,
     pub(super) me_socks_kdf_policy: AtomicU8,
+    pub(super) me_reader_route_data_wait_ms: Arc<AtomicU64>,
     pub(super) me_route_no_writer_mode: AtomicU8,
     pub(super) me_route_no_writer_wait: Duration,
     pub(super) me_route_inline_recovery_attempts: u32,
@@ -287,6 +288,7 @@ impl MePool {
         me_route_backpressure_base_timeout_ms: u64,
         me_route_backpressure_high_timeout_ms: u64,
         me_route_backpressure_high_watermark_pct: u8,
+        me_reader_route_data_wait_ms: u64,
         me_health_interval_ms_unhealthy: u64,
         me_health_interval_ms_healthy: u64,
         me_warn_rate_limit_ms: u64,
@@ -460,6 +462,7 @@ impl MePool {
             me_writer_pick_mode: AtomicU8::new(me_writer_pick_mode.as_u8()),
             me_writer_pick_sample_size: AtomicU8::new(me_writer_pick_sample_size.clamp(2, 4)),
             me_socks_kdf_policy: AtomicU8::new(me_socks_kdf_policy.as_u8()),
+            me_reader_route_data_wait_ms: Arc::new(AtomicU64::new(me_reader_route_data_wait_ms)),
             me_route_no_writer_mode: AtomicU8::new(me_route_no_writer_mode.as_u8()),
             me_route_no_writer_wait: Duration::from_millis(me_route_no_writer_wait_ms),
             me_route_inline_recovery_attempts,
@@ -650,9 +653,12 @@ impl MePool {
         route_backpressure_base_timeout_ms: u64,
         route_backpressure_high_timeout_ms: u64,
         route_backpressure_high_watermark_pct: u8,
+        reader_route_data_wait_ms: u64,
     ) {
         self.me_socks_kdf_policy
             .store(socks_kdf_policy.as_u8(), Ordering::Relaxed);
+        self.me_reader_route_data_wait_ms
+            .store(reader_route_data_wait_ms, Ordering::Relaxed);
         self.registry.update_route_backpressure_policy(
             route_backpressure_base_timeout_ms,
             route_backpressure_high_timeout_ms,

src/transport/middle_proxy/pool_writer.rs

@@ -208,6 +208,7 @@ impl MePool {
         let keepalive_jitter_signal = self.me_keepalive_jitter;
         let cancel_reader_token = cancel.clone();
         let cancel_ping_token = cancel_ping.clone();
+        let reader_route_data_wait_ms = self.me_reader_route_data_wait_ms.clone();
 
         tokio::spawn(async move {
             let res = reader_loop(
@@ -225,6 +226,7 @@ impl MePool {
                 writer_id,
                 degraded.clone(),
                 rtt_ema_ms_x10.clone(),
+                reader_route_data_wait_ms,
                 cancel_reader_token.clone(),
             )
             .await;

src/transport/middle_proxy/reader.rs

@@ -1,7 +1,7 @@
 use std::collections::HashMap;
 use std::io::ErrorKind;
 use std::sync::Arc;
-use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering};
 use std::time::Instant;
 
 use bytes::{Bytes, BytesMut};
@@ -35,6 +35,7 @@ pub(crate) async fn reader_loop(
     _writer_id: u64,
     degraded: Arc<AtomicBool>,
     writer_rtt_ema_ms_x10: Arc<AtomicU32>,
+    reader_route_data_wait_ms: Arc<AtomicU64>,
     cancel: CancellationToken,
 ) -> Result<()> {
     let mut raw = enc_leftover;
@@ -57,17 +58,14 @@ pub(crate) async fn reader_loop(
 
         let blocks = raw.len() / 16 * 16;
         if blocks > 0 {
+            let mut chunk = raw.split_to(blocks);
             let mut new_iv = [0u8; 16];
-            new_iv.copy_from_slice(&raw[blocks - 16..blocks]);
-
-            let mut chunk = vec![0u8; blocks];
-            chunk.copy_from_slice(&raw[..blocks]);
+            new_iv.copy_from_slice(&chunk[blocks - 16..blocks]);
             AesCbc::new(dk, div)
-                .decrypt_in_place(&mut chunk)
+                .decrypt_in_place(&mut chunk[..])
                 .map_err(|e| ProxyError::Crypto(format!("{e}")))?;
             div = new_iv;
             dec.extend_from_slice(&chunk);
-            let _ = raw.split_to(blocks);
         }
 
         while dec.len() >= 12 {
@@ -85,7 +83,7 @@ pub(crate) async fn reader_loop(
                 break;
             }
 
-            let frame = dec.split_to(fl);
+            let frame = dec.split_to(fl).freeze();
             let pe = fl - 4;
             let ec = u32::from_le_bytes(frame[pe..pe + 4].try_into().unwrap());
             let actual_crc = rpc_crc(crc_mode, &frame[..pe]);
@@ -111,21 +109,27 @@ pub(crate) async fn reader_loop(
             }
             expected_seq = expected_seq.wrapping_add(1);
 
-            let payload = &frame[8..pe];
+            let payload = frame.slice(8..pe);
             if payload.len() < 4 {
                 continue;
             }
 
             let pt = u32::from_le_bytes(payload[0..4].try_into().unwrap());
-            let body = &payload[4..];
+            let body = payload.slice(4..);
 
             if pt == RPC_PROXY_ANS_U32 && body.len() >= 12 {
                 let flags = u32::from_le_bytes(body[0..4].try_into().unwrap());
                 let cid = u64::from_le_bytes(body[4..12].try_into().unwrap());
-                let data = Bytes::copy_from_slice(&body[12..]);
+                let data = body.slice(12..);
                 trace!(cid, flags, len = data.len(), "RPC_PROXY_ANS");
 
-                let routed = reg.route_nowait(cid, MeResponse::Data { flags, data }).await;
+                let data_wait_ms = reader_route_data_wait_ms.load(Ordering::Relaxed);
+                let routed = if data_wait_ms == 0 {
+                    reg.route_nowait(cid, MeResponse::Data { flags, data }).await
+                } else {
+                    reg.route_with_timeout(cid, MeResponse::Data { flags, data }, data_wait_ms)
+                        .await
+                };
                 if !matches!(routed, RouteResult::Routed) {
                     match routed {
                         RouteResult::NoConn => stats.increment_me_route_drop_no_conn(),

src/transport/middle_proxy/registry.rs

@@ -231,6 +231,57 @@ impl ConnRegistry {
         }
     }
 
+    pub async fn route_with_timeout(
+        &self,
+        id: u64,
+        resp: MeResponse,
+        timeout_ms: u64,
+    ) -> RouteResult {
+        if timeout_ms == 0 {
+            return self.route_nowait(id, resp).await;
+        }
+
+        let tx = {
+            let inner = self.inner.read().await;
+            inner.map.get(&id).cloned()
+        };
+
+        let Some(tx) = tx else {
+            return RouteResult::NoConn;
+        };
+
+        match tx.try_send(resp) {
+            Ok(()) => RouteResult::Routed,
+            Err(TrySendError::Closed(_)) => RouteResult::ChannelClosed,
+            Err(TrySendError::Full(resp)) => {
+                let high_watermark_pct = self
+                    .route_backpressure_high_watermark_pct
+                    .load(Ordering::Relaxed)
+                    .clamp(1, 100);
+                let used = self.route_channel_capacity.saturating_sub(tx.capacity());
+                let used_pct = if self.route_channel_capacity == 0 {
+                    100
+                } else {
+                    (used.saturating_mul(100) / self.route_channel_capacity) as u8
+                };
+                let high_profile = used_pct >= high_watermark_pct;
+                let timeout_dur = Duration::from_millis(timeout_ms.max(1));
+
+                match tokio::time::timeout(timeout_dur, tx.send(resp)).await {
+                    Ok(Ok(())) => RouteResult::Routed,
+                    Ok(Err(_)) => RouteResult::ChannelClosed,
+                    Err(_) => {
+                        if high_profile {
+                            RouteResult::QueueFullHigh
+                        } else {
+                            RouteResult::QueueFullBase
+                        }
+                    }
+                }
+            }
+        }
+    }
+
     pub async fn bind_writer(
         &self,
         conn_id: u64,