feat(tunnel): prioritize interactive mux operations

Classify tunnel mux messages by dispatch urgency before applying the coalescing wait. Plain connection opens, connect-and-send opens, and data-bearing TCP or UDP operations now bypass the short batching delay once any already queued work has been drained. Empty polling operations and close notices remain batch-friendly so idle long-poll cadence and cleanup traffic can still piggyback without forcing extra Apps Script batches.

The change leaves batch serialization, response indexing, payload-size limits, operation-count limits, deployment selection, and Apps Script quota accounting unchanged. It only decides whether the mux should wait for additional operations before processing the current group, reducing avoidable latency for interactive flows while preserving batching behavior for low-urgency traffic.

Add focused unit coverage for immediate opening and payload-carrying messages, batchable empty polls and closes, and mixed groups where one urgent operation should short-circuit the wait.

Author: maybeknott

docs/guide.md

@@ -232,6 +232,8 @@ More deployments = more total concurrency = lower per-session latency. Each batc
 - **4 MB payload cap** per batch — well under Apps Script's 50 MB limit
 - **30 s timeout** per batch — slow / dead targets can't block other sessions forever
 
+Opening/data-bearing tunnel operations bypass the short coalescing wait; empty polls and close notices stay batch-friendly.
+
 ### Full mode quick start
 
 1. Deploy [`CodeFull.gs`](../assets/apps_script/CodeFull.gs) as a Web App on **each Google account** (same steps as `Code.gs`, but use the full-mode script that forwards to your tunnel-node). One deployment per account — the 30-concurrent limit is per account, so multiple deployments on one account share the pool. To scale, use more accounts:

src/tunnel_client.rs

@@ -300,6 +300,30 @@ enum MuxMsg {
     },
 }
 
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+enum DispatchPriority {
+    Immediate,
+    Batch,
+}
+
+impl MuxMsg {
+    fn dispatch_priority(&self) -> DispatchPriority {
+        match self {
+            MuxMsg::Connect { .. } | MuxMsg::ConnectData { .. } => DispatchPriority::Immediate,
+            MuxMsg::Data { data, .. }
+            | MuxMsg::UdpOpen { data, .. }
+            | MuxMsg::UdpData { data, .. } => {
+                if data.is_empty() {
+                    DispatchPriority::Batch
+                } else {
+                    DispatchPriority::Immediate
+                }
+            }
+            MuxMsg::Close { .. } => DispatchPriority::Batch,
+        }
+    }
+}
+
 /// Raw, not-yet-encoded form of a batch operation. Lives only inside
 /// `mux_loop` and gets converted to `BatchOp` (with base64-encoded `d`)
 /// inside `fire_batch`'s spawned task — keeping the encoding work off
@@ -762,6 +786,8 @@ async fn mux_loop(mut rx: mpsc::UnboundedReceiver<MuxMsg>, fronter: Arc<DomainFr
         // queue. Once the first op lands, the adaptive coalesce loop waits
         // in `coalesce_step` increments (resetting on each new arrival, up
         // to `coalesce_max`) so concurrent ops land in the same batch.
+        // Opening/data-bearing ops bypass that wait after any already-queued
+        // work is drained; empty polls and closes remain batch-friendly.
         match rx.recv().await {
             Some(msg) => msgs.push(msg),
             None => break,
@@ -775,6 +801,9 @@ async fn mux_loop(mut rx: mpsc::UnboundedReceiver<MuxMsg>, fronter: Arc<DomainFr
                 // Reset the soft deadline — more ops are arriving.
                 soft_deadline = tokio::time::Instant::now() + coalesce_step;
             }
+            if has_immediate_priority(&msgs) {
+                break;
+            }
             let now = tokio::time::Instant::now();
             let wait_until = soft_deadline.min(hard_deadline);
             if now >= wait_until {
@@ -908,6 +937,11 @@ async fn mux_loop(mut rx: mpsc::UnboundedReceiver<MuxMsg>, fronter: Arc<DomainFr
     }
 }
 
+fn has_immediate_priority(msgs: &[MuxMsg]) -> bool {
+    msgs.iter()
+        .any(|msg| msg.dispatch_priority() == DispatchPriority::Immediate)
+}
+
 /// Per-iteration accumulator for `mux_loop`. Owns the three fields that
 /// the data-bearing arms used to mutate in lockstep, with a single
 /// `push_or_fire` entry point so the cap-then-push pattern lives in one
@@ -2607,6 +2641,86 @@ mod tests {
         );
     }
 
+    fn batched_reply_for_test() -> BatchedReply {
+        oneshot::channel::<Result<(TunnelResponse, String), String>>().0
+    }
+
+    #[test]
+    fn mux_priority_marks_opening_and_payload_ops_immediate() {
+        let (connect_reply, _connect_rx) = oneshot::channel();
+        let connect = MuxMsg::Connect {
+            host: "example.com".into(),
+            port: 443,
+            reply: connect_reply,
+        };
+        let connect_data = MuxMsg::ConnectData {
+            host: "example.com".into(),
+            port: 443,
+            data: Bytes::from_static(b"\x16\x03\x01"),
+            reply: batched_reply_for_test(),
+        };
+        let data = MuxMsg::Data {
+            sid: "sid-1".into(),
+            data: Bytes::from_static(b"payload"),
+            seq: None,
+            wseq: None,
+            reply: batched_reply_for_test(),
+        };
+        let udp_open = MuxMsg::UdpOpen {
+            host: "example.com".into(),
+            port: 53,
+            data: Bytes::from_static(b"dns"),
+            reply: batched_reply_for_test(),
+        };
+
+        assert_eq!(connect.dispatch_priority(), DispatchPriority::Immediate);
+        assert_eq!(connect_data.dispatch_priority(), DispatchPriority::Immediate);
+        assert_eq!(data.dispatch_priority(), DispatchPriority::Immediate);
+        assert_eq!(udp_open.dispatch_priority(), DispatchPriority::Immediate);
+    }
+
+    #[test]
+    fn mux_priority_keeps_empty_polls_and_closes_batchable() {
+        let empty_data = MuxMsg::Data {
+            sid: "sid-1".into(),
+            data: Bytes::new(),
+            seq: None,
+            wseq: None,
+            reply: batched_reply_for_test(),
+        };
+        let empty_udp = MuxMsg::UdpData {
+            sid: "sid-2".into(),
+            data: Bytes::new(),
+            reply: batched_reply_for_test(),
+        };
+        let close = MuxMsg::Close {
+            sid: "sid-3".into(),
+        };
+
+        assert_eq!(empty_data.dispatch_priority(), DispatchPriority::Batch);
+        assert_eq!(empty_udp.dispatch_priority(), DispatchPriority::Batch);
+        assert_eq!(close.dispatch_priority(), DispatchPriority::Batch);
+        assert!(!has_immediate_priority(&[empty_data, empty_udp, close]));
+    }
+
+    #[test]
+    fn mux_priority_detects_immediate_inside_batchable_group() {
+        let empty_poll = MuxMsg::Data {
+            sid: "sid-1".into(),
+            data: Bytes::new(),
+            seq: None,
+            wseq: None,
+            reply: batched_reply_for_test(),
+        };
+        let payload = MuxMsg::UdpData {
+            sid: "sid-2".into(),
+            data: Bytes::from_static(b"packet"),
+            reply: batched_reply_for_test(),
+        };
+
+        assert!(has_immediate_priority(&[empty_poll, payload]));
+    }
+
     #[test]
     fn should_fire_first_op_never_fires() {
         // Empty accumulator: even a single op larger than the payload cap