From 919b13b166c609f245967fa8212a64fd10d44c3a Mon Sep 17 00:00:00 2001
From: yyoyoian-pixel <yyoyoian@gmail.com>
Date: Sat, 16 May 2026 16:46:08 +0200
Subject: [PATCH] feat(tunnel): pipelined polls with adaptive depth, wseq
 ordering, STUN blocking (#1115)

feat(tunnel): pipelined full-tunnel polls, ordered writes, and STUN blocking

Merged trusted PR #1115 by @yyoyoian-pixel after local verification and a small maintainer fix on the PR branch.

---
Answered via LLM, Supervised @therealaleph
---
 android/app/src/main/AndroidManifest.xml      |   1 +
 .../java/com/therealaleph/mhrv/ConfigStore.kt |   5 +
 .../com/therealaleph/mhrv/MhrvVpnService.kt   |  16 +
 .../main/java/com/therealaleph/mhrv/Native.kt |   7 +
 .../therealaleph/mhrv/PipelineDebugOverlay.kt | 174 ++++
 .../com/therealaleph/mhrv/ui/HomeScreen.kt    | 121 +++
 scripts/bench-pipeline.sh                     | 140 +++
 src/android_jni.rs                            |  16 +-
 src/bin/ui.rs                                 |  20 +
 src/config.rs                                 |   8 +
 src/domain_fronter.rs                         |   6 +
 src/proxy_server.rs                           |  13 +
 src/tunnel_client.rs                          | 958 ++++++++++++++----
 tunnel-node/src/main.rs                       | 146 ++-
 14 files changed, 1422 insertions(+), 209 deletions(-)
 create mode 100644 android/app/src/main/java/com/therealaleph/mhrv/PipelineDebugOverlay.kt
 create mode 100755 scripts/bench-pipeline.sh

diff --git a/android/app/src/main/AndroidManifest.xml b/android/app/src/main/AndroidManifest.xml
index 4d74ca5..cce8e61 100644
--- a/android/app/src/main/AndroidManifest.xml
+++ b/android/app/src/main/AndroidManifest.xml
@@ -17,6 +17,7 @@
       prompt.
     -->
     <uses-permission android:name="android.permission.QUERY_ALL_PACKAGES" />
+    <uses-permission android:name="android.permission.SYSTEM_ALERT_WINDOW" />
 
     <!--
       App-launcher visibility filter. Complements QUERY_ALL_PACKAGES:
diff --git a/android/app/src/main/java/com/therealaleph/mhrv/ConfigStore.kt b/android/app/src/main/java/com/therealaleph/mhrv/ConfigStore.kt
index 0b98273..d00d59e 100644
--- a/android/app/src/main/java/com/therealaleph/mhrv/ConfigStore.kt
+++ b/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)) }
diff --git a/android/app/src/main/java/com/therealaleph/mhrv/MhrvVpnService.kt b/android/app/src/main/java/com/therealaleph/mhrv/MhrvVpnService.kt
index 053edc2..59aebb7 100644
--- a/android/app/src/main/java/com/therealaleph/mhrv/MhrvVpnService.kt
+++ b/android/app/src/main/java/com/therealaleph/mhrv/MhrvVpnService.kt
@@ -35,6 +35,7 @@ class MhrvVpnService : VpnService() {
     private var proxyHandle: Long = 0L
     private var tun2proxyThread: Thread? = null
     private val tun2proxyRunning = AtomicBoolean(false)
+    private var debugOverlay: PipelineDebugOverlay? = null
 
     // Idempotency guard. teardown() is reachable from three paths:
     //   1. ACTION_STOP onStartCommand branch (background thread)
@@ -149,6 +150,7 @@ class MhrvVpnService : VpnService() {
             Log.i(TAG, "PROXY_ONLY mode: listeners up, skipping VpnService/TUN")
             VpnState.setProxyHandle(proxyHandle)
             VpnState.setRunning(true)
+            showDebugOverlay()
             return
         }
 
@@ -314,6 +316,16 @@ class MhrvVpnService : VpnService() {
         // a failed-to-establish run.
         VpnState.setProxyHandle(proxyHandle)
         VpnState.setRunning(true)
+        showDebugOverlay()
+    }
+
+    private fun showDebugOverlay() {
+        if (debugOverlay != null) return
+        if (!android.provider.Settings.canDrawOverlays(this)) {
+            Log.w(TAG, "overlay permission not granted — skipping debug overlay")
+            return
+        }
+        debugOverlay = PipelineDebugOverlay(this).also { it.show() }
     }
 
     /**
@@ -434,6 +446,10 @@ class MhrvVpnService : VpnService() {
             Log.w(TAG, "tun2proxy thread still alive after join timeout — proceeding anyway")
         }
 
+        // Hide debug overlay before flipping UI state.
+        debugOverlay?.hide()
+        debugOverlay = null
+
         // Flip UI state last — the button reverts to Connect only after
         // the native-side cleanup actually happened, not optimistically.
         VpnState.setProxyHandle(0L)
diff --git a/android/app/src/main/java/com/therealaleph/mhrv/Native.kt b/android/app/src/main/java/com/therealaleph/mhrv/Native.kt
index 798d523..40482af 100644
--- a/android/app/src/main/java/com/therealaleph/mhrv/Native.kt
+++ b/android/app/src/main/java/com/therealaleph/mhrv/Native.kt
@@ -110,6 +110,13 @@ object Native {
      */
     external fun statsJson(handle: Long): String
 
+    /**
+     * Pipeline debug overlay snapshot. Returns a JSON blob with elevated
+     * session count, batch semaphore usage, and recent ramp/drop events.
+     * Temporary — for debugging pipeline behavior on-device.
+     */
+    external fun pipelineDebugJson(): String
+
     /**
      * Start tun2proxy via its CLI args C API (`tun2proxy_run_with_cli_args`).
      * Resolved at runtime via dlsym from libtun2proxy.so — no fork needed.
diff --git a/android/app/src/main/java/com/therealaleph/mhrv/PipelineDebugOverlay.kt b/android/app/src/main/java/com/therealaleph/mhrv/PipelineDebugOverlay.kt
new file mode 100644
index 0000000..c53ac4e
--- /dev/null
+++ b/android/app/src/main/java/com/therealaleph/mhrv/PipelineDebugOverlay.kt
@@ -0,0 +1,174 @@
+package com.therealaleph.mhrv
+
+import android.content.Context
+import android.graphics.Color
+import android.graphics.PixelFormat
+import android.os.Handler
+import android.os.Looper
+import android.util.TypedValue
+import android.view.Gravity
+import android.view.MotionEvent
+import android.view.View
+import android.view.WindowManager
+import android.widget.LinearLayout
+import android.widget.TextView
+import org.json.JSONObject
+
+/**
+ * Transparent system overlay showing pipeline debug stats.
+ * Draggable, semi-transparent, shown on top of all apps.
+ * Temporary — remove when pipelining is validated.
+ */
+class PipelineDebugOverlay(private val context: Context) {
+
+    private val wm = context.getSystemService(Context.WINDOW_SERVICE) as WindowManager
+    private val handler = Handler(Looper.getMainLooper())
+    private var root: View? = null
+
+    private lateinit var tvElevated: TextView
+    private lateinit var tvBatches: TextView
+    private lateinit var tvEvents: TextView
+
+    private val pollInterval = 500L
+
+    fun show() {
+        if (root != null) return
+
+        val dp = { px: Int ->
+            TypedValue.applyDimension(TypedValue.COMPLEX_UNIT_DIP, px.toFloat(), context.resources.displayMetrics).toInt()
+        }
+
+        val layout = LinearLayout(context).apply {
+            orientation = LinearLayout.VERTICAL
+            setBackgroundColor(Color.argb(160, 0, 0, 0))
+            setPadding(dp(8), dp(6), dp(8), dp(6))
+        }
+
+        val titleTv = TextView(context).apply {
+            text = "Pipeline Debug"
+            setTextColor(Color.argb(220, 100, 255, 100))
+            textSize = 11f
+        }
+        layout.addView(titleTv)
+
+        tvElevated = TextView(context).apply {
+            setTextColor(Color.WHITE)
+            textSize = 10f
+        }
+        layout.addView(tvElevated)
+
+        tvBatches = TextView(context).apply {
+            setTextColor(Color.WHITE)
+            textSize = 10f
+        }
+        layout.addView(tvBatches)
+
+        tvEvents = TextView(context).apply {
+            setTextColor(Color.argb(200, 200, 200, 200))
+            textSize = 9f
+            maxLines = 8
+        }
+        layout.addView(tvEvents)
+
+        val params = WindowManager.LayoutParams(
+            WindowManager.LayoutParams.WRAP_CONTENT,
+            WindowManager.LayoutParams.WRAP_CONTENT,
+            WindowManager.LayoutParams.TYPE_APPLICATION_OVERLAY,
+            WindowManager.LayoutParams.FLAG_NOT_FOCUSABLE or
+                WindowManager.LayoutParams.FLAG_NOT_TOUCH_MODAL,
+            PixelFormat.TRANSLUCENT,
+        ).apply {
+            gravity = Gravity.TOP or Gravity.START
+            x = dp(8)
+            y = dp(80)
+        }
+
+        // Draggable
+        var startX = 0
+        var startY = 0
+        var startTouchX = 0f
+        var startTouchY = 0f
+        layout.setOnTouchListener { _, event ->
+            when (event.action) {
+                MotionEvent.ACTION_DOWN -> {
+                    startX = params.x
+                    startY = params.y
+                    startTouchX = event.rawX
+                    startTouchY = event.rawY
+                    true
+                }
+                MotionEvent.ACTION_MOVE -> {
+                    params.x = startX + (event.rawX - startTouchX).toInt()
+                    params.y = startY + (event.rawY - startTouchY).toInt()
+                    wm.updateViewLayout(layout, params)
+                    true
+                }
+                else -> false
+            }
+        }
+
+        root = layout
+        wm.addView(layout, params)
+        schedulePoll()
+    }
+
+    fun hide() {
+        handler.removeCallbacksAndMessages(null)
+        root?.let {
+            try { wm.removeView(it) } catch (_: Throwable) {}
+        }
+        root = null
+    }
+
+    private fun schedulePoll() {
+        handler.postDelayed(::poll, pollInterval)
+    }
+
+    private fun poll() {
+        if (root == null) return
+        Thread {
+            try {
+                val json = Native.pipelineDebugJson()
+                handler.post { applyJson(json) }
+            } catch (_: Throwable) {}
+            schedulePoll()
+        }.start()
+    }
+
+    private fun applyJson(json: String) {
+        if (root == null) return
+        try {
+            if (json.isNotBlank()) {
+                val obj = JSONObject(json)
+                val elevated = obj.optInt("elevated", 0)
+                val maxElev = obj.optInt("max_elevated", 0)
+                val batches = obj.optInt("active_batches", 0)
+                val maxBatch = obj.optInt("max_batch_slots", 0)
+
+                val sessions = obj.optInt("active_sessions", 0)
+                tvElevated.text = "Sessions: $sessions  Elevated: $elevated / $maxElev"
+                tvBatches.text = "Batches: $batches / $maxBatch"
+
+                val sessArr = obj.optJSONArray("sessions")
+                val sessLines = if (sessArr != null && sessArr.length() > 0) {
+                    (0 until sessArr.length()).joinToString("\n") { i ->
+                        val s = sessArr.getJSONObject(i)
+                        val sid = s.optString("sid", "?")
+                        val d = s.optInt("depth", 0)
+                        val inf = s.optInt("inflight", 0)
+                        val e = if (s.optBoolean("elevated", false)) " E" else ""
+                        "$sid d=$d f=$inf$e"
+                    }
+                } else ""
+
+                val arr = obj.optJSONArray("events")
+                val evtLines = if (arr != null && arr.length() > 0) {
+                    val start = maxOf(0, arr.length() - 5)
+                    (start until arr.length()).joinToString("\n") { arr.getString(it) }
+                } else ""
+
+                tvEvents.text = listOf(sessLines, evtLines).filter { it.isNotEmpty() }.joinToString("\n---\n")
+            }
+        } catch (_: Throwable) {}
+    }
+}
diff --git a/android/app/src/main/java/com/therealaleph/mhrv/ui/HomeScreen.kt b/android/app/src/main/java/com/therealaleph/mhrv/ui/HomeScreen.kt
index d228b72..b06bc5c 100644
--- a/android/app/src/main/java/com/therealaleph/mhrv/ui/HomeScreen.kt
+++ b/android/app/src/main/java/com/therealaleph/mhrv/ui/HomeScreen.kt
@@ -491,6 +491,7 @@ fun HomeScreen(
             // client-side estimate only sees what this device relayed,
             // not what other devices on the same deployment consumed.
             UsageTodayCard()
+            PipelineDebugCard()
 
             CollapsibleSection(title = stringResource(R.string.sec_live_logs), initiallyExpanded = false) {
                 LiveLogPane()
@@ -1287,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,
@@ -1645,6 +1668,104 @@ private fun UsageRow(label: String, value: String) {
     }
 }
 
+@Composable
+private fun PipelineDebugCard() {
+    val isRunning by VpnState.isRunning.collectAsState()
+    if (!isRunning) return
+
+    var json by remember { mutableStateOf("") }
+    LaunchedEffect(isRunning) {
+        if (!isRunning) return@LaunchedEffect
+        while (true) {
+            val result = withContext(Dispatchers.IO) {
+                runCatching { Native.pipelineDebugJson() }
+            }
+            json = result.getOrDefault("")
+            if (result.isFailure) {
+                android.util.Log.e("PipeDbg", "pipelineDebugJson failed", result.exceptionOrNull())
+            }
+            delay(500)
+        }
+    }
+
+    val obj = remember(json) {
+        if (json.isBlank()) null
+        else runCatching { JSONObject(json) }.getOrNull()
+    }
+    if (obj == null) return
+
+    val elevated = obj.optInt("elevated", 0)
+    val maxElevated = obj.optInt("max_elevated", 0)
+    val batches = obj.optInt("active_batches", 0)
+    val maxBatches = obj.optInt("max_batch_slots", 0)
+    val events = remember(json) {
+        val arr = obj.optJSONArray("events") ?: return@remember emptyList<String>()
+        (0 until arr.length()).map { arr.getString(it) }
+    }
+
+    Spacer(Modifier.height(8.dp))
+    ElevatedCard(modifier = Modifier.fillMaxWidth()) {
+        Column(
+            modifier = Modifier.padding(12.dp),
+            verticalArrangement = Arrangement.spacedBy(4.dp),
+        ) {
+            Text(
+                "Pipeline Debug",
+                style = MaterialTheme.typography.titleSmall,
+            )
+            Row(
+                modifier = Modifier.fillMaxWidth(),
+                horizontalArrangement = Arrangement.SpaceBetween,
+            ) {
+                Text("Elevated", style = MaterialTheme.typography.bodySmall)
+                Text(
+                    "$elevated / $maxElevated",
+                    style = MaterialTheme.typography.bodySmall,
+                    fontFamily = FontFamily.Monospace,
+                )
+            }
+            Row(
+                modifier = Modifier.fillMaxWidth(),
+                horizontalArrangement = Arrangement.SpaceBetween,
+            ) {
+                Text("Batches in-flight", style = MaterialTheme.typography.bodySmall)
+                Text(
+                    "$batches / $maxBatches",
+                    style = MaterialTheme.typography.bodySmall,
+                    fontFamily = FontFamily.Monospace,
+                )
+            }
+            if (events.isNotEmpty()) {
+                Spacer(Modifier.height(4.dp))
+                Text("Events", style = MaterialTheme.typography.labelSmall)
+                Box(
+                    modifier = Modifier
+                        .fillMaxWidth()
+                        .heightIn(max = 150.dp)
+                        .clip(RoundedCornerShape(4.dp))
+                        .background(MaterialTheme.colorScheme.surfaceVariant)
+                        .padding(6.dp)
+                ) {
+                    val listState = rememberLazyListState()
+                    LaunchedEffect(events.size) {
+                        if (events.isNotEmpty()) listState.animateScrollToItem(events.size - 1)
+                    }
+                    LazyColumn(state = listState) {
+                        items(events) { ev ->
+                            Text(
+                                ev,
+                                style = MaterialTheme.typography.bodySmall,
+                                fontFamily = FontFamily.Monospace,
+                                fontSize = 10.sp,
+                            )
+                        }
+                    }
+                }
+            }
+        }
+    }
+}
+
 private fun fmtBytes(b: Long): String {
     val k = 1024L
     val m = k * k
diff --git a/scripts/bench-pipeline.sh b/scripts/bench-pipeline.sh
new file mode 100755
index 0000000..65fd2ab
--- /dev/null
+++ b/scripts/bench-pipeline.sh
@@ -0,0 +1,140 @@
+#!/usr/bin/env bash
+#
+# bench-pipeline.sh — compare throughput: serial (depth=1) vs pipelined (depth=10)
+#
+# Builds mhrv-rs twice (patching the INFLIGHT_ACTIVE constant), runs each
+# as a local SOCKS5 proxy, downloads through the full tunnel, reports.
+#
+# Usage:
+#   ./scripts/bench-pipeline.sh [CONFIG_FILE]
+#
+# Default: config.json
+
+set -euo pipefail
+
+CONFIG="${1:-config.json}"
+RUNS=3
+SOCKS_PORT=18088
+HTTP_PORT=18087
+TEST_URL="https://speed.cloudflare.com/__down?bytes=5000000"
+SRC="src/tunnel_client.rs"
+TMPDIR_BENCH=$(mktemp -d)
+
+cleanup() {
+    rm -rf "$TMPDIR_BENCH"
+    kill $PROXY_PID 2>/dev/null || true
+    # Restore original constant
+    sed -i '' "s/^const INFLIGHT_ACTIVE: usize = [0-9]*/const INFLIGHT_ACTIVE: usize = 10/" "$SRC" 2>/dev/null || true
+}
+trap cleanup EXIT
+
+if [ ! -f "$CONFIG" ]; then
+    echo "ERROR: Config not found: $CONFIG"
+    exit 1
+fi
+
+echo "╔══════════════════════════════════════════════╗"
+echo "║     Pipeline Throughput Benchmark            ║"
+echo "╠══════════════════════════════════════════════╣"
+echo "║ Config:    $CONFIG"
+echo "║ Test URL:  $TEST_URL"
+echo "║ Runs:      $RUNS per mode"
+echo "╚══════════════════════════════════════════════╝"
+echo ""
+
+# Write a temp config with our ports
+TEMP_CONFIG="$TMPDIR_BENCH/config.json"
+python3 -c "
+import json
+with open('$CONFIG') as f:
+    c = json.load(f)
+c['listen_port'] = $HTTP_PORT
+c['socks5_port'] = $SOCKS_PORT
+c['log_level'] = 'warn'
+with open('$TEMP_CONFIG', 'w') as f:
+    json.dump(c, f)
+"
+
+run_test() {
+    local label="$1"
+    local binary="$2"
+    echo "━━━ $label ━━━"
+
+    $binary -c "$TEMP_CONFIG" &
+    PROXY_PID=$!
+    sleep 3
+
+    if ! kill -0 $PROXY_PID 2>/dev/null; then
+        echo "  ERROR: Proxy failed to start"
+        return
+    fi
+
+    # Wait for proxy
+    for attempt in $(seq 1 15); do
+        if curl -s --socks5-hostname localhost:$SOCKS_PORT --connect-timeout 5 -o /dev/null https://www.google.com 2>/dev/null; then
+            break
+        fi
+        sleep 1
+    done
+
+    local total_bytes=0
+    local total_time=0
+
+    for i in $(seq 1 $RUNS); do
+        local result
+        result=$(curl -s --socks5-hostname localhost:$SOCKS_PORT \
+            -o /dev/null \
+            -w '%{size_download} %{time_total} %{speed_download}' \
+            --connect-timeout 30 \
+            --max-time 90 \
+            "$TEST_URL" 2>/dev/null || echo "0 999 0")
+
+        local bytes time_s speed
+        bytes=$(echo "$result" | awk '{print $1}')
+        time_s=$(echo "$result" | awk '{print $2}')
+        speed=$(echo "$result" | awk '{printf "%.0f", $3/1024}')
+
+        total_bytes=$((total_bytes + ${bytes%.*}))
+        total_time=$(echo "$total_time + $time_s" | bc)
+
+        printf "  Run %d: %.1fs  %s KB/s\n" "$i" "$time_s" "$speed"
+    done
+
+    local avg_speed avg_time
+    avg_speed=$(echo "scale=1; $total_bytes / $total_time / 1024" | bc 2>/dev/null || echo "0")
+    avg_time=$(echo "scale=1; $total_time / $RUNS" | bc 2>/dev/null || echo "0")
+    printf "  ➜ Average: %s KB/s  (%.1fs per download)\n\n" "$avg_speed" "$avg_time"
+
+    kill $PROXY_PID 2>/dev/null || true
+    wait $PROXY_PID 2>/dev/null || true
+    sleep 1
+
+    echo "$label|$avg_speed|$avg_time" >> "$TMPDIR_BENCH/results.txt"
+}
+
+# Build serial (depth=1)
+echo "Building serial mode (INFLIGHT_ACTIVE=1)..."
+sed -i '' "s/^const INFLIGHT_ACTIVE: usize = [0-9]*/const INFLIGHT_ACTIVE: usize = 1/" "$SRC"
+cargo build --release 2>&1 | tail -1
+cp target/release/mhrv-rs "$TMPDIR_BENCH/mhrv-serial"
+
+# Build pipelined (depth=10)
+echo "Building pipelined mode (INFLIGHT_ACTIVE=10)..."
+sed -i '' "s/^const INFLIGHT_ACTIVE: usize = [0-9]*/const INFLIGHT_ACTIVE: usize = 10/" "$SRC"
+cargo build --release 2>&1 | tail -1
+cp target/release/mhrv-rs "$TMPDIR_BENCH/mhrv-pipelined"
+
+echo ""
+
+# Run tests
+run_test "Serial (depth=1)" "$TMPDIR_BENCH/mhrv-serial"
+run_test "Pipelined (depth=10)" "$TMPDIR_BENCH/mhrv-pipelined"
+
+# Summary
+echo "╔══════════════════════════════════════════════╗"
+echo "║               RESULTS                       ║"
+echo "╠══════════════════════════════════════════════╣"
+while IFS='|' read -r label speed time; do
+    printf "║  %-25s %6s KB/s  %5ss\n" "$label" "$speed" "$time"
+done < "$TMPDIR_BENCH/results.txt"
+echo "╚══════════════════════════════════════════════╝"
diff --git a/src/android_jni.rs b/src/android_jni.rs
index 91b4fa5..a551e83 100644
--- a/src/android_jni.rs
+++ b/src/android_jni.rs
@@ -199,7 +199,7 @@ pub extern "system" fn Java_com_therealaleph_mhrv_Native_startProxy(
         // Try to build the runtime first — if allocation fails we want to
         // know before spinning up anything stateful.
         let rt = match tokio::runtime::Builder::new_multi_thread()
-            .worker_threads(2)
+            .worker_threads(4)
             .enable_all()
             .thread_name("mhrv-worker")
             .build()
@@ -483,6 +483,20 @@ pub extern "system" fn Java_com_therealaleph_mhrv_Native_statsJson<'a>(
     env.new_string(out).map(|s| s.into_raw()).unwrap_or(std::ptr::null_mut())
 }
 
+/// `Native.pipelineDebugJson()` -> String. Snapshot of pipeline debug state:
+/// elevated session count, batch semaphore usage, recent ramp/drop events.
+/// Temporary — for the debug overlay.
+#[no_mangle]
+pub extern "system" fn Java_com_therealaleph_mhrv_Native_pipelineDebugJson<'a>(
+    env: JNIEnv<'a>,
+    _class: JClass,
+) -> jstring {
+    let out = safe(String::new(), AssertUnwindSafe(|| {
+        crate::tunnel_client::pipeline_debug::to_json()
+    }));
+    env.new_string(out).map(|s| s.into_raw()).unwrap_or(std::ptr::null_mut())
+}
+
 // ---------------------------------------------------------------------------
 // tun2proxy CLI API wrapper (dlsym — no fork or patch needed)
 // ---------------------------------------------------------------------------
diff --git a/src/bin/ui.rs b/src/bin/ui.rs
index 7bb46b1..e0f8f6d 100644
--- a/src/bin/ui.rs
+++ b/src/bin/ui.rs
@@ -256,6 +256,10 @@ struct FormState {
     /// drop the user's setting. Not currently exposed as a UI control;
     /// users edit `block_quic` directly in `config.json` (Issue #213).
     block_quic: bool,
+    /// Round-tripped from config.json and exposed beside QUIC blocking.
+    /// Default true to push WebRTC apps toward TCP TURN instead of slow
+    /// UDP ICE retries.
+    block_stun: bool,
     /// Round-tripped from config.json. Not exposed as a UI control —
     /// users edit `disable_padding` directly when needed (Issue #391).
     /// Default false (padding active).
@@ -387,6 +391,7 @@ fn load_form() -> (FormState, Option<String>) {
             youtube_via_relay: c.youtube_via_relay,
             passthrough_hosts: c.passthrough_hosts.clone(),
             block_quic: c.block_quic,
+            block_stun: c.block_stun,
             disable_padding: c.disable_padding,
             force_http1: c.force_http1,
             tunnel_doh: c.tunnel_doh,
@@ -426,6 +431,7 @@ fn load_form() -> (FormState, Option<String>) {
             youtube_via_relay: false,
             passthrough_hosts: Vec::new(),
             block_quic: true,
+            block_stun: true,
             disable_padding: false,
             force_http1: false,
             tunnel_doh: true,
@@ -587,6 +593,7 @@ impl FormState {
             // control yet). Round-trip through the file so save
             // doesn't drop a user-set true.
             block_quic: self.block_quic,
+            block_stun: self.block_stun,
             // Issue #391: disable_padding is config-only for now.
             // Round-trip preserves the user's choice.
             disable_padding: self.disable_padding,
@@ -688,6 +695,9 @@ struct ConfigWire<'a> {
     /// emit only when the user has explicitly disabled the block.
     #[serde(skip_serializing_if = "is_true")]
     block_doh: bool,
+    /// Default true. Emit only when the user disables STUN/TURN blocking.
+    #[serde(skip_serializing_if = "is_true")]
+    block_stun: bool,
     #[serde(skip_serializing_if = "Vec::is_empty")]
     fronting_groups: &'a Vec<FrontingGroup>,
     /// Auto-blacklist tuning + batch timeout (#391, #444, #430). Skip
@@ -781,6 +791,7 @@ impl<'a> From<&'a Config> for ConfigWire<'a> {
             tunnel_doh: c.tunnel_doh,
             bypass_doh_hosts: &c.bypass_doh_hosts,
             block_doh: c.block_doh,
+            block_stun: c.block_stun,
             fronting_groups: &c.fronting_groups,
             auto_blacklist_strikes: c.auto_blacklist_strikes,
             auto_blacklist_window_secs: c.auto_blacklist_window_secs,
@@ -1274,6 +1285,15 @@ impl eframe::App for App {
                                  Issue #213, #793.",
                             );
                     });
+                    ui.horizontal(|ui| {
+                        ui.add_space(120.0 + 8.0);
+                        ui.checkbox(&mut self.form.block_stun, "Block STUN/TURN UDP")
+                            .on_hover_text(
+                                "Drop WebRTC STUN/TURN UDP ports 3478, 5349, and 19302 so apps \
+                                 such as Meet, Discord, and WhatsApp move to TCP TURN instead of \
+                                 waiting on UDP ICE retries.",
+                            );
+                    });
                 });
             });
 
diff --git a/src/config.rs b/src/config.rs
index 02ae2aa..d4251aa 100644
--- a/src/config.rs
+++ b/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
diff --git a/src/domain_fronter.rs b/src/domain_fronter.rs
index 2c1ac5e..bfcebe5 100644
--- a/src/domain_fronter.rs
+++ b/src/domain_fronter.rs
@@ -514,6 +514,8 @@ pub struct TunnelResponse {
     /// `e` only when this is `None` and compatibility is needed.
     #[serde(default)]
     pub code: Option<String>,
+    #[serde(default)]
+    pub seq: Option<u64>,
 }
 
 /// A single op in a batch tunnel request.
@@ -528,6 +530,10 @@ pub struct BatchOp {
     pub port: Option<u16>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub d: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub seq: Option<u64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub wseq: Option<u64>,
 }
 
 /// Batch tunnel response from Apps Script / tunnel node.
diff --git a/src/proxy_server.rs b/src/proxy_server.rs
index 3b2364a..209bbc5 100644
--- a/src/proxy_server.rs
+++ b/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(),
@@ -940,6 +942,17 @@ 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 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?;
+        sock.flush().await?;
+        return Ok(());
+    }
+
     tracing::info!("SOCKS5 CONNECT -> {}:{}", host, port);
 
     // Success reply with zeroed BND.
diff --git a/src/tunnel_client.rs b/src/tunnel_client.rs
index d09817a..887561a 100644
--- a/src/tunnel_client.rs
+++ b/src/tunnel_client.rs
@@ -5,7 +5,7 @@
 //! Each Apps Script deployment (account) gets its own concurrency pool of
 //! 30 in-flight requests — matching the per-account Apps Script limit.
 
-use std::collections::HashMap;
+use std::collections::{BTreeMap, HashMap};
 // `AtomicU64` from `std::sync::atomic` requires hardware-backed 64-bit
 // atomics, which 32-bit MIPS (`mipsel-unknown-linux-musl` — our OpenWRT
 // router target) does not provide — the std type isn't even defined
@@ -20,6 +20,7 @@ use std::time::{Duration, Instant};
 use base64::engine::general_purpose::STANDARD as B64;
 use base64::Engine;
 use bytes::{Bytes, BytesMut};
+use futures_util::stream::{FuturesUnordered, StreamExt};
 use tokio::io::{AsyncReadExt, AsyncWrite, AsyncWriteExt};
 use tokio::net::TcpStream;
 use tokio::sync::{mpsc, oneshot, Semaphore};
@@ -52,12 +53,34 @@ const MAX_BATCH_OPS: usize = 50;
 /// batch makes exactly one attempt (see `fire_batch` docs).
 const REPLY_TIMEOUT_SLACK: Duration = Duration::from_secs(5);
 
+/// Per-inflight reply timeout used by the pipelined poll loop. Each
+/// in-flight future independently times out after this duration so a
+/// dead target on the tunnel-node side doesn't block the session.
+const REPLY_TIMEOUT: Duration = Duration::from_secs(35);
+
 /// How long we'll briefly hold the client socket after the local
 /// CONNECT/SOCKS5 handshake, waiting for the client's first bytes (the
 /// TLS ClientHello for HTTPS). Bundling those bytes with the tunnel-node
 /// connect saves one Apps Script round-trip per new flow.
 const CLIENT_FIRST_DATA_WAIT: Duration = Duration::from_millis(50);
 
+/// Floor depth after a drop (first empty reply).
+const INFLIGHT_IDLE: usize = 1;
+
+/// Optimistic starting depth — every session gets 2 in-flight polls
+/// without needing an elevation permit. Drops to IDLE on first empty.
+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 = 4;
+
+/// How many consecutive empty replies before dropping from active to idle depth.
+const INFLIGHT_COOLDOWN: u32 = 3;
+
+/// Max sessions that can run at elevated pipeline depth per deployment.
+const MAX_ELEVATED_PER_DEPLOYMENT: u64 = 30;
+
 /// Adaptive coalesce defaults: after each new op arrives, wait another
 /// step for more ops. Resets on every arrival, up to max from the first
 /// op. Overridable via config `coalesce_step_ms` / `coalesce_max_ms`.
@@ -109,6 +132,89 @@ const UNREACHABLE_CACHE_TTL: Duration = Duration::from_secs(30);
 /// hosts; we don't want a runaway map. Pruned opportunistically on insert.
 const UNREACHABLE_CACHE_MAX: usize = 256;
 
+// ---------------------------------------------------------------------------
+// Pipeline debug overlay state — temporary, polled from Android UI.
+// ---------------------------------------------------------------------------
+pub(crate) mod pipeline_debug {
+    use std::collections::VecDeque;
+    use std::sync::{Mutex, OnceLock};
+    use portable_atomic::AtomicU64;
+    use std::sync::atomic::Ordering;
+
+    const EVENT_CAP: usize = 30;
+
+    struct SessionInfo {
+        depth: usize,
+        inflight: usize,
+        elevated: bool,
+    }
+
+    struct State {
+        events: Mutex<VecDeque<String>>,
+        elevated: AtomicU64,
+        max_elevated: AtomicU64,
+        active_batches: AtomicU64,
+        max_batch_slots: AtomicU64,
+        active_sessions: AtomicU64,
+        sessions: Mutex<std::collections::HashMap<String, SessionInfo>>,
+    }
+
+    fn state() -> &'static State {
+        static S: OnceLock<State> = OnceLock::new();
+        S.get_or_init(|| State {
+            events: Mutex::new(VecDeque::with_capacity(EVENT_CAP)),
+            elevated: AtomicU64::new(0),
+            max_elevated: AtomicU64::new(0),
+            active_batches: AtomicU64::new(0),
+            max_batch_slots: AtomicU64::new(0),
+            active_sessions: AtomicU64::new(0),
+            sessions: Mutex::new(std::collections::HashMap::new()),
+        })
+    }
+
+    pub fn push_event(_msg: String) {}
+    pub fn set_limits(_max_elev: u64, _max_batches: u64) {}
+    pub fn set_elevated(_n: u64) {}
+    pub fn batch_acquire() {}
+    pub fn batch_release() {}
+    pub fn session_start(_sid: &str) {}
+    pub fn session_end(_sid: &str) {}
+    pub fn session_update(_sid: &str, _depth: usize, _inflight: usize, _elevated: bool) {}
+
+    pub fn to_json() -> String {
+        let s = state();
+        let events_json = if let Ok(g) = s.events.lock() {
+            let escaped: Vec<String> = g.iter().map(|e| {
+                format!("\"{}\"", e.replace('\\', "\\\\").replace('"', "\\\""))
+            }).collect();
+            format!("[{}]", escaped.join(","))
+        } else {
+            "[]".to_string()
+        };
+        let sessions_json = if let Ok(g) = s.sessions.lock() {
+            let entries: Vec<String> = g.iter().map(|(sid, info)| {
+                format!(
+                    r#"{{"sid":"{}","depth":{},"inflight":{},"elevated":{}}}"#,
+                    sid, info.depth, info.inflight, info.elevated,
+                )
+            }).collect();
+            format!("[{}]", entries.join(","))
+        } else {
+            "[]".to_string()
+        };
+        format!(
+            r#"{{"elevated":{},"max_elevated":{},"active_batches":{},"max_batch_slots":{},"active_sessions":{},"sessions":{},"events":{}}}"#,
+            s.elevated.load(Ordering::Relaxed),
+            s.max_elevated.load(Ordering::Relaxed),
+            s.active_batches.load(Ordering::Relaxed),
+            s.max_batch_slots.load(Ordering::Relaxed),
+            s.active_sessions.load(Ordering::Relaxed),
+            sessions_json,
+            events_json,
+        )
+    }
+}
+
 /// Ports where the *server* speaks first (SMTP banner, SSH identification,
 /// POP3/IMAP greeting, FTP banner). On these, waiting for client bytes
 /// gains nothing and just adds handshake latency — skip the pre-read.
@@ -174,6 +280,8 @@ enum MuxMsg {
     Data {
         sid: String,
         data: Bytes,
+        seq: Option<u64>,
+        wseq: Option<u64>,
         reply: BatchedReply,
     },
     UdpOpen {
@@ -209,10 +317,12 @@ struct PendingOp {
     /// only `connect_data`, which uses presence of `d` as the signal
     /// that the caller is opting into the bundled-first-bytes flow).
     encode_empty: bool,
+    seq: Option<u64>,
+    wseq: Option<u64>,
 }
 
 pub struct TunnelMux {
-    tx: mpsc::Sender<MuxMsg>,
+    tx: mpsc::UnboundedSender<MuxMsg>,
     /// Set to `true` after the first time the tunnel-node rejects
     /// `connect_data` as unsupported. Subsequent sessions skip the
     /// optimistic path entirely and go straight to plain connect + data.
@@ -289,6 +399,9 @@ pub struct TunnelMux {
     /// `request_timeout_secs` would see sessions abandon replies just
     /// before the batch would have completed.
     reply_timeout: Duration,
+    /// How many sessions are currently at elevated pipeline depth (>= 3).
+    elevated_sessions: AtomicU64,
+    max_elevated: u64,
 }
 
 impl TunnelMux {
@@ -319,7 +432,7 @@ impl TunnelMux {
         );
         let step = if coalesce_step_ms > 0 { coalesce_step_ms } else { DEFAULT_COALESCE_STEP_MS };
         let max = if coalesce_max_ms > 0 { coalesce_max_ms } else { DEFAULT_COALESCE_MAX_MS };
-        tracing::info!("batch coalesce: step={}ms max={}ms", step, max);
+        tracing::info!("batch coalesce: step={}ms max={}ms, pipeline max depth: {}, optimist: {}", step, max, INFLIGHT_ACTIVE, INFLIGHT_OPTIMIST);
         // Reply timeout co-varies with `request_timeout_secs` so an
         // operator who raises the batch budget doesn't have sessions
         // abandoning replies just before the HTTP round-trip would
@@ -328,7 +441,11 @@ impl TunnelMux {
         let reply_timeout = fronter
             .batch_timeout()
             .saturating_add(REPLY_TIMEOUT_SLACK);
-        let (tx, rx) = mpsc::channel(512);
+        pipeline_debug::set_limits(
+            MAX_ELEVATED_PER_DEPLOYMENT * unique_n as u64,
+            (CONCURRENCY_PER_DEPLOYMENT * unique_n) as u64,
+        );
+        let (tx, rx) = mpsc::unbounded_channel();
         tokio::spawn(mux_loop(rx, fronter, step, max));
         Arc::new(Self {
             tx,
@@ -344,6 +461,8 @@ impl TunnelMux {
             preread_total_events: AtomicU64::new(0),
             unreachable_cache: Mutex::new(HashMap::new()),
             reply_timeout,
+            elevated_sessions: AtomicU64::new(0),
+            max_elevated: MAX_ELEVATED_PER_DEPLOYMENT * unique_n as u64,
         })
     }
 
@@ -354,8 +473,12 @@ impl TunnelMux {
         self.reply_timeout
     }
 
+    fn send_sync(&self, msg: MuxMsg) {
+        let _ = self.tx.send(msg);
+    }
+
     async fn send(&self, msg: MuxMsg) {
-        let _ = self.tx.send(msg).await;
+        let _ = self.tx.send(msg);
     }
 
     pub async fn udp_open(
@@ -616,7 +739,7 @@ impl TunnelMux {
     }
 }
 
-async fn mux_loop(mut rx: mpsc::Receiver<MuxMsg>, fronter: Arc<DomainFronter>, coalesce_step_ms: u64, coalesce_max_ms: u64) {
+async fn mux_loop(mut rx: mpsc::UnboundedReceiver<MuxMsg>, fronter: Arc<DomainFronter>, coalesce_step_ms: u64, coalesce_max_ms: u64) {
     let coalesce_step = Duration::from_millis(coalesce_step_ms);
     let coalesce_max = Duration::from_millis(coalesce_max_ms);
     // One semaphore per deployment ID, each allowing 30 concurrent requests.
@@ -704,10 +827,12 @@ async fn mux_loop(mut rx: mpsc::Receiver<MuxMsg>, fronter: Arc<DomainFronter>, c
                         port: Some(port),
                         data: Some(data),
                         encode_empty: true,
+                        seq: None,
+                        wseq: None,
                     };
                     accum.push_or_fire(op, op_bytes, reply, &sems, &fronter).await;
                 }
-                MuxMsg::Data { sid, data, reply } => {
+                MuxMsg::Data { sid, data, seq, wseq, reply } => {
                     let op_bytes = encoded_len(data.len());
                     let op = PendingOp {
                         op: "data",
@@ -716,6 +841,8 @@ async fn mux_loop(mut rx: mpsc::Receiver<MuxMsg>, fronter: Arc<DomainFronter>, c
                         port: None,
                         data: if data.is_empty() { None } else { Some(data) },
                         encode_empty: false,
+                        seq,
+                        wseq,
                     };
                     accum.push_or_fire(op, op_bytes, reply, &sems, &fronter).await;
                 }
@@ -733,6 +860,8 @@ async fn mux_loop(mut rx: mpsc::Receiver<MuxMsg>, fronter: Arc<DomainFronter>, c
                         port: Some(port),
                         data: if data.is_empty() { None } else { Some(data) },
                         encode_empty: false,
+                        seq: None,
+                        wseq: None,
                     };
                     accum.push_or_fire(op, op_bytes, reply, &sems, &fronter).await;
                 }
@@ -745,6 +874,8 @@ async fn mux_loop(mut rx: mpsc::Receiver<MuxMsg>, fronter: Arc<DomainFronter>, c
                         port: None,
                         data: if data.is_empty() { None } else { Some(data) },
                         encode_empty: false,
+                        seq: None,
+                        wseq: None,
                     };
                     accum.push_or_fire(op, op_bytes, reply, &sems, &fronter).await;
                 }
@@ -764,6 +895,8 @@ async fn mux_loop(mut rx: mpsc::Receiver<MuxMsg>, fronter: Arc<DomainFronter>, c
                 port: None,
                 data: None,
                 encode_empty: false,
+                seq: None,
+                wseq: None,
             });
         }
 
@@ -868,6 +1001,8 @@ fn encode_pending(p: PendingOp) -> BatchOp {
         host: p.host,
         port: p.port,
         d,
+        seq: p.seq,
+        wseq: p.wseq,
     }
 }
 
@@ -896,9 +1031,13 @@ async fn fire_batch(
         .cloned()
         .unwrap_or_else(|| Arc::new(Semaphore::new(CONCURRENCY_PER_DEPLOYMENT)));
     let permit = sem.acquire_owned().await.unwrap();
+    pipeline_debug::batch_acquire();
     let f = fronter.clone();
 
     tokio::spawn(async move {
+        struct BatchGuard;
+        impl Drop for BatchGuard { fn drop(&mut self) { pipeline_debug::batch_release(); } }
+        let _batch_guard = BatchGuard;
         let _permit = permit;
         let t0 = std::time::Instant::now();
         let n_ops = pending_ops.len();
@@ -962,6 +1101,10 @@ async fn fire_batch(
                     if let Some(resp) = batch_resp.r.get(idx) {
                         let _ = reply.send(Ok((resp.clone(), script_id.clone())));
                     } else {
+                        tracing::error!(
+                            "batch response mismatch: idx={} but r.len()={} (sent {} ops) from script {}",
+                            idx, batch_resp.r.len(), n_ops, sid_short,
+                        );
                         let _ = reply.send(Err(format!(
                             "missing response in batch from script {}",
                             sid_short
@@ -1081,6 +1224,7 @@ pub async fn tunnel_connection(
     };
 
     tracing::info!("tunnel session {} opened for {}:{}", sid, host, port);
+    pipeline_debug::session_start(&sid);
 
     // Run the first-response write + tunnel_loop inside an async block so
     // any io-error propagates via `?` without bypassing the Close below.
@@ -1110,6 +1254,7 @@ pub async fn tunnel_connection(
     .await;
 
     mux.send(MuxMsg::Close { sid: sid.clone() }).await;
+    pipeline_debug::session_end(&sid);
     tracing::info!("tunnel session {} closed for {}:{}", sid, host, port);
     result
 }
@@ -1267,181 +1412,479 @@ fn is_connect_data_unsupported_error_str(e: &str) -> bool {
     (e.contains("unknown op") || e.contains("unknown tunnel op")) && e.contains("connect_data")
 }
 
+/// Metadata for one in-flight Data op, returned alongside its reply.
+struct InflightMeta {
+    seq: u64,
+    was_empty_poll: bool,
+    send_at: Instant,
+}
+
+
 async fn tunnel_loop(
     sock: &mut TcpStream,
     sid: &str,
     mux: &Arc<TunnelMux>,
-    mut pending_client_data: Option<Bytes>,
+    pending_client_data: Option<Bytes>,
 ) -> std::io::Result<()> {
     let (mut reader, mut writer) = sock.split();
-    // `BytesMut` + `read_buf` + a per-read decision between
-    // `split().freeze()` (zero-copy) and `copy_from_slice` + `clear`
-    // (right-sized copy, buffer reused).
-    //
-    // Why the split decision: `bytes` 1.x refcounts the *whole*
-    // backing allocation, so a frozen `Bytes` from a partial read
-    // pins all `READ_CHUNK` bytes until it drops. Under semaphore
-    // saturation or reply timeouts, dozens of small TLS records or
-    // HTTP/2 frames can each retain ~64 KB instead of their actual
-    // payload size — order-of-magnitude memory regression on
-    // constrained targets (router builds with 64 MB RAM).
-    //
-    // Threshold: at ≥ half-buffer the saved memcpy outweighs the
-    // wasted slack, and these reads are typically streaming bulk
-    // transfer where the `Bytes` flushes through the mux quickly.
-    // Below that, copy out and `clear()` so the same allocation
-    // serves the next read — equivalent memory profile to the old
-    // `vec![0u8; 65536]` + `to_vec()` code on small-read workloads.
-    const READ_CHUNK: usize = 65536;
-    const ZERO_COPY_THRESHOLD: usize = READ_CHUNK / 2;
-    let mut buf = BytesMut::with_capacity(READ_CHUNK);
+
+    let inflight_cap = INFLIGHT_ACTIVE;
+    let mut max_inflight = INFLIGHT_OPTIMIST.min(inflight_cap);
     let mut consecutive_empty = 0u32;
+    let mut consecutive_data: u32 = 0;
+    let mut is_elevated = false;
+    let mut total_download_bytes: u64 = 0;
+    let mut next_send_seq: u64 = 0;
+    let mut next_write_seq: u64 = 0;
+    let mut next_data_write_seq: u64 = 0;
+    let mut eof_seen = false;
+    let mut client_closed = false;
+    let mut pending_writes: BTreeMap<u64, (TunnelResponse, String)> = BTreeMap::new();
 
-    loop {
-        // Cadence depends on whether the tunnel-node is doing long-poll
-        // drains. With long-poll, the server holds empty polls open up
-        // to its `LONGPOLL_DEADLINE` (~5 s currently), so the client
-        // can keep this read timeout short — the wait is on the wire,
-        // not here. Against *legacy* tunnel-nodes (no long-poll, fast
-        // empty replies), the same short cadence + always-poll behavior
-        // would generate continuous round-trips on idle sessions and
-        // burn Apps Script quota.
-        //
-        // Both the read timeout and the skip-empty-when-idle decision
-        // are gated on `all_legacy` — i.e. *every known deployment is
-        // currently legacy*. Per-deployment "skip when this script is
-        // legacy" sounds appealing but is unsafe: the next op's
-        // deployment is chosen by `next_script_id()` only when the
-        // batch fires, so the loop can't predict where the empty poll
-        // will land. Suppressing polls based on the *previous* reply's
-        // script would stall remote→client data on mixed setups —
-        // round-robin would never reach the long-poll-capable peer for
-        // this session if every iteration short-circuits before
-        // sending. Cost of the conservative gate: legacy peers see
-        // some wasted empty polls when at least one peer is healthy,
-        // bounded by round-robin fan-out. Worth it to keep pushed
-        // bytes flowing.
-        let all_legacy = mux.all_servers_legacy();
-        let client_data = if let Some(data) = pending_client_data.take() {
-            Some(data)
-        } else {
-            let read_timeout = match (all_legacy, consecutive_empty) {
-                (_, 0) => Duration::from_millis(20),
-                (_, 1) => Duration::from_millis(80),
-                (_, 2) => Duration::from_millis(200),
-                (false, _) => Duration::from_millis(500),
-                (true, _) => Duration::from_secs(30),
-            };
+    // Buffered upload data waiting to be sent (when pipeline is full).
+    let mut buffered_upload: Option<Bytes> = None;
 
-            buf.reserve(READ_CHUNK);
-            match tokio::time::timeout(read_timeout, reader.read_buf(&mut buf)).await {
-                Ok(Ok(0)) => break,
-                Ok(Ok(n)) => {
-                    consecutive_empty = 0;
-                    if n >= ZERO_COPY_THRESHOLD {
-                        // Big read: split off the filled region. The
-                        // frozen `Bytes` is at-least-half-full, so the
-                        // saved 64 KB memcpy outweighs the brief
-                        // retention until the mux drains.
-                        Some(buf.split().freeze())
-                    } else {
-                        // Small read: copy out a payload-sized `Bytes`
-                        // and `clear()` so the buffer is reused on the
-                        // next iter (no `reserve` allocation needed
-                        // because the alloc stays uniquely owned).
-                        // Bounds retention to actual data even when
-                        // the mux is backpressured.
-                        let owned = Bytes::copy_from_slice(&buf[..n]);
-                        buf.clear();
-                        Some(owned)
-                    }
-                }
-                Ok(Err(_)) => break,
-                Err(_) => None,
+    enum ReplyOutcome {
+        Ok(TunnelResponse, String),
+        BatchErr(String),
+        Timeout,
+        Dropped,
+    }
+    type ReplyFut =
+        std::pin::Pin<Box<dyn std::future::Future<Output = (InflightMeta, ReplyOutcome)> + Send>>;
+    let mut inflight: FuturesUnordered<ReplyFut> = FuturesUnordered::new();
+
+    // Helper: wrap a reply_rx into a ReplyFut with timeout.
+    fn wrap_reply(
+        meta: InflightMeta,
+        reply_rx: oneshot::Receiver<Result<(TunnelResponse, String), String>>,
+    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = (InflightMeta, ReplyOutcome)> + Send>> {
+        Box::pin(async move {
+            match tokio::time::timeout(REPLY_TIMEOUT, reply_rx).await {
+                Ok(Ok(Ok((r, sid)))) => (meta, ReplyOutcome::Ok(r, sid)),
+                Ok(Ok(Err(e))) => (meta, ReplyOutcome::BatchErr(e)),
+                Ok(Err(_)) => (meta, ReplyOutcome::Dropped),
+                Err(_) => (meta, ReplyOutcome::Timeout),
             }
-        };
-
-        // Skip empty polls only when *every* deployment is legacy. With
-        // even one long-poll-capable peer, round-robin will land some
-        // empty polls there where the server holds them open and can
-        // deliver pushed bytes — that's the whole point of long-poll,
-        // so we must keep emitting. See the `all_legacy` comment above
-        // for why a per-deployment gate here would stall mixed setups.
-        if all_legacy && client_data.is_none() && consecutive_empty > 3 {
-            continue;
-        }
-
-        let data = client_data.unwrap_or_else(Bytes::new);
-        let was_empty_poll = data.is_empty();
+        })
+    }
 
+    /// Send an empty poll Data op. Returns the InflightMeta and reply rx.
+    #[inline]
+    fn send_empty_poll(
+        sid: &str,
+        next_send_seq: &mut u64,
+        mux: &Arc<TunnelMux>,
+    ) -> (
+        InflightMeta,
+        oneshot::Receiver<Result<(TunnelResponse, String), String>>,
+    ) {
+        let seq = *next_send_seq;
+        *next_send_seq += 1;
         let (reply_tx, reply_rx) = oneshot::channel();
         let send_at = Instant::now();
-        mux.send(MuxMsg::Data {
+        mux.send_sync(MuxMsg::Data {
+            sid: sid.to_string(),
+            data: Bytes::new(),
+            seq: Some(seq),
+            wseq: None,
+            reply: reply_tx,
+        });
+        let meta = InflightMeta { seq, was_empty_poll: true, send_at };
+        (meta, reply_rx)
+    }
+
+    /// Send a data op with wseq. Returns the InflightMeta and reply rx.
+    #[inline]
+    fn send_data_op(
+        sid: &str,
+        data: Bytes,
+        next_send_seq: &mut u64,
+        next_data_write_seq: &mut u64,
+        mux: &Arc<TunnelMux>,
+    ) -> (
+        InflightMeta,
+        oneshot::Receiver<Result<(TunnelResponse, String), String>>,
+    ) {
+        let seq = *next_send_seq;
+        *next_send_seq += 1;
+        let wseq = *next_data_write_seq;
+        *next_data_write_seq += 1;
+        let (reply_tx, reply_rx) = oneshot::channel();
+        let send_at = Instant::now();
+        let sid_short = &sid[..sid.len().min(8)];
+        tracing::info!(
+            "sess {}: upload send seq={} wseq={} len={}B",
+            sid_short, seq, wseq, data.len(),
+        );
+        mux.send_sync(MuxMsg::Data {
             sid: sid.to_string(),
             data,
+            seq: Some(seq),
+            wseq: Some(wseq),
             reply: reply_tx,
-        })
-        .await;
+        });
+        let meta = InflightMeta { seq, was_empty_poll: false, send_at };
+        (meta, reply_rx)
+    }
 
-        // Bounded-wait on reply: if the batch this op landed in is slow
-        // (dead target on the tunnel-node side), don't block this session
-        // forever — timeout and let it retry on the next tick.
-        let (resp, script_id) = match tokio::time::timeout(mux.reply_timeout(), reply_rx).await {
-            Ok(Ok(Ok((r, sid_used)))) => (r, sid_used),
-            Ok(Ok(Err(e))) => {
-                tracing::debug!("tunnel data error: {}", e);
-                break;
-            }
-            Ok(Err(_)) => break, // channel dropped
-            Err(_) => {
-                tracing::warn!("sess {}: reply timeout, retrying", &sid[..sid.len().min(8)]);
-                consecutive_empty = consecutive_empty.saturating_add(1);
-                continue;
-            }
-        };
-
-        // Per-deployment legacy detection: an empty-in/empty-out round
-        // trip that finishes well under `LEGACY_DETECT_THRESHOLD` is
-        // structurally impossible on a long-poll-capable tunnel-node
-        // (the server holds the response either until data arrives or
-        // until its long-poll deadline). One observation marks *this
-        // specific* deployment as legacy for `LEGACY_RECOVER_AFTER`;
-        // peers stay on the fast path. The aggregate `all_legacy` gate
-        // only flips once *every* deployment has been so marked.
-        if was_empty_poll {
-            let reply_was_empty = resp.d.as_deref().map(str::is_empty).unwrap_or(true);
-            if reply_was_empty && send_at.elapsed() < LEGACY_DETECT_THRESHOLD {
-                mux.mark_server_no_longpoll(&script_id);
-            }
-        }
-
-        if let Some(ref e) = resp.e {
-            tracing::debug!("tunnel error: {}", e);
-            break;
-        }
-
-        let got_data = match write_tunnel_response(&mut writer, &resp).await? {
-            WriteOutcome::Wrote => true,
-            WriteOutcome::NoData => false,
-            WriteOutcome::BadBase64 => {
-                // Tunnel-node gave us garbage; tear the session down but
-                // do NOT propagate as an io error — the caller's Close
-                // guard will clean up on the tunnel-node side.
-                break;
-            }
-        };
-
-        if resp.eof.unwrap_or(false) {
-            break;
-        }
-
-        if got_data {
-            consecutive_empty = 0;
-        } else {
-            consecutive_empty = consecutive_empty.saturating_add(1);
+    // ── Initial path: send pending client data or read from client ──
+    if let Some(data) = pending_client_data {
+        if !data.is_empty() {
+            let (meta, reply_rx) = send_data_op(sid, data, &mut next_send_seq, &mut next_data_write_seq, mux);
+            tracing::debug!(
+                "sess {}: pending data seq={}",
+                &sid[..sid.len().min(8)],
+                meta.seq,
+            );
+            inflight.push(wrap_reply(meta, reply_rx));
         }
     }
 
+    // Send initial pre-fill empty polls (optimist depth), staggered
+    // 1s apart so they land in separate batches. The pending data op
+    // (if any) already occupies one slot.
+    {
+        let polls_to_send = max_inflight.saturating_sub(inflight.len());
+        for i in 0..polls_to_send {
+            if i > 0 {
+                tokio::time::sleep(Duration::from_secs(1)).await;
+            }
+            let (meta, reply_rx) = send_empty_poll(sid, &mut next_send_seq, mux);
+            tracing::debug!(
+                "sess {}: prefill poll seq={}, inflight={}",
+                &sid[..sid.len().min(8)],
+                meta.seq,
+                inflight.len() + 1,
+            );
+            inflight.push(wrap_reply(meta, reply_rx));
+        }
+    }
+
+    // Timer for staggered refill polls — fires in the select, never blocks.
+    let mut refill_at: Option<std::pin::Pin<Box<tokio::time::Sleep>>> = None;
+    let mut refill_steps: u32 = 0;
+
+    // Schedule initial refill if pre-fill didn't fill all slots.
+    if inflight.len() < max_inflight {
+        refill_at = Some(Box::pin(tokio::time::sleep(Duration::from_millis(100))));
+        refill_steps = 0;
+    }
+
+    // Read buffer for client socket.
+    let mut read_buf = BytesMut::with_capacity(65536);
+
+    // Main select loop — handles both upload reads and download replies.
+    loop {
+        // If nothing in flight and tunnel EOF, we're done.
+        if inflight.is_empty() && eof_seen {
+            break;
+        }
+
+        // If nothing in flight and client closed, we're done.
+        if inflight.is_empty() && client_closed {
+            break;
+        }
+
+        // If eof was seen but inflight is not empty, give remaining
+        // replies a short grace period to deliver any buffered data
+        // before the remote connection closed. After 500ms, abandon them.
+        if eof_seen && !inflight.is_empty() {
+            match tokio::time::timeout(Duration::from_millis(500), inflight.next()).await {
+                Ok(Some((meta, ReplyOutcome::Ok(resp, script_id)))) => {
+                    if meta.seq == next_write_seq {
+                        let _ = write_tunnel_response(&mut writer, &resp).await;
+                        next_write_seq += 1;
+                        while let Some(entry) = pending_writes.first_entry() {
+                            if *entry.key() != next_write_seq { break; }
+                            let (_, (buffered_resp, _)) = entry.remove_entry();
+                            let _ = write_tunnel_response(&mut writer, &buffered_resp).await;
+                            next_write_seq += 1;
+                        }
+                    } else {
+                        pending_writes.insert(meta.seq, (resp, script_id));
+                    }
+                    continue;
+                }
+                _ => break,
+            }
+        }
+
+        // When inflight is empty and we haven't seen eof, read from
+        // client or send an empty poll to keep the session alive.
+        if inflight.is_empty() && !eof_seen {
+            let all_legacy = mux.all_servers_legacy();
+
+            // If all servers are legacy and we've had many consecutive
+            // empties, wait for client data before sending.
+            if all_legacy && consecutive_empty > 3 && !client_closed {
+                read_buf.reserve(65536);
+                match reader.read_buf(&mut read_buf).await {
+                    Ok(0) => break,
+                    Ok(n) => {
+                        consecutive_empty = 0;
+                        let data = extract_bytes(&mut read_buf, n);
+                        let (meta, reply_rx) = send_data_op(sid, data, &mut next_send_seq, &mut next_data_write_seq, mux);
+                        inflight.push(wrap_reply(meta, reply_rx));
+                        continue;
+                    }
+                    Err(_) => break,
+                }
+            }
+
+            let (meta, reply_rx) = send_empty_poll(sid, &mut next_send_seq, mux);
+            tracing::debug!(
+                "sess {}: keepalive poll seq={}", &sid[..sid.len().min(8)], meta.seq
+            );
+            inflight.push(wrap_reply(meta, reply_rx));
+        }
+
+        // Can we read from the client? Yes if not closed, not eof, and
+        // we have room for more inflight ops (fast-path allows +4 extra).
+        let can_read = !client_closed && !eof_seen && inflight.len() < max_inflight + 4;
+
+        tokio::select! {
+            biased;
+
+            // Refill timer: 100ms steps, send empty poll after 10 steps
+            // (1s) for batch separation.
+            _ = async { refill_at.as_mut().unwrap().await }, if refill_at.is_some() => {
+                refill_at = None;
+                if !eof_seen && inflight.len() < max_inflight {
+                    refill_steps += 1;
+
+                    if refill_steps >= 10 {
+                        // Check buffered upload first — merge into a data
+                        // op instead of sending an empty poll.
+                        if let Some(data) = buffered_upload.take() {
+                            let (meta, reply_rx) = send_data_op(sid, data, &mut next_send_seq, &mut next_data_write_seq, mux);
+                            inflight.push(wrap_reply(meta, reply_rx));
+                        } else {
+                            let (meta, reply_rx) = send_empty_poll(sid, &mut next_send_seq, mux);
+                            inflight.push(wrap_reply(meta, reply_rx));
+                        }
+                        refill_steps = 0;
+
+                        if inflight.len() < max_inflight && max_inflight > INFLIGHT_IDLE {
+                            refill_at = Some(Box::pin(tokio::time::sleep(Duration::from_millis(100))));
+                        }
+                    } else {
+                        refill_at = Some(Box::pin(tokio::time::sleep(Duration::from_millis(100))));
+                    }
+                }
+            }
+
+            // Process completed replies.
+            Some((meta, outcome)) = inflight.next() => {
+                match outcome {
+                    ReplyOutcome::Ok(resp, script_id) => {
+                        let has_data = resp.d.as_ref().map(|d| !d.is_empty()).unwrap_or(false);
+                        tracing::debug!(
+                            "sess {}: recv seq={}, rtt={:?}, data={}, inflight={}",
+                            &sid[..sid.len().min(8)],
+                            meta.seq,
+                            meta.send_at.elapsed(),
+                            has_data,
+                            inflight.len(),
+                        );
+                        if resp.seq.is_none() {
+                            max_inflight = 1;
+                        }
+
+                        if let Some(ref e) = resp.e {
+                            tracing::debug!("tunnel error: {}", e);
+                            break;
+                        }
+
+                        let is_eof = resp.eof.unwrap_or(false);
+                        let resp_has_seq = resp.seq.is_some();
+
+                        // Write in-order to client.
+                        if meta.seq == next_write_seq {
+                            let got_data = match write_tunnel_response(&mut writer, &resp).await? {
+                                WriteOutcome::Wrote => true,
+                                WriteOutcome::NoData => false,
+                                WriteOutcome::BadBase64 => break,
+                            };
+                            next_write_seq += 1;
+                            if got_data {
+                                consecutive_empty = 0;
+                                consecutive_data = consecutive_data.saturating_add(1);
+                                let bytes = resp.d.as_ref().map(|d| d.len() as u64 * 3 / 4).unwrap_or(0);
+                                total_download_bytes += bytes;
+                            } else {
+                                consecutive_empty = consecutive_empty.saturating_add(1);
+                            }
+                            if is_eof {
+                                eof_seen = true;
+                            }
+
+                            // Flush buffered out-of-order writes.
+                            while let Some(entry) = pending_writes.first_entry() {
+                                if *entry.key() != next_write_seq { break; }
+                                let (_, (buffered_resp, _)) = entry.remove_entry();
+                                let buf_eof = buffered_resp.eof.unwrap_or(false);
+                                match write_tunnel_response(&mut writer, &buffered_resp).await? {
+                                    WriteOutcome::Wrote => {
+                                        consecutive_empty = 0;
+                                        consecutive_data = consecutive_data.saturating_add(1);
+                                        let bytes = buffered_resp.d.as_ref().map(|d| d.len() as u64 * 3 / 4).unwrap_or(0);
+                                        total_download_bytes += bytes;
+                                    }
+                                    WriteOutcome::NoData => {
+                                        consecutive_empty = consecutive_empty.saturating_add(1);
+                                    }
+                                    WriteOutcome::BadBase64 => break,
+                                }
+                                next_write_seq += 1;
+                                if buf_eof {
+                                    eof_seen = true;
+                                }
+                            }
+                        } else {
+                            pending_writes.insert(meta.seq, (resp, script_id));
+                        }
+
+                        // Send buffered upload data now that a slot freed up.
+                        if let Some(data) = buffered_upload.take() {
+                            if inflight.len() < max_inflight {
+                                let (meta, reply_rx) = send_data_op(sid, data, &mut next_send_seq, &mut next_data_write_seq, mux);
+                                consecutive_empty = 0;
+                                inflight.push(wrap_reply(meta, reply_rx));
+                            } else {
+                                buffered_upload = Some(data);
+                            }
+                        }
+
+                        // Adaptive pipeline depth management.
+                        tracing::info!(
+                            "sess {}: depth={} cd={} ce={} inf={} has_seq={}",
+                            &sid[..sid.len().min(8)],
+                            max_inflight, consecutive_data, consecutive_empty, inflight.len(), resp_has_seq,
+                        );
+                        if resp_has_seq {
+                            let prev = max_inflight;
+                            if consecutive_empty >= 2 && max_inflight > INFLIGHT_IDLE {
+                                max_inflight = INFLIGHT_IDLE.min(inflight_cap);
+                                if is_elevated {
+                                    let n = mux.elevated_sessions.fetch_sub(1, Ordering::Relaxed);
+                                    pipeline_debug::set_elevated(n.saturating_sub(1));
+                                    is_elevated = false;
+                                }
+                            } else if consecutive_data >= 1 && max_inflight < INFLIGHT_OPTIMIST {
+                                max_inflight = INFLIGHT_OPTIMIST.min(inflight_cap);
+                            } else if consecutive_data >= 2
+                                && max_inflight >= INFLIGHT_OPTIMIST
+                                && max_inflight < inflight_cap
+                                && total_download_bytes >= 32 * 1024
+                            {
+                                if !is_elevated {
+                                    let cur = mux.elevated_sessions.load(Ordering::Relaxed);
+                                    if cur < mux.max_elevated {
+                                        let n = mux.elevated_sessions.fetch_add(1, Ordering::Relaxed);
+                                        pipeline_debug::set_elevated(n + 1);
+                                        is_elevated = true;
+                                        max_inflight = (max_inflight + 1).min(inflight_cap);
+                                    }
+                                } else {
+                                    max_inflight = (max_inflight + 1).min(inflight_cap);
+                                }
+                            }
+                            pipeline_debug::session_update(sid, max_inflight, inflight.len(), is_elevated);
+                            if max_inflight != prev {
+                                tracing::info!(
+                                    "sess {}: pipeline {} -> {}{}",
+                                    &sid[..sid.len().min(8)],
+                                    prev,
+                                    max_inflight,
+                                    if is_elevated { " [elevated]" } else { "" },
+                                );
+                                pipeline_debug::push_event(format!(
+                                    "{} {}->{}{}",
+                                    &sid[..sid.len().min(8)],
+                                    prev,
+                                    max_inflight,
+                                    if is_elevated { " E" } else { "" },
+                                ));
+                            }
+                        }
+
+                        // Schedule refill if pipeline needs more polls.
+                        if !eof_seen
+                            && inflight.len() < max_inflight
+                            && refill_at.is_none()
+                        {
+                            refill_at = Some(Box::pin(tokio::time::sleep(
+                                if max_inflight > INFLIGHT_IDLE { Duration::from_millis(100) } else { Duration::ZERO }
+                            )));
+                            refill_steps = 0;
+                        }
+                    }
+                    ReplyOutcome::BatchErr(e) => {
+                        tracing::debug!("tunnel data error: {}", e);
+                        break;
+                    }
+                    ReplyOutcome::Timeout => {
+                        tracing::warn!(
+                            "sess {}: reply timeout (seq {}), retrying",
+                            &sid[..sid.len().min(8)],
+                            meta.seq,
+                        );
+                        consecutive_empty = consecutive_empty.saturating_add(1);
+                    }
+                    ReplyOutcome::Dropped => {
+                        break;
+                    }
+                }
+            }
+
+            // Read from client (overlapped with reply processing).
+            result = async {
+                read_buf.reserve(65536);
+                reader.read_buf(&mut read_buf).await
+            }, if can_read => {
+                match result {
+                    Ok(0) => {
+                        client_closed = true;
+                    }
+                    Ok(n) => {
+                        let data = extract_bytes(&mut read_buf, n);
+                        if inflight.len() < max_inflight {
+                            // Normal path: send immediately as data op.
+                            let (meta, reply_rx) = send_data_op(sid, data, &mut next_send_seq, &mut next_data_write_seq, mux);
+                            consecutive_empty = 0;
+                            inflight.push(wrap_reply(meta, reply_rx));
+                        } else if inflight.len() < max_inflight + 4 {
+                            // Fast-path: pipeline full but under +4 extra.
+                            let (meta, reply_rx) = send_data_op(sid, data, &mut next_send_seq, &mut next_data_write_seq, mux);
+                            consecutive_empty = 0;
+                            inflight.push(wrap_reply(meta, reply_rx));
+                        } else {
+                            // Buffer upload data until a slot frees up.
+                            if let Some(ref mut existing) = buffered_upload {
+                                // Merge: append new data to existing buffer.
+                                let mut merged = BytesMut::with_capacity(existing.len() + data.len());
+                                merged.extend_from_slice(existing);
+                                merged.extend_from_slice(&data);
+                                *existing = merged.freeze();
+                            } else {
+                                buffered_upload = Some(data);
+                            }
+                        }
+                    }
+                    Err(_) => {
+                        client_closed = true;
+                    }
+                }
+            }
+        }
+    }
+
+    // Release elevation permit.
+    if is_elevated {
+        let n = mux.elevated_sessions.fetch_sub(1, Ordering::Relaxed);
+        pipeline_debug::set_elevated(n.saturating_sub(1));
+    }
     Ok(())
 }
 
@@ -1479,6 +1922,20 @@ where
     }
 }
 
+/// Extract bytes from the read buffer, applying the zero-copy threshold.
+/// Reads >= half the buffer use split+freeze (zero-copy); smaller reads
+/// copy out and clear so the buffer allocation is reused.
+fn extract_bytes(buf: &mut BytesMut, n: usize) -> Bytes {
+    const ZERO_COPY_THRESHOLD: usize = 65536 / 2;
+    if n >= ZERO_COPY_THRESHOLD {
+        buf.split().freeze()
+    } else {
+        let owned = Bytes::copy_from_slice(&buf[..n]);
+        buf.clear();
+        owned
+    }
+}
+
 pub fn decode_udp_packets(resp: &TunnelResponse) -> Result<Vec<Vec<u8>>, String> {
     let Some(pkts) = resp.pkts.as_ref() else {
         return Ok(Vec::new());
@@ -1507,6 +1964,7 @@ mod tests {
             eof: None,
             e: e.map(str::to_string),
             code: code.map(str::to_string),
+            seq: None,
         }
     }
 
@@ -1714,7 +2172,7 @@ mod tests {
     /// Build a TunnelMux whose send channel is exposed to the test rather
     /// than wired to a real DomainFronter. Lets tests assert what messages
     /// the client would emit without needing network or apps_script.
-    fn mux_for_test() -> (Arc<TunnelMux>, mpsc::Receiver<MuxMsg>) {
+    fn mux_for_test() -> (Arc<TunnelMux>, mpsc::UnboundedReceiver<MuxMsg>) {
         mux_for_test_with(2)
     }
 
@@ -1722,8 +2180,8 @@ mod tests {
     /// per-deployment legacy state's aggregate gate (`all_servers_legacy`)
     /// requires `legacy_deployments.len() == num_scripts`, so tests that
     /// exercise that gate need to control how many "deployments" exist.
-    fn mux_for_test_with(num_scripts: usize) -> (Arc<TunnelMux>, mpsc::Receiver<MuxMsg>) {
-        let (tx, rx) = mpsc::channel(16);
+    fn mux_for_test_with(num_scripts: usize) -> (Arc<TunnelMux>, mpsc::UnboundedReceiver<MuxMsg>) {
+        let (tx, rx) = mpsc::unbounded_channel();
         let mux = Arc::new(TunnelMux {
             tx,
             connect_data_unsupported: Arc::new(AtomicBool::new(false)),
@@ -1741,6 +2199,8 @@ mod tests {
             // generous fixed value here; production derives this from
             // `fronter.batch_timeout()` (see `TunnelMux::start`).
             reply_timeout: Duration::from_secs(35),
+            elevated_sessions: AtomicU64::new(0),
+            max_elevated: MAX_ELEVATED_PER_DEPLOYMENT * num_scripts as u64,
         });
         (mux, rx)
     }
@@ -1802,14 +2262,17 @@ mod tests {
         let addr = listener.local_addr().unwrap();
         let accept = tokio::spawn(async move { listener.accept().await.unwrap().0 });
         let _client = TcpStream::connect(addr).await.unwrap();
-        let mut server_side = accept.await.unwrap();
+        let server_side = accept.await.unwrap();
 
         let (mux, mut rx) = mux_for_test();
         let pending = Some(Bytes::from_static(b"CLIENTHELLO"));
 
         let loop_handle = tokio::spawn({
             let mux = mux.clone();
-            async move { tunnel_loop(&mut server_side, "sid-under-test", &mux, pending).await }
+            async move {
+                let mut server_side = server_side;
+                tunnel_loop(&mut server_side, "sid-under-test", &mux, pending).await
+            }
         });
 
         // The first message tunnel_loop emits must be Data carrying the
@@ -1820,7 +2283,7 @@ mod tests {
             .expect("mux channel closed unexpectedly");
 
         match msg {
-            MuxMsg::Data { sid, data, reply } => {
+            MuxMsg::Data { sid, data, reply, .. } => {
                 assert_eq!(sid, "sid-under-test");
                 assert_eq!(&data[..], b"CLIENTHELLO");
                 // Reply with eof so tunnel_loop unwinds cleanly.
@@ -1832,6 +2295,7 @@ mod tests {
                         eof: Some(true),
                         e: None,
                         code: None,
+                        seq: Some(0),
                     },
                     "test-script".to_string(),
                 )));
@@ -1849,7 +2313,26 @@ mod tests {
             ),
         }
 
-        let _ = tokio::time::timeout(Duration::from_secs(2), loop_handle)
+        // With pipelining (INFLIGHT_OPTIMIST=2), the second op is
+        // launched after a 1 s stagger sleep, so we need to wait long
+        // enough for it to arrive. Reply to any remaining messages so the
+        // loop can exit cleanly.
+        let mut seq = 1u64;
+        while let Ok(Some(msg)) = tokio::time::timeout(Duration::from_millis(1500), rx.recv()).await {
+            if let MuxMsg::Data { reply, .. } = msg {
+                let _ = reply.send(Ok((
+                    TunnelResponse {
+                        sid: Some("sid-under-test".into()),
+                        d: None, pkts: None, eof: Some(true),
+                        e: None, code: None, seq: Some(seq),
+                    },
+                    "test-script".to_string(),
+                )));
+                seq += 1;
+            }
+        }
+
+        let _ = tokio::time::timeout(Duration::from_secs(4), loop_handle)
             .await
             .expect("tunnel_loop did not exit after eof");
     }
@@ -1873,7 +2356,7 @@ mod tests {
         let addr = listener.local_addr().unwrap();
         let accept = tokio::spawn(async move { listener.accept().await.unwrap().0 });
         let _client = TcpStream::connect(addr).await.unwrap();
-        let mut server_side = accept.await.unwrap();
+        let server_side = accept.await.unwrap();
 
         // 2 deployments, only A marked legacy → all_servers_legacy = false.
         let (mux, mut rx) = mux_for_test_with(2);
@@ -1882,7 +2365,10 @@ mod tests {
 
         let loop_handle = tokio::spawn({
             let mux = mux.clone();
-            async move { tunnel_loop(&mut server_side, "sid-mixed", &mux, None).await }
+            async move {
+                let mut server_side = server_side;
+                tunnel_loop(&mut server_side, "sid-mixed", &mux, None).await
+            }
         });
 
         // Reply to 6 empty polls, all from A. With the regression
@@ -1892,19 +2378,20 @@ mod tests {
         // the aggregate gate stays false and the loop keeps polling.
         // The 60 s timeout below is paused-time, so it only "elapses"
         // if rx.recv() truly never resolves (i.e. the loop has stalled).
-        for i in 0..6u32 {
+        let mut received = 0u32;
+        while received < 6 {
             let msg = tokio::time::timeout(Duration::from_secs(60), rx.recv())
                 .await
                 .unwrap_or_else(|_| panic!(
                     "loop stopped emitting at iteration {} — regression: per-deployment skip-when-idle stalled session even though long-poll-capable peer was available",
-                    i
+                    received
                 ))
                 .expect("mux channel closed unexpectedly");
             match msg {
-                MuxMsg::Data { sid, data, reply } => {
+                MuxMsg::Data { sid, data, seq, reply, .. } => {
                     assert_eq!(sid, "sid-mixed");
                     assert!(data.is_empty(), "expected empty poll, got {} bytes", data.len());
-                    let last = i == 5;
+                    let last = received == 5;
                     let _ = reply.send(Ok((
                         TunnelResponse {
                             sid: Some("sid-mixed".into()),
@@ -1913,13 +2400,15 @@ mod tests {
                             eof: if last { Some(true) } else { None },
                             e: None,
                             code: None,
+                            seq,
                         },
                         "script-A".to_string(),
                     )));
+                    received += 1;
                 }
                 _ => panic!(
                     "iteration {}: expected Data poll, got a different MuxMsg variant",
-                    i
+                    received
                 ),
             }
         }
@@ -2120,6 +2609,8 @@ mod tests {
             port: None,
             data: Some(Bytes::from_static(b"x")),
             encode_empty: false,
+            seq: None,
+            wseq: None,
         };
         let mk_reply = || oneshot::channel::<Result<(TunnelResponse, String), String>>().0;
 
@@ -2165,6 +2656,8 @@ mod tests {
             port: None,
             data: Some(Bytes::from_static(b"hello")),
             encode_empty: false,
+            seq: None,
+            wseq: None,
         };
         let b = encode_pending(op);
         assert_eq!(b.op, "data");
@@ -2182,6 +2675,8 @@ mod tests {
             port: None,
             data: None,
             encode_empty: false,
+            seq: None,
+            wseq: None,
         };
         assert!(encode_pending(empty_poll).d.is_none());
 
@@ -2193,6 +2688,8 @@ mod tests {
             port: None,
             data: None,
             encode_empty: false,
+            seq: None,
+            wseq: None,
         };
         assert!(encode_pending(udp_poll).d.is_none());
 
@@ -2204,6 +2701,8 @@ mod tests {
             port: None,
             data: None,
             encode_empty: false,
+            seq: None,
+            wseq: None,
         };
         assert!(encode_pending(close).d.is_none());
     }
@@ -2221,6 +2720,8 @@ mod tests {
             port: Some(443),
             data: Some(Bytes::new()),
             encode_empty: true,
+            seq: None,
+            wseq: None,
         };
         let b = encode_pending(op);
         assert_eq!(b.op, "connect_data");
@@ -2236,6 +2737,8 @@ mod tests {
             port: Some(443),
             data: Some(Bytes::from_static(b"\x16\x03\x01")), // ClientHello prefix
             encode_empty: true,
+            seq: None,
+            wseq: None,
         };
         let b = encode_pending(op);
         assert_eq!(b.d.as_deref(), Some(B64.encode(b"\x16\x03\x01").as_str()));
@@ -2260,4 +2763,107 @@ mod tests {
         // Five record_* calls, so trigger counter is at 5.
         assert_eq!(mux.preread_total_events.load(Ordering::Relaxed), 5);
     }
+
+    /// Client data written to the socket *during* the reply wait must be
+    /// buffered and sent in a subsequent op — not blocked until the reply
+    /// arrives and a fresh read-timeout elapses.
+    #[tokio::test]
+    async fn tunnel_loop_reads_client_during_reply_wait() {
+        use tokio::io::AsyncWriteExt;
+        use tokio::net::TcpListener;
+
+        let listener = TcpListener::bind(("127.0.0.1", 0)).await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let accept = tokio::spawn(async move { listener.accept().await.unwrap().0 });
+        let mut client = TcpStream::connect(addr).await.unwrap();
+        let server_side = accept.await.unwrap();
+
+        let (mux, mut rx) = mux_for_test();
+
+        let loop_handle = tokio::spawn({
+            let mux = mux.clone();
+            async move {
+                let mut server_side = server_side;
+                tunnel_loop(&mut server_side, "sid-overlap", &mux, None).await
+            }
+        });
+
+        // With pipelining (N=2), the loop may send two ops before we
+        // can write client data. Collect all initial ops, reply to each,
+        // then write data and check a subsequent op carries it.
+        let mut pending_replies: Vec<BatchedReply> = Vec::new();
+        let mut seq: u64 = 0;
+
+        // Drain initial ops (up to N=2).
+        while let Ok(Some(msg)) = tokio::time::timeout(Duration::from_millis(500), rx.recv()).await {
+            if let MuxMsg::Data { reply, .. } = msg {
+                pending_replies.push(reply);
+            }
+            if pending_replies.len() >= INFLIGHT_ACTIVE { break; }
+        }
+
+        // Write client data while replies are pending.
+        client.write_all(b"UPLOAD_DATA").await.unwrap();
+        client.flush().await.unwrap();
+        tokio::time::sleep(Duration::from_millis(50)).await;
+
+        // Reply to all pending ops (no eof, no data).
+        for reply in pending_replies.drain(..) {
+            let _ = reply.send(Ok((
+                TunnelResponse {
+                    sid: Some("sid-overlap".into()),
+                    d: None, pkts: None, eof: None,
+                    e: None, code: None, seq: Some(seq),
+                },
+                "test-script".to_string(),
+            )));
+            seq += 1;
+        }
+
+        // Now check that a subsequent op carries the buffered upload data.
+        let mut found_upload = false;
+        for _ in 0..4 {
+            let msg = match tokio::time::timeout(Duration::from_secs(2), rx.recv()).await {
+                Ok(Some(m)) => m,
+                _ => break,
+            };
+            if let MuxMsg::Data { data, reply, .. } = msg {
+                if &data[..] == b"UPLOAD_DATA" {
+                    found_upload = true;
+                }
+                let _ = reply.send(Ok((
+                    TunnelResponse {
+                        sid: Some("sid-overlap".into()),
+                        d: None, pkts: None,
+                        eof: Some(found_upload),
+                        e: None, code: None, seq: Some(seq),
+                    },
+                    "test-script".to_string(),
+                )));
+                seq += 1;
+                if found_upload { break; }
+            }
+        }
+        assert!(found_upload, "upload data must appear in a subsequent op");
+
+        // Drain any remaining in-flight ops (stagger sleep is 1 s,
+        // so allow enough time for late-arriving ops).
+        while let Ok(Some(msg)) = tokio::time::timeout(Duration::from_millis(1500), rx.recv()).await {
+            if let MuxMsg::Data { reply, .. } = msg {
+                let _ = reply.send(Ok((
+                    TunnelResponse {
+                        sid: Some("sid-overlap".into()),
+                        d: None, pkts: None, eof: Some(true),
+                        e: None, code: None, seq: Some(seq),
+                    },
+                    "test-script".to_string(),
+                )));
+                seq += 1;
+            }
+        }
+
+        let _ = tokio::time::timeout(Duration::from_secs(4), loop_handle)
+            .await
+            .expect("tunnel_loop did not exit after eof");
+    }
 }
diff --git a/tunnel-node/src/main.rs b/tunnel-node/src/main.rs
index e3bc3c0..3380908 100644
--- a/tunnel-node/src/main.rs
+++ b/tunnel-node/src/main.rs
@@ -71,7 +71,7 @@ const STRAGGLER_SETTLE_MAX: Duration = Duration::from_millis(1000);
 /// `BATCH_TIMEOUT` (30 s) and Apps Script's UrlFetch ceiling (~60 s).
 /// Tested on censored networks in Iran where users reported smoother
 /// Telegram video playback and fewer session resets at this value.
-const LONGPOLL_DEADLINE: Duration = Duration::from_secs(15);
+const LONGPOLL_DEADLINE: Duration = Duration::from_secs(4);
 
 /// Bound on each UDP session's inbound queue. Beyond this we drop oldest
 /// to keep recent voice/media packets moving — a stale RTP frame is
@@ -153,6 +153,11 @@ struct SessionInner {
     /// to wake the drain phase as soon as any session has something to
     /// ship, replacing the old fixed-sleep heuristic.
     notify: Notify,
+    /// Sequence-ordered write buffer: pipelined data ops may arrive
+    /// out of order (different batches completing at different times).
+    /// We buffer out-of-order writes and flush in seq order.
+    next_write_seq: Mutex<Option<u64>>,
+    pending_writes: Mutex<std::collections::BTreeMap<u64, Vec<u8>>>,
 }
 
 struct ManagedSession {
@@ -212,6 +217,8 @@ async fn create_session(host: &str, port: u16) -> std::io::Result<ManagedSession
         eof: AtomicBool::new(false),
         last_active: Mutex::new(Instant::now()),
         notify: Notify::new(),
+        next_write_seq: Mutex::new(None),
+        pending_writes: Mutex::new(std::collections::BTreeMap::new()),
     });
 
     let inner_ref = inner.clone();
@@ -231,6 +238,8 @@ fn create_udpgw_session() -> ManagedSession {
         eof: AtomicBool::new(false),
         last_active: Mutex::new(Instant::now()),
         notify: Notify::new(),
+        next_write_seq: Mutex::new(None),
+        pending_writes: Mutex::new(std::collections::BTreeMap::new()),
     });
 
     let inner_ref = inner.clone();
@@ -241,7 +250,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; 2 * 1024 * 1024];
     loop {
         match reader.read(&mut buf).await {
             Ok(0) => {
@@ -643,17 +652,19 @@ struct TunnelResponse {
     #[serde(skip_serializing_if = "Option::is_none")] eof: Option<bool>,
     #[serde(skip_serializing_if = "Option::is_none")] e: Option<String>,
     #[serde(skip_serializing_if = "Option::is_none")] code: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")] seq: Option<u64>,
 }
 
 impl TunnelResponse {
     fn error(msg: impl Into<String>) -> Self {
-        Self { sid: None, d: None, pkts: None, eof: None, e: Some(msg.into()), code: None }
+        Self { sid: None, d: None, pkts: None, eof: None, e: Some(msg.into()), code: None, seq: None }
     }
     fn unsupported_op(op: &str) -> Self {
         Self {
             sid: None, d: None, pkts: None, eof: None,
             e: Some(format!("unknown op: {}", op)),
             code: Some(CODE_UNSUPPORTED_OP.into()),
+            seq: None,
         }
     }
 }
@@ -675,6 +686,8 @@ struct BatchOp {
     #[serde(default)] host: Option<String>,
     #[serde(default)] port: Option<u16>,
     #[serde(default)] d: Option<String>, // base64 data
+    #[serde(default)] seq: Option<u64>,
+    #[serde(default)] wseq: Option<u64>,
 }
 
 #[derive(Serialize)]
@@ -797,8 +810,8 @@ async fn handle_batch(
     // map lock isn't held across the per-session read_buf / packets
     // mutex acquisition — without this, every other batch (and every
     // connect/close op) head-of-line-blocks behind the drain.
-    let mut tcp_drains: Vec<(usize, String, Arc<SessionInner>)> = Vec::new();
-    let mut udp_drains: Vec<(usize, String, Arc<UdpSessionInner>)> = Vec::new();
+    let mut tcp_drains: Vec<(usize, String, Arc<SessionInner>, Option<u64>)> = Vec::new();
+    let mut udp_drains: Vec<(usize, String, Arc<UdpSessionInner>, Option<u64>)> = Vec::new();
     // True iff the batch contained any op that performed a real action
     // upstream — a new connection or a non-empty data write. A batch of
     // only empty "data" / "udp_data" polls (and possibly closes) leaves
@@ -893,15 +906,60 @@ async fn handle_batch(
                             };
                             if !bytes.is_empty() {
                                 had_writes_or_connects = true;
-                                let mut w = inner.writer.lock().await;
-                                let _ = w.write_all(&bytes).await;
-                                let _ = w.flush().await;
+                                tracing::info!(
+                                    "session {} upload {}B wseq={:?}",
+                                    &sid[..sid.len().min(8)], bytes.len(), op.wseq,
+                                );
+                                match op.wseq {
+                                    None => {
+                                        // Old client (no wseq): write immediately.
+                                        let mut w = inner.writer.lock().await;
+                                        let _ = w.write_all(&bytes).await;
+                                        let _ = w.flush().await;
+                                    }
+                                    Some(wseq) => {
+                                        let mut nws = inner.next_write_seq.lock().await;
+                                        let expected = nws.get_or_insert(wseq);
+
+                                        if wseq < *expected {
+                                            // Stale / duplicate — skip.
+                                            tracing::debug!(
+                                                "session {} wseq {} < expected {} — skipping",
+                                                &sid[..sid.len().min(8)], wseq, *expected,
+                                            );
+                                        } else if wseq == *expected {
+                                            // In order — write immediately.
+                                            let mut w = inner.writer.lock().await;
+                                            let _ = w.write_all(&bytes).await;
+                                            *expected += 1;
+
+                                            // Flush any buffered writes that
+                                            // are now in sequence.
+                                            let mut pw = inner.pending_writes.lock().await;
+                                            while let Some(entry) = pw.first_entry() {
+                                                if *entry.key() != *expected { break; }
+                                                let (_, buffered) = entry.remove_entry();
+                                                let _ = w.write_all(&buffered).await;
+                                                *expected += 1;
+                                            }
+                                            let _ = w.flush().await;
+                                        } else {
+                                            // Out of order — buffer for later.
+                                            tracing::debug!(
+                                                "session {} wseq {} > expected {} — buffering",
+                                                &sid[..sid.len().min(8)], wseq, *expected,
+                                            );
+                                            let mut pw = inner.pending_writes.lock().await;
+                                            pw.insert(wseq, bytes);
+                                        }
+                                    }
+                                }
                             }
                         }
                     }
-                    tcp_drains.push((i, sid, inner));
+                    tcp_drains.push((i, sid, inner, op.seq));
                 } else {
-                    results.push((i, eof_response(sid)));
+                    results.push((i, eof_response(sid, op.seq)));
                 }
             }
             "udp_data" => {
@@ -942,9 +1000,9 @@ async fn handle_batch(
                     if had_uplink {
                         *inner.last_active.lock().await = Instant::now();
                     }
-                    udp_drains.push((i, sid, inner));
+                    udp_drains.push((i, sid, inner, op.seq));
                 } else {
-                    results.push((i, eof_response(sid)));
+                    results.push((i, eof_response(sid, op.seq)));
                 }
             }
             "close" => {
@@ -964,11 +1022,11 @@ async fn handle_batch(
         match join {
             Ok((i, NewConn::Connect(r))) => results.push((i, r)),
             Ok((i, NewConn::ConnectData(Ok((sid, inner))))) => {
-                tcp_drains.push((i, sid, inner));
+                tcp_drains.push((i, sid, inner, None));
             }
             Ok((i, NewConn::ConnectData(Err(r)))) => results.push((i, r)),
             Ok((i, NewConn::UdpOpen(Ok((sid, inner))))) => {
-                udp_drains.push((i, sid, inner));
+                udp_drains.push((i, sid, inner, None));
             }
             Ok((i, NewConn::UdpOpen(Err(r)))) => results.push((i, r)),
             Err(e) => {
@@ -999,9 +1057,9 @@ async fn handle_batch(
         // don't need to re-acquire the sessions map lock here. Cloning
         // the Arc is just a refcount bump.
         let tcp_inners: Vec<Arc<SessionInner>> =
-            tcp_drains.iter().map(|(_, _, inner)| inner.clone()).collect();
+            tcp_drains.iter().map(|(_, _, inner, _)| inner.clone()).collect();
         let udp_inners: Vec<Arc<UdpSessionInner>> =
-            udp_drains.iter().map(|(_, _, inner)| inner.clone()).collect();
+            udp_drains.iter().map(|(_, _, inner, _)| inner.clone()).collect();
 
         // Wake on whichever side has work first. The previous
         // `tokio::join!` was conjunctive — a TCP burst still paid the
@@ -1086,17 +1144,33 @@ async fn handle_batch(
         // Apps Script's 50 MiB response ceiling. This cap stops one session
         // short of the cliff; deferred sessions drain on the next poll.
         let mut remaining_budget: usize = BATCH_RESPONSE_BUDGET;
-        for (i, sid, inner) in &tcp_drains {
-            let (data, eof) = drain_now(inner, remaining_budget).await;
-            let drained = data.len();
-            if eof {
+        for (i, sid, inner, seq) in &tcp_drains {
+            // 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; }
+                let hit_session_cap = data.len() >= TCP_DRAIN_MAX_BYTES;
+                all_data.extend_from_slice(&data);
+                if final_eof || hit_session_cap || 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)));
+            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;
             }
         }
@@ -1119,12 +1193,12 @@ async fn handle_batch(
         // trap that motivated the TCP-side fix reappears, and tracking
         // eof from the drain return rather than the atomic catches it.
         let mut udp_eof_sids: Vec<String> = Vec::new();
-        for (i, sid, inner) in &udp_drains {
+        for (i, sid, inner, seq) in &udp_drains {
             let (packets, eof) = drain_udp_now(inner).await;
             if eof {
                 udp_eof_sids.push(sid.clone());
             }
-            results.push((*i, udp_drain_response(sid.clone(), packets, eof)));
+            results.push((*i, udp_drain_response(sid.clone(), packets, eof, *seq)));
         }
         if !udp_eof_sids.is_empty() {
             let mut sessions = state.udp_sessions.lock().await;
@@ -1147,7 +1221,7 @@ async fn handle_batch(
     (StatusCode::OK, [(header::CONTENT_TYPE, "application/json")], json)
 }
 
-fn tcp_drain_response(sid: String, data: Vec<u8>, eof: bool) -> TunnelResponse {
+fn tcp_drain_response(sid: String, data: Vec<u8>, eof: bool, seq: Option<u64>) -> TunnelResponse {
     TunnelResponse {
         sid: Some(sid),
         d: if data.is_empty() { None } else { Some(B64.encode(&data)) },
@@ -1155,10 +1229,11 @@ fn tcp_drain_response(sid: String, data: Vec<u8>, eof: bool) -> TunnelResponse {
         eof: Some(eof),
         e: None,
         code: None,
+        seq,
     }
 }
 
-fn udp_drain_response(sid: String, packets: Vec<Vec<u8>>, eof: bool) -> TunnelResponse {
+fn udp_drain_response(sid: String, packets: Vec<Vec<u8>>, eof: bool, seq: Option<u64>) -> TunnelResponse {
     let pkts = if packets.is_empty() {
         None
     } else {
@@ -1171,10 +1246,11 @@ fn udp_drain_response(sid: String, packets: Vec<Vec<u8>>, eof: bool) -> TunnelRe
         eof: Some(eof),
         e: None,
         code: None,
+        seq,
     }
 }
 
-fn eof_response(sid: String) -> TunnelResponse {
+fn eof_response(sid: String, seq: Option<u64>) -> TunnelResponse {
     TunnelResponse {
         sid: Some(sid),
         d: None,
@@ -1182,6 +1258,7 @@ fn eof_response(sid: String) -> TunnelResponse {
         eof: Some(true),
         e: None,
         code: None,
+        seq,
     }
 }
 
@@ -1228,7 +1305,7 @@ async fn handle_connect(state: &AppState, host: Option<String>, port: Option<u16
     let sid = uuid::Uuid::new_v4().to_string();
     tracing::info!("session {} -> {}:{}", sid, host, port);
     state.sessions.lock().await.insert(sid.clone(), session);
-    TunnelResponse { sid: Some(sid), d: None, pkts: None, eof: Some(false), e: None, code: None }
+    TunnelResponse { sid: Some(sid), d: None, pkts: None, eof: Some(false), e: None, code: None, seq: None }
 }
 
 /// Open a session and write the client's first bytes in one round trip.
@@ -1350,6 +1427,7 @@ async fn handle_connect_data_single(
         eof: Some(eof),
         e: None,
         code: None,
+        seq: None,
     }
 }
 
@@ -1398,7 +1476,7 @@ async fn handle_data_single(state: &AppState, sid: Option<String>, data: Option<
         sid: Some(sid),
         d: if data.is_empty() { None } else { Some(B64.encode(&data)) },
         pkts: None,
-        eof: Some(eof), e: None, code: None,
+        eof: Some(eof), e: None, code: None, seq: None,
     }
 }
 
@@ -1415,7 +1493,7 @@ async fn handle_close(state: &AppState, sid: Option<String>) -> TunnelResponse {
         s.reader_handle.abort();
         tracing::info!("udp session {} closed by client", sid);
     }
-    TunnelResponse { sid: Some(sid), d: None, pkts: None, eof: Some(true), e: None, code: None }
+    TunnelResponse { sid: Some(sid), d: None, pkts: None, eof: Some(true), e: None, code: None, seq: None }
 }
 
 // ---------------------------------------------------------------------------
@@ -1736,6 +1814,8 @@ mod tests {
             eof: AtomicBool::new(false),
             last_active: Mutex::new(Instant::now()),
             notify: Notify::new(),
+            next_write_seq: Mutex::new(None),
+            pending_writes: Mutex::new(std::collections::BTreeMap::new()),
         })
     }
 
@@ -1987,6 +2067,8 @@ mod tests {
             eof: AtomicBool::new(false),
             last_active: Mutex::new(Instant::now()),
             notify: Notify::new(),
+            next_write_seq: Mutex::new(None),
+            pending_writes: Mutex::new(std::collections::BTreeMap::new()),
         });
         let _reader_handle = tokio::spawn(reader_task(reader, inner.clone()));
 
@@ -2343,7 +2425,7 @@ mod tests {
         );
 
         // The `udp_drain_response` helper threads eof into `eof: Some(true)`.
-        let resp = udp_drain_response("zombie".into(), pkts, eof);
+        let resp = udp_drain_response("zombie".into(), pkts, eof, None);
         assert_eq!(resp.eof, Some(true));
         assert!(resp.pkts.is_none());
     }