feat(tunnel): drain loop, upload coalesce, block_stun config, tuning

Tunnel-node:
- Drain loop: keep reading until buffer empty (max 1s), accumulates
  up to 2MB+ per drain for streaming video (was 100KB)
- Upload size logging for debugging
- 512KB reader buffer (was 64KB)
- LONGPOLL_DEADLINE 4s

Client:
- INFLIGHT_ACTIVE 4 (was 10) to prevent semaphore exhaustion
- Upload loop-read in initial path (1s max, accumulates fat uploads)
- Fast-path 200ms coalesce loop (was single 20ms read)
- 32KB download threshold for elevation (prevents keep-alive sessions
  like Telegram from over-elevating)
- consecutive_data no longer resets on single empties
- block_stun config (default true) with Android UI toggle
- 512KB client read buffer

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: vahidlazio

android/app/src/main/java/com/therealaleph/mhrv/ConfigStore.kt

@@ -108,6 +108,8 @@ data class MhrvConfig(
     val coalesceMaxMs: Int = 1000,
     /** Block QUIC (UDP/443). QUIC over TCP tunnel causes meltdown. */
     val blockQuic: Boolean = true,
+    /** Block STUN/TURN ports (3478/5349/19302). Forces WebRTC TCP fallback. */
+    val blockStun: Boolean = true,
     val upstreamSocks5: String = "",
 
     /**
@@ -231,6 +233,7 @@ data class MhrvConfig(
             if (coalesceStepMs != 10) put("coalesce_step_ms", coalesceStepMs)
             if (coalesceMaxMs != 1000) put("coalesce_max_ms", coalesceMaxMs)
             put("block_quic", blockQuic)
+            put("block_stun", blockStun)
             if (upstreamSocks5.isNotBlank()) {
                 put("upstream_socks5", upstreamSocks5.trim())
             }
@@ -344,6 +347,7 @@ object ConfigStore {
         if (cfg.coalesceStepMs != defaults.coalesceStepMs) obj.put("coalesce_step_ms", cfg.coalesceStepMs)
         if (cfg.coalesceMaxMs != defaults.coalesceMaxMs) obj.put("coalesce_max_ms", cfg.coalesceMaxMs)
         if (cfg.blockQuic != defaults.blockQuic) obj.put("block_quic", cfg.blockQuic)
+        if (cfg.blockStun != defaults.blockStun) obj.put("block_stun", cfg.blockStun)
         if (cfg.upstreamSocks5.isNotBlank()) obj.put("upstream_socks5", cfg.upstreamSocks5)
         if (cfg.passthroughHosts.isNotEmpty()) obj.put("passthrough_hosts", JSONArray().apply { cfg.passthroughHosts.forEach { put(it) } })
         if (cfg.tunnelDoh != defaults.tunnelDoh) obj.put("tunnel_doh", cfg.tunnelDoh)
@@ -449,6 +453,7 @@ object ConfigStore {
             coalesceStepMs = obj.optInt("coalesce_step_ms", 10),
             coalesceMaxMs = obj.optInt("coalesce_max_ms", 1000),
             blockQuic = obj.optBoolean("block_quic", true),
+            blockStun = obj.optBoolean("block_stun", true),
             upstreamSocks5 = obj.optString("upstream_socks5", ""),
             passthroughHosts = obj.optJSONArray("passthrough_hosts")?.let { arr ->
                 buildList { for (i in 0 until arr.length()) add(arr.optString(i)) }

android/app/src/main/java/com/therealaleph/mhrv/ui/HomeScreen.kt

@@ -1288,6 +1288,28 @@ private fun AdvancedSettings(
             )
         }
 
+        // Block STUN/TURN toggle
+        Row(
+            verticalAlignment = Alignment.CenterVertically,
+            modifier = Modifier.fillMaxWidth(),
+        ) {
+            Column(modifier = Modifier.weight(1f)) {
+                Text(
+                    "Block STUN/TURN",
+                    style = MaterialTheme.typography.bodyMedium,
+                )
+                Text(
+                    "Reject STUN/TURN ports (3478/5349/19302). Forces WebRTC apps (Meet, WhatsApp) to TCP fallback — instant connect.",
+                    style = MaterialTheme.typography.bodySmall,
+                    color = MaterialTheme.colorScheme.onSurfaceVariant,
+                )
+            }
+            Switch(
+                checked = cfg.blockStun,
+                onCheckedChange = { onChange(cfg.copy(blockStun = it)) },
+            )
+        }
+
         // Block DoH toggle
         Row(
             verticalAlignment = Alignment.CenterVertically,

src/config.rs

@@ -202,6 +202,13 @@ pub struct Config {
     /// flag lets users who care about consistency over peak speed
     /// opt out of QUIC at the source rather than discovering its
     /// failure modes later. Issue #213.
+    /// Block STUN/TURN UDP ports (3478, 5349, 19302) at the SOCKS5 listener.
+    /// Forces WebRTC apps (Google Meet, Discord, WhatsApp) to fall back to
+    /// TCP TURN on port 443, skipping the 10-30s UDP ICE timeout. Default
+    /// true — TCP fallback works for all tested apps and connects instantly.
+    #[serde(default = "default_block_stun")]
+    pub block_stun: bool,
+
     #[serde(default = "default_block_quic")]
     pub block_quic: bool,
     /// When true, suppress the random `_pad` field that v1.8.0+ adds
@@ -497,6 +504,7 @@ fn default_tunnel_doh() -> bool { true }
 /// Default for `block_quic`: `true`. QUIC over the TCP-based tunnel
 /// causes TCP-over-TCP meltdown (<1 Mbps). Browsers fall back to
 /// HTTPS/TCP within seconds of the silent UDP drop. Issue #793.
+fn default_block_stun() -> bool { true }
 fn default_block_quic() -> bool { true }
 
 /// Default for `block_doh`: `true` (browser DoH is rejected so the

src/proxy_server.rs

@@ -241,6 +241,7 @@ pub struct RewriteCtx {
     /// callers fall back to TCP/HTTPS. See config.rs `block_quic` for
     /// the trade-off. Issue #213.
     pub block_quic: bool,
+    pub block_stun: bool,
     /// If true, route DoH CONNECTs around the Apps Script tunnel via
     /// plain TCP. Default false via `Config::tunnel_doh = true` (flipped
     /// in v1.9.0, issue #468). See `DEFAULT_DOH_HOSTS` and
@@ -507,6 +508,7 @@ impl ProxyServer {
             youtube_via_relay: config.youtube_via_relay,
             passthrough_hosts: config.passthrough_hosts.clone(),
             block_quic: config.block_quic,
+            block_stun: config.block_stun,
             bypass_doh: !config.tunnel_doh,
             block_doh: config.block_doh,
             bypass_doh_hosts: config.bypass_doh_hosts.clone(),
@@ -943,7 +945,7 @@ async fn handle_socks5_client(
     // Reject STUN/TURN UDP ports immediately so WebRTC (Meet,
     // Telegram calls) skips UDP ICE candidates and falls back to
     // TCP TURN on :443 without waiting for a timeout.
-    if matches!(port, 3478 | 5349 | 19302) {
+    if rewrite_ctx.block_stun && matches!(port, 3478 | 5349 | 19302) {
         tracing::info!("SOCKS5 CONNECT -> {}:{} (STUN/TURN blocked, forcing TCP fallback)", host, port);
         sock.write_all(&[0x05, 0x05, 0x00, 0x01, 0, 0, 0, 0, 0, 0])
             .await?;

src/tunnel_client.rs

@@ -67,7 +67,7 @@ const INFLIGHT_OPTIMIST: usize = 2;
 
 /// Maximum pipeline depth when data is actively flowing. Ramps up on
 /// data-bearing replies, drops back to IDLE after consecutive empties.
-const INFLIGHT_ACTIVE: usize = 10;
+const INFLIGHT_ACTIVE: usize = 4;
 
 /// How many consecutive empty replies before dropping from active to idle depth.
 const INFLIGHT_COOLDOWN: u32 = 3;
@@ -1426,7 +1426,7 @@ async fn tunnel_loop(
     mut pending_client_data: Option<Bytes>,
 ) -> std::io::Result<()> {
     let (mut reader, mut writer) = sock.split();
-    const READ_CHUNK: usize = 65536;
+    const READ_CHUNK: usize = 512 * 1024;
     let mut buf = BytesMut::with_capacity(READ_CHUNK);
     let mut consecutive_empty = 0u32;
     let mut buffered_upload: Option<Bytes> = None;
@@ -1479,7 +1479,26 @@ async fn tunnel_loop(
                     Ok(Ok(0)) => break,
                     Ok(Ok(n)) => {
                         consecutive_empty = 0;
-                        Some(extract_bytes(&mut buf, n))
+                        let mut data = extract_bytes(&mut buf, n);
+                        // Loop-read: accumulate more upload data (up to 1s)
+                        let deadline = Instant::now() + Duration::from_secs(1);
+                        loop {
+                            if Instant::now() >= deadline { break; }
+                            buf.reserve(READ_CHUNK);
+                            match tokio::time::timeout(Duration::from_millis(20), reader.read_buf(&mut buf)).await {
+                                Ok(Ok(0)) => { upload_closed = true; break; }
+                                Ok(Ok(n)) => {
+                                    let extra = extract_bytes(&mut buf, n);
+                                    let mut combined = bytes::BytesMut::with_capacity(data.len() + extra.len());
+                                    combined.extend_from_slice(&data);
+                                    combined.extend_from_slice(&extra);
+                                    data = combined.freeze();
+                                }
+                                Ok(Err(_)) => { upload_closed = true; break; }
+                                Err(_) => break, // no more data
+                            }
+                        }
+                        Some(data)
                     }
                     Ok(Err(_)) => break,
                     Err(_) => None,
@@ -1804,23 +1823,27 @@ async fn tunnel_loop(
                 // outside the depth cap so uploads aren't blocked
                 // behind polls waiting at the tunnel-node.
                 if buffered_upload.is_some() && inflight.len() >= max_inflight && inflight.len() < max_inflight + 4 {
-                    // Brief coalesce: wait 20ms for more client data
-                    // to arrive so we batch multiple small writes into
-                    // one op instead of one per read.
-                    buf.reserve(READ_CHUNK);
-                    match tokio::time::timeout(Duration::from_millis(20), reader.read_buf(&mut buf)).await {
-                        Ok(Ok(0)) => { break; }
-                        Ok(Ok(n)) => {
-                            let extra = extract_bytes(&mut buf, n);
-                            let merged = buffered_upload.take().unwrap();
-                            let mut combined = bytes::BytesMut::with_capacity(merged.len() + extra.len());
-                            combined.extend_from_slice(&merged);
-                            combined.extend_from_slice(&extra);
-                            buffered_upload = Some(combined.freeze());
+                    // Loop-coalesce: keep reading client data up to
+                    // 200ms so we pack a fatter upload per op.
+                    let coalesce_deadline = Instant::now() + Duration::from_millis(200);
+                    loop {
+                        if Instant::now() >= coalesce_deadline { break; }
+                        buf.reserve(READ_CHUNK);
+                        match tokio::time::timeout(Duration::from_millis(20), reader.read_buf(&mut buf)).await {
+                            Ok(Ok(0)) => { upload_closed = true; break; }
+                            Ok(Ok(n)) => {
+                                let extra = extract_bytes(&mut buf, n);
+                                let merged = buffered_upload.take().unwrap();
+                                let mut combined = bytes::BytesMut::with_capacity(merged.len() + extra.len());
+                                combined.extend_from_slice(&merged);
+                                combined.extend_from_slice(&extra);
+                                buffered_upload = Some(combined.freeze());
+                            }
+                            Ok(Err(_)) => { upload_closed = true; break; }
+                            Err(_) => break, // no more data
                         }
-                        Ok(Err(_)) => { break; }
-                        Err(_) => {} // timeout — no more data, send what we have
                     }
+                    if upload_closed { break; }
                     let data = buffered_upload.take().unwrap();
                     let seq = next_send_seq;
                     next_send_seq += 1;

tunnel-node/src/main.rs

@@ -241,7 +241,7 @@ fn create_udpgw_session() -> ManagedSession {
 }
 
 async fn reader_task(mut reader: impl AsyncRead + Unpin, session: Arc<SessionInner>) {
-    let mut buf = vec![0u8; 65536];
+    let mut buf = vec![0u8; 512 * 1024];
     loop {
         match reader.read(&mut buf).await {
             Ok(0) => {
@@ -896,6 +896,7 @@ async fn handle_batch(
                             };
                             if !bytes.is_empty() {
                                 had_writes_or_connects = true;
+                                tracing::info!("session {} upload {}KB", &sid[..sid.len().min(8)], bytes.len() / 1024);
                                 let mut w = inner.writer.lock().await;
                                 let _ = w.write_all(&bytes).await;
                                 let _ = w.flush().await;
@@ -1090,16 +1091,31 @@ async fn handle_batch(
         // short of the cliff; deferred sessions drain on the next poll.
         let mut remaining_budget: usize = BATCH_RESPONSE_BUDGET;
         for (i, sid, inner, seq) in &tcp_drains {
-            let (data, eof) = drain_now(inner, remaining_budget).await;
-            let drained = data.len();
-            if eof {
+            // Drain in a loop: keep reading until the buffer is empty
+            // so we catch data that arrives during the drain itself.
+            let mut all_data = Vec::new();
+            let mut final_eof = false;
+            let drain_deadline = Instant::now() + Duration::from_secs(1);
+            loop {
+                let (data, eof) = drain_now(inner, remaining_budget.saturating_sub(all_data.len())).await;
+                if eof { final_eof = true; }
+                if data.is_empty() { break; }
+                all_data.extend_from_slice(&data);
+                if final_eof || all_data.len() >= remaining_budget { break; }
+                if Instant::now() >= drain_deadline { break; }
+                // Brief yield to let reader_task finish its current read
+                tokio::task::yield_now().await;
+            }
+            let drained = all_data.len();
+            if drained > 0 {
+                tracing::info!("session {} drained {}KB", &sid[..sid.len().min(8)], drained / 1024);
+            }
+            if final_eof {
                 tcp_eof_sids.push(sid.clone());
             }
-            results.push((*i, tcp_drain_response(sid.clone(), data, eof, *seq)));
+            results.push((*i, tcp_drain_response(sid.clone(), all_data, final_eof, *seq)));
             remaining_budget = remaining_budget.saturating_sub(drained);
             if remaining_budget == 0 {
-                // Budget exhausted; remaining sessions in `tcp_drains` keep
-                // their buffered data and pick up next batch.
                 break;
             }
         }