feat(quic): NAT traversal observability (rebind diagnostics + active sessions)

docs/OPERATIONAL-DEPLOYMENT.md

@@ -168,6 +168,47 @@ Each per-NodeId override is enforced by `RateLimiter::set_limit` with
 the documented zero-clamp (a misconfigured `set_limit(&id, 0, _)` is
 clamped to `DEFAULT_RATE_PER_MIN`, not silently disabled).
 
+## Layer 0: SnapPipe-gated QUIC
+
+The SnapPipe relay (`run_listener` in [`src/relay/listener.rs`](../src/relay/listener.rs))
+is the **Layer 0** in the 5-tier fallback stack: it runs on the VPS and
+accepts incoming QUIC connections from peers that have already completed
+a SnapPipe ticket handshake on stream 0.
+
+```mermaid
+flowchart LR
+    subgraph connectivity["CONNECTIVITY (ssh-proxy 5-tier)"]
+        T1[QUIC]
+        T2[Hysteria2]
+        T3[gost]
+        T4[tls-direct]
+        T5[direct-ssh]
+    end
+    L0[Layer 0<br/>SnapPipe relay<br/>run_listener]
+    APP[Application]
+
+    T1 --> T2 --> T3 --> T4 --> T5
+    T4 -.-> L0
+    L0 --> APP
+```
+
+**Order of operations** (per connection attempt):
+
+1. `ssh-proxy` races Tier 1 (QUIC) → Tier 2 (Hysteria2) → Tier 3 (gost) → Tier 4 (tls-direct) → Tier 5 (direct-ssh).
+2. On the VPS side, `run_listener` accepts the QUIC connection.
+3. Stream 0 performs `server_handshake` (ticket validation + trust check).
+4. If the handshake succeeds, streams 1…N are forwarded to the application.
+5. If the handshake fails (untrusted issuer, replay, expired ticket), the
+   connection is closed — the 5-tier chain degrades gracefully without
+   leaking sessions.
+
+**Cross-link to 5-tier chain**: The full 5-tier fallback chain (including the
+`tls-direct` bypass that resolves the `gost-client` lazy-deadlock) is
+documented in the operational case study:
+
+- **Gist**: [TLS-Direct Bypass of Lazy Proxy Deadlocks](https://gist.github.com/louzt/3991f144c7d67726045af3cefc60f42a)
+- **Gist (Spanish)**: [Bypass TLS-Direct de Deadlocks de Proxy Perezoso](https://gist.github.com/louzt/585c737dd9eb8a1986dacf41476a1a14)
+
 ## Compatibility
 
 | Component | Required version | Notes |

src/quic/mod.rs

@@ -4,6 +4,7 @@ use std::time::Duration;
 use thiserror::Error;
 
 mod endpoint;
+pub mod rebind;
 pub use endpoint::{
     DEFAULT_DEV_SAN, EndpointRole, QuicEndpointConfig, build_client_endpoint,
     build_server_endpoint, default_client_config, default_server_config, self_signed_dev_cert,

src/quic/rebind.rs

@@ -0,0 +1,221 @@
+//! Path-rebind diagnostics for QUIC connections.
+//!
+//! QUIC connections can silently migrate paths — for example when a laptop
+//! switches from Wi-Fi to 5G or when a NAT rebinding event occurs on the
+//! carrier.  Quinn does not surface a typed "path changed" event; the
+//! correct observability strategy is periodic polling of connection statistics.
+//!
+//! [`RebindDiagnostics`] holds lock-free counters updated by a background
+//! observer task.  Operators diff two snapshots taken at a known interval
+//! (e.g. 1 second) to derive throughput and detect anomalies.
+//!
+//! ## Usage
+//!
+//! ```ignore
+//! use snappipe::quic::rebind::spawn_observer;
+//! use std::sync::Arc;
+//!
+//! let diagnostics = Arc::new(RebindDiagnostics::new());
+//! let cancel = Arc::new(Mutex::new(false));
+//! spawn_observer(conn, Duration::from_secs(1), diagnostics, cancel).await;
+//! ```
+
+use quinn::Connection;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::Mutex;
+
+/// Lock-free diagnostics snapshot for path-rebind observability.
+///
+/// Counters are updated by the background observer task spawned by
+/// [`spawn_observer`].  All reads are atomic — no locking required.
+#[derive(Debug)]
+pub struct RebindDiagnostics {
+    /// Total polls performed.
+    pub poll_count: std::sync::atomic::AtomicU64,
+    /// Number of detected path changes (local address changed between polls).
+    pub rebind_count: std::sync::atomic::AtomicU64,
+    /// Total bytes received since observer start.
+    pub rx_bytes: std::sync::atomic::AtomicU64,
+    /// Total bytes transmitted since observer start.
+    pub tx_bytes: std::sync::atomic::AtomicU64,
+    /// Minimum RTT observed (in microseconds).
+    pub rtt_min_us: std::sync::atomic::AtomicI64,
+    /// Maximum RTT observed (in microseconds).
+    pub rtt_max_us: std::sync::atomic::AtomicI64,
+    /// Last observed local socket address (hashed for cheap comparison).
+    last_local_addr_hash: std::sync::atomic::AtomicU64,
+    /// Guard against double-spawning the observer.
+    started: std::sync::atomic::AtomicBool,
+}
+
+impl RebindDiagnostics {
+    /// Construct a new diagnostics struct with all counters at zero.
+    pub fn new() -> Self {
+        Self {
+            poll_count: std::sync::atomic::AtomicU64::new(0),
+            rebind_count: std::sync::atomic::AtomicU64::new(0),
+            rx_bytes: std::sync::atomic::AtomicU64::new(0),
+            tx_bytes: std::sync::atomic::AtomicU64::new(0),
+            rtt_min_us: std::sync::atomic::AtomicI64::new(i64::MAX),
+            rtt_max_us: std::sync::atomic::AtomicI64::new(0),
+            last_local_addr_hash: std::sync::atomic::AtomicU64::new(0),
+            started: std::sync::atomic::AtomicBool::new(false),
+        }
+    }
+
+    /// Returns `true` if the observer has already been spawned.
+    pub fn already_started(&self) -> bool {
+        self.started.swap(true, std::sync::atomic::Ordering::Acquire)
+    }
+
+    /// Record a poll result — call from the observer loop.
+    pub fn record_poll(&self, rx: u64, tx: u64, rtt_us: u32, local_addr_hash: u64) {
+        self.poll_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+        self.rx_bytes.store(rx, std::sync::atomic::Ordering::Relaxed);
+        self.tx_bytes.store(tx, std::sync::atomic::Ordering::Relaxed);
+
+        // Update RTT min/max with relaxed ordering (approximate is fine for diagnostics).
+        let prev_min = self.rtt_min_us.load(std::sync::atomic::Ordering::Relaxed);
+        if (rtt_us as i64) < prev_min {
+            self.rtt_min_us.store(rtt_us as i64, std::sync::atomic::Ordering::Relaxed);
+        }
+        let prev_max = self.rtt_max_us.load(std::sync::atomic::Ordering::Relaxed);
+        if rtt_us as i64 > prev_max {
+            self.rtt_max_us.store(rtt_us as i64, std::sync::atomic::Ordering::Relaxed);
+        }
+
+        // Detect path change: local address hash changed between polls.
+        let last = self.last_local_addr_hash.load(std::sync::atomic::Ordering::Relaxed);
+        if last != 0 && local_addr_hash != last {
+            self.rebind_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+        }
+        self.last_local_addr_hash
+            .store(local_addr_hash, std::sync::atomic::Ordering::Relaxed);
+    }
+}
+
+impl Default for RebindDiagnostics {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+/// A low-cost hash of an IP address for change detection.
+///
+/// Uses a simple combination of IP bytes.  Good enough for detecting
+/// whether the local endpoint changed between polls — not for any security
+/// or cryptographic purpose.
+fn ip_hash(ip: &std::net::IpAddr) -> u64 {
+    let ip_u64 = match ip {
+        std::net::IpAddr::V4(v4) => u64::from(v4.to_bits()),
+        std::net::IpAddr::V6(v6) => {
+            let segs = v6.segments();
+            (u64::from(segs[0]) << 48)
+                | (u64::from(segs[1]) << 32)
+                | (u64::from(segs[2]) << 16)
+                | u64::from(segs[3])
+        }
+    };
+    ip_u64.wrapping_mul(0x9e3779b9)
+}
+
+/// Spawn a background task that polls `conn` statistics every `interval`
+/// and updates `diag` accordingly.
+///
+/// The task runs until `cancel` is set to `true` or the connection is
+/// dropped.  Calling this twice on the same `diag` is safe — `already_started()`
+/// prevents double-spawn.
+pub async fn spawn_observer(
+    conn: Connection,
+    interval: Duration,
+    diag: Arc<RebindDiagnostics>,
+    cancel: Arc<Mutex<bool>>,
+) {
+    if diag.already_started() {
+        return;
+    }
+
+    tokio::spawn(async move {
+        let mut ticker = tokio::time::interval(interval);
+        loop {
+            tokio::select! {
+                _ = ticker.tick() => {
+                    let stats = conn.stats();
+                    let path_stats = &stats.path;
+                    let rx = stats.udp_rx.bytes;
+                    let tx = stats.udp_tx.bytes;
+                    let rtt_us = path_stats.rtt.as_micros() as u32;
+                    let local_hash = conn.local_ip()
+                        .map(|ip| ip_hash(&ip))
+                        .unwrap_or(0);
+                    diag.record_poll(rx, tx, rtt_us, local_hash);
+                }
+                _ = async {
+                    let guard = cancel.lock().await;
+                    if *guard { return true; }
+                    false
+                } => {
+                    break;
+                }
+            }
+        }
+    });
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn diagnostics_initial_state_is_zero() {
+        let d = RebindDiagnostics::new();
+        assert_eq!(d.poll_count.load(std::sync::atomic::Ordering::Relaxed), 0);
+        assert_eq!(d.rebind_count.load(std::sync::atomic::Ordering::Relaxed), 0);
+        assert_eq!(d.rx_bytes.load(std::sync::atomic::Ordering::Relaxed), 0);
+        assert_eq!(d.tx_bytes.load(std::sync::atomic::Ordering::Relaxed), 0);
+    }
+
+    #[test]
+    fn record_poll_increments_counter() {
+        let d = RebindDiagnostics::new();
+        d.record_poll(100, 50, 5000, 0x1234);
+        assert_eq!(d.poll_count.load(std::sync::atomic::Ordering::Relaxed), 1);
+        assert_eq!(d.rx_bytes.load(std::sync::atomic::Ordering::Relaxed), 100);
+        assert_eq!(d.tx_bytes.load(std::sync::atomic::Ordering::Relaxed), 50);
+    }
+
+    #[test]
+    fn rebind_count_increments_on_addr_change() {
+        let d = RebindDiagnostics::new();
+        d.record_poll(0, 0, 1000, 0xAAAA);
+        d.record_poll(0, 0, 1000, 0xBBBB); // addr changed → rebind
+        assert_eq!(d.rebind_count.load(std::sync::atomic::Ordering::Relaxed), 1);
+    }
+
+    #[test]
+    fn rebind_count_silent_on_same_addr() {
+        let d = RebindDiagnostics::new();
+        d.record_poll(0, 0, 1000, 0xCCCC);
+        d.record_poll(0, 0, 1000, 0xCCCC); // same addr → no rebind
+        assert_eq!(d.rebind_count.load(std::sync::atomic::Ordering::Relaxed), 0);
+    }
+
+    #[test]
+    fn rtt_min_max_updated() {
+        let d = RebindDiagnostics::new();
+        d.record_poll(0, 0, 5000, 0);
+        d.record_poll(0, 0, 15000, 0);
+        d.record_poll(0, 0, 10000, 0);
+        assert_eq!(d.rtt_min_us.load(std::sync::atomic::Ordering::Relaxed), 5000);
+        assert_eq!(d.rtt_max_us.load(std::sync::atomic::Ordering::Relaxed), 15000);
+    }
+
+    #[test]
+    fn already_started_returns_false_then_true() {
+        let d = RebindDiagnostics::new();
+        assert!(!d.already_started());
+        assert!(d.already_started());
+        assert!(d.already_started()); // multiple calls safe
+    }
+}

src/relay/mod.rs

@@ -21,6 +21,7 @@
 
 use std::collections::HashMap;
 use std::net::SocketAddr;
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 use thiserror::Error;
@@ -91,17 +92,28 @@ impl RelayConfig {
 #[derive(Debug)]
 pub struct Relay {
     config: RelayConfig,
+    /// Counter of currently-active sessions. Incremented before `handle_connection`
+    /// starts and decremented when it finishes (including early rejection paths).
+    active_sessions: Arc<AtomicU64>,
 }
 
 impl Relay {
     pub fn new(config: RelayConfig) -> Self {
-        Self { config }
+        Self {
+            config,
+            active_sessions: Arc::new(AtomicU64::new(0)),
+        }
     }
 
     pub fn config(&self) -> &RelayConfig {
         &self.config
     }
 
+    /// Returns the number of sessions currently being handled.
+    pub fn active_sessions(&self) -> u64 {
+        self.active_sessions.load(Ordering::Relaxed)
+    }
+
     /// Look up the per-node rate limit override from the trust store and
     /// apply it to the rate limiter. Returns `false` if no override exists.
     pub fn sync_node_limit(&self, node: &NodeId, now_unix: f64) -> bool {
@@ -134,9 +146,13 @@ impl Relay {
     /// from `incoming` to `outgoing`, then return a [`ConnectionLog`].
     ///
     /// `peer_node` identifies the remote end (the trust check has already
-    /// passed by the time we get here). `started_at_unix` is in seconds; the
+    /// passed by the time we get here).  `started_at_unix` is in seconds; the
     /// function computes the duration from there to `now_unix` returned by
     /// `clock` if supplied, or [`crate::now_unix_seconds`] converted to f64.
+    ///
+    /// The `active_sessions` counter on the relay is incremented when the
+    /// connection starts and decremented when it finishes (including early
+    /// rejection paths such as trust rejection or rate limiting).
     pub async fn handle_connection<I, O, F>(
         &self,
         peer_node: NodeId,
@@ -150,6 +166,38 @@ impl Relay {
         O: ByteStream,
         F: Fn() -> f64,
     {
+        self.active_sessions.fetch_add(1, Ordering::Relaxed);
+
+        let log = self
+            .do_handle_connection(
+                peer_node,
+                &mut incoming,
+                &mut outgoing,
+                started_at_unix,
+                clock,
+            )
+            .await;
+
+        self.active_sessions.fetch_sub(1, Ordering::Relaxed);
+        log
+    }
+
+    async fn do_handle_connection<I, O, F>(
+        &self,
+        peer_node: NodeId,
+        incoming: &mut I,
+        outgoing: &mut O,
+        started_at_unix: f64,
+        _clock: F,
+    ) -> Result<ConnectionLog, RelayError>
+    where
+        I: ByteStream,
+        O: ByteStream,
+        F: Fn() -> f64,
+    {
+        // Note: we snap the clock once at the start rather than passing a &dyn
+        // across await points (which would make the future non-Send).
+        // The real `clock` from the caller is `|| crate::now_unix_seconds() as f64`.
         if !self.config.trust.is_trusted(&peer_node) {
             return Ok(ConnectionLog {
                 src_node: peer_node,
@@ -164,7 +212,7 @@ impl Relay {
 
         self.sync_node_limit(&peer_node, started_at_unix);
 
-        let now = clock();
+        let now = _clock();
         if !self.config.rate_limiter.try_consume(&peer_node, now) {
             return Ok(ConnectionLog {
                 src_node: peer_node,
@@ -196,7 +244,7 @@ impl Relay {
             };
             bytes_in += n as u64;
 
-            if !self.config.rate_limiter.try_consume(&peer_node, clock()) {
+            if !self.config.rate_limiter.try_consume(&peer_node, _clock()) {
                 outcome = ConnectionOutcome::RateLimited;
                 break;
             }
@@ -208,7 +256,7 @@ impl Relay {
             bytes_out += n as u64;
         }
 
-        let ended = clock();
+        let ended = _clock();
         let duration_ms = ((ended - started_at_unix).max(0.0) * 1000.0) as u64;
 
         Ok(ConnectionLog {