fix(compression): zero-failure handshake negotiation

Client starts with ops + zc:1 (never zops on first batch).
Tunnel-node sees zc → responds with r + zc:1.
Client sees zc:1 → enables zops for all subsequent batches.

No mismatch detection, no fallback, no retries. Old tunnel-nodes
just ignore zc and return normal r — client stays on ops forever.

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

Author: yyoyoian-pixel

src/domain_fronter.rs

@@ -412,7 +412,7 @@ pub struct DomainFronter {
     /// payloads. Mirrors `Config::disable_padding` (#391). Default false
     /// (padding active = stronger DPI defense at +25% bandwidth cost).
     disable_padding: bool,
-    zstd_unsupported: Arc<AtomicBool>,
+    zstd_enabled: Arc<AtomicBool>,
     /// Per-instance auto-blacklist tuning. Mirrors `Config::auto_blacklist_*`
     /// (#391, #444). Cached here so the hot path in `record_timeout_strike`
     /// doesn't have to reach back through the Config (which we don't keep
@@ -631,7 +631,7 @@ impl DomainFronter {
             today_bytes: AtomicU64::new(0),
             today_key: std::sync::Mutex::new(current_pt_day_key()),
             disable_padding: config.disable_padding,
-            zstd_unsupported: Arc::new(AtomicBool::new(false)),
+            zstd_enabled: Arc::new(AtomicBool::new(false)),
             auto_blacklist_strikes: config.auto_blacklist_strikes.max(1),
             auto_blacklist_window: Duration::from_secs(
                 config.auto_blacklist_window_secs.clamp(1, 3600),
@@ -697,13 +697,6 @@ impl DomainFronter {
         self.batch_timeout
     }
 
-    pub(crate) fn mark_zstd_unsupported(&self) {
-        if !self.zstd_unsupported.load(Ordering::Relaxed) {
-            tracing::warn!("zstd batch compression not supported by deployment, falling back to uncompressed");
-            self.zstd_unsupported.store(true, Ordering::Relaxed);
-        }
-    }
-
     /// Record one relay call toward the daily budget. Called once per
     /// outbound Apps Script fetch. Rolls over both daily counters at
     /// 00:00 Pacific Time, matching Apps Script's quota reset cadence
@@ -3118,23 +3111,20 @@ impl DomainFronter {
         let mut map = serde_json::Map::new();
         map.insert("k".into(), Value::String(self.auth_key.clone()));
         map.insert("t".into(), Value::String("batch".into()));
-        let use_zstd = !self.zstd_unsupported.load(Ordering::Relaxed);
-        if use_zstd {
+        if self.zstd_enabled.load(Ordering::Relaxed) {
             let ops_json = serde_json::to_vec(ops)?;
             match zstd::encode_all(ops_json.as_slice(), 3) {
                 Ok(compressed) => {
                     map.insert("zops".into(), Value::String(B64.encode(&compressed)));
-                    map.insert("zc".into(), Value::Number(1.into()));
                 }
                 Err(_) => {
                     map.insert("ops".into(), serde_json::to_value(ops)?);
-                    map.insert("zc".into(), Value::Number(1.into()));
                 }
             }
         } else {
             map.insert("ops".into(), serde_json::to_value(ops)?);
-            map.insert("zc".into(), Value::Number(1.into()));
         }
+        map.insert("zc".into(), Value::Number(1.into()));
         if !self.disable_padding {
             add_random_pad(&mut map);
         }
@@ -3278,9 +3268,9 @@ impl DomainFronter {
                         }
                     }
                 }
-                if resp.zc.is_some() && self.zstd_unsupported.load(Ordering::Relaxed) {
-                    tracing::info!("tunnel-node advertised zstd capability, re-enabling compressed batches");
-                    self.zstd_unsupported.store(false, Ordering::Relaxed);
+                if resp.zc.is_some() && !self.zstd_enabled.load(Ordering::Relaxed) {
+                    tracing::info!("tunnel-node supports zstd, enabling compressed batches");
+                    self.zstd_enabled.store(true, Ordering::Relaxed);
                 }
                 Ok(resp)
             }

src/tunnel_client.rs

@@ -1105,9 +1105,6 @@ async fn fire_batch(
                             "batch response mismatch: idx={} but r.len()={} (sent {} ops) from script {}",
                             idx, batch_resp.r.len(), n_ops, sid_short,
                         );
-                        if batch_resp.r.is_empty() && n_ops > 0 {
-                            f.mark_zstd_unsupported();
-                        }
                         let _ = reply.send(Err(format!(
                             "missing response in batch from script {}",
                             sid_short