perf(tunnel): dynamic coalescing, O(n) upload buffer, network_preset config

Steps 3 + 4 of the perf/relay-speed series.

--- Step 3: Dynamic coalescing (tunnel_client.rs + config.rs + proxy_server.rs) ---

Replace static DEFAULT_COALESCE_STEP/MAX constants with Arc<AtomicU64> values
stored in TunnelMux. A new RttTracker (ring buffer of last 8 batch RTTs)
measures median round-trip time and auto-adjusts the coalesce window:

  Slow preset (median > 1200ms): step=150ms  max=600ms
  Fast preset (median <  700ms): step= 50ms  max=300ms

Hysteresis: 3 consecutive sub-threshold readings required to leave Slow,
preventing flapping on bursty congestion. mux_loop reads the atomics at each
new window boundary so preset changes take effect without restart.

Config: network_preset = "auto" (default) | "fast" | "slow"
  - auto starts at Fast and adapts via RttTracker
  - fast/slow lock the preset and skip RTT measurement

Explicit coalesce_step_ms / coalesce_max_ms override still honoured when set;
presence of either disables auto-adaptation.

Note: TunnelMux is only started in mode=full. Mode=apps_script relays each
connection directly through DomainFronter and does not go through this path.

New unit tests: rtt_tracker_preset_selection_slow,
rtt_tracker_preset_selection_fast, rtt_tracker_hysteresis_prevents_premature_flip.

Fixed copy_bidirectional test in proxy_server.rs: a_client was moved into
write_task then borrowed for reading — split into separate read/write halves
before spawning so both directions can be independently asserted.

--- Step 4: O(n) amortised buffered upload merge (tunnel_client.rs) ---

Change buffered_upload from Option<Bytes> to Option<BytesMut>. The old merge
path allocated a fresh buffer and copied all accumulated data on every new
upload chunk under pipeline congestion — O(n^2) total copies for n chunks.
BytesMut extends in-place (amortised O(n)); freeze() at send time is a
zero-copy Arc pointer bump.

Author: CaptainMirage

src/config.rs

@@ -112,6 +112,13 @@ pub struct Config {
     /// Hard cap on total coalesce wait (ms). 0 = use compiled default (1000ms).
     #[serde(default)]
     pub coalesce_max_ms: u16,
+    /// Adaptive coalescing preset. One of "auto" (default), "fast", or "slow".
+    /// "auto" measures batch RTT and switches automatically.
+    /// "fast" uses 50ms/300ms windows — best for broadband/fiber.
+    /// "slow" uses 150ms/600ms windows — best for slow links (Iran cable, mobile).
+    /// Leave unset or set to "auto" for automatic detection.
+    #[serde(default)]
+    pub network_preset: Option<String>,
     /// Optional explicit SNI rotation pool for outbound TLS to `google_ip`.
     /// Empty / missing = auto-expand from `front_domain` (current default of
     /// {www, mail, drive, docs, calendar}.google.com). Set to an explicit list
@@ -785,6 +792,8 @@ pub struct TomlRelay {
     #[serde(default)]
     pub coalesce_max_ms: u16,
     #[serde(default)]
+    pub network_preset: Option<String>,
+    #[serde(default)]
     pub youtube_via_relay: bool,
     #[serde(default)]
     pub normalize_x_graphql: bool,
@@ -935,6 +944,7 @@ impl From<TomlConfig> for Config {
             parallel_relay: t.relay.parallel_relay,
             coalesce_step_ms: t.relay.coalesce_step_ms,
             coalesce_max_ms: t.relay.coalesce_max_ms,
+            network_preset: t.relay.network_preset,
             sni_hosts: t.network.sni_hosts,
             fetch_ips_from_api: t.scan.fetch_ips_from_api,
             max_ips_to_scan: t.scan.max_ips_to_scan,
@@ -977,6 +987,7 @@ impl From<&Config> for TomlConfig {
                 enable_batching: c.enable_batching,
                 coalesce_step_ms: c.coalesce_step_ms,
                 coalesce_max_ms: c.coalesce_max_ms,
+                network_preset: c.network_preset.clone(),
                 youtube_via_relay: c.youtube_via_relay,
                 normalize_x_graphql: c.normalize_x_graphql,
                 disable_padding: c.disable_padding,

src/proxy_server.rs

@@ -220,6 +220,7 @@ pub struct ProxyServer {
     tunnel_mux: Option<Arc<TunnelMux>>,
     coalesce_step_ms: u64,
     coalesce_max_ms: u64,
+    network_preset: Option<String>,
 }
 
 pub struct RewriteCtx {
@@ -527,6 +528,7 @@ impl ProxyServer {
             tunnel_mux: None, // initialized in run() inside the tokio runtime
             coalesce_step_ms: if config.coalesce_step_ms > 0 { config.coalesce_step_ms as u64 } else { 10 },
             coalesce_max_ms: if config.coalesce_max_ms > 0 { config.coalesce_max_ms as u64 } else { 1000 },
+            network_preset: config.network_preset.clone(),
         })
     }
 
@@ -548,7 +550,7 @@ impl ProxyServer {
         // Initialize TunnelMux inside the runtime (tokio::spawn requires it).
         if self.rewrite_ctx.mode == Mode::Full {
             if let Some(f) = self.fronter.as_ref() {
-                self.tunnel_mux = Some(TunnelMux::start(f.clone(), self.coalesce_step_ms, self.coalesce_max_ms));
+                self.tunnel_mux = Some(TunnelMux::start(f.clone(), self.coalesce_step_ms, self.coalesce_max_ms, self.network_preset.as_deref()));
             }
         }
 
@@ -3698,51 +3700,53 @@ mod tests {
         assert!(FrontingGroupResolved::from_config(&bad).is_err());
     }
 
-    /// Verifies that copy_bidirectional_with_sizes correctly transfers data
-    /// in both directions through in-memory duplex pipes.
+    /// Verifies that copy_bidirectional_with_sizes correctly relays data in
+    /// both directions. Splits each duplex into read/write halves so the
+    /// write task and read assertions can operate independently.
     #[tokio::test(flavor = "current_thread")]
     async fn copy_bidirectional_large_buf_roundtrip() {
-        use tokio::io::{AsyncReadExt, AsyncWriteExt};
+        use tokio::io::AsyncWriteExt;
 
         // a_client <-> a_server  and  b_client <-> b_server
-        let (mut a_client, mut a_server) = tokio::io::duplex(128 * 1024);
-        let (mut b_client, mut b_server) = tokio::io::duplex(128 * 1024);
+        let (a_client, mut a_server) = tokio::io::duplex(128 * 1024);
+        let (b_client, mut b_server) = tokio::io::duplex(128 * 1024);
 
-        let payload_a = vec![0xAAu8; 32 * 1024]; // 32 KB a→b
-        let payload_b = vec![0xBBu8; 48 * 1024]; // 48 KB b→a
+        let (mut a_client_read, mut a_client_write) = tokio::io::split(a_client);
+        let (mut b_client_read, mut b_client_write) = tokio::io::split(b_client);
 
-        let pa = payload_a.clone();
-        let pb = payload_b.clone();
+        let payload_ab = vec![0xAAu8; 32 * 1024]; // a→b
+        let payload_ba = vec![0xBBu8; 48 * 1024]; // b→a
 
-        // Writer task: sends payloads and shuts down its write half.
+        let pa = payload_ab.clone();
+        let pb = payload_ba.clone();
+
+        // Writer task: fills both pipes and shuts down write halves so the
+        // relay sees EOF and can propagate the shutdown to the other side.
         let write_task = tokio::spawn(async move {
-            a_client.write_all(&pa).await.unwrap();
-            a_client.shutdown().await.unwrap();
-            b_client.write_all(&pb).await.unwrap();
-            b_client.shutdown().await.unwrap();
+            a_client_write.write_all(&pa).await.unwrap();
+            a_client_write.shutdown().await.unwrap();
+            b_client_write.write_all(&pb).await.unwrap();
+            b_client_write.shutdown().await.unwrap();
         });
 
         // Relay task: bridges a_server <-> b_server with 64 KB buffers.
         let relay_task = tokio::spawn(async move {
-            let _ = tokio::io::copy_bidirectional_with_sizes(
-                &mut a_server,
-                &mut b_server,
-                65536,
-                65536,
-            )
-            .await;
+            tokio::io::copy_bidirectional_with_sizes(&mut a_server, &mut b_server, 65536, 65536)
+                .await
+                .unwrap();
         });
 
-        // Read what came through.
-        let mut got_b = Vec::new();
-        let read_b = tokio::io::AsyncReadExt::read_to_end(&mut a_client, &mut got_b);
-        // a_client write half is done; read remaining bytes from b direction.
-        // Reconstruct handles: duplex was already consumed above; test through
-        // the relay instead by reading from the write-task's pair.
-        // (Simplified: just verify write+relay complete without panic.)
         write_task.await.unwrap();
         relay_task.await.unwrap();
-        drop(read_b); // readers already at EOF after shutdown
-        // Passed: no panic, bidirectional copy completed cleanly.
+
+        // After relay finishes it has propagated each shutdown to the opposite
+        // side, so both client read halves are at EOF.
+        let mut got_at_b = Vec::new();
+        b_client_read.read_to_end(&mut got_at_b).await.unwrap();
+        assert_eq!(got_at_b, payload_ab, "data written to A must arrive at B");
+
+        let mut got_at_a = Vec::new();
+        a_client_read.read_to_end(&mut got_at_a).await.unwrap();
+        assert_eq!(got_at_a, payload_ba, "data written to B must arrive at A");
     }
 }