crates/skald-relay-client/src/ws.rs

@@ -8,10 +8,12 @@
 //! their own `WsMessage` variants and never appear as protobuf.
 //!
 //! Reconnection uses exponential backoff (1,2,4,…,60 s) with jitter, and the
-//! whole loop is cancellable on stop.
+//! whole loop is cancellable on stop. A live session is kept honest by the
+//! [`Liveness`] probe — without it a silently broken path parks the loop forever
+//! (see that type's docs).
 
 use std::sync::Arc;
-use std::time::Duration;
+use std::time::{Duration, Instant};
 
 use anyhow::{anyhow, Result};
 use futures_util::{SinkExt, StreamExt};
@@ -27,11 +29,65 @@ use tracing::{debug, info, warn};
 
 use crate::state::RelayState;
 
+/// How often the agent sends its **own** WS `Ping` on a live session.
+const PING_INTERVAL_SECS: u64 = 20;
+
+/// No inbound frame for this long ⇒ the session is dead; drop it and redial.
+/// Two and a half of the relay's 30 s pings, comfortably under its own 120 s
+/// idle close.
+const IDLE_TIMEOUT_SECS: u64 = 75;
+
+/// Per-session liveness knobs.
+///
+/// The probe exists because a purely *reactive* session cannot notice its own
+/// death. We answer the relay's `Ping` with a `Pong` and otherwise send nothing
+/// for long stretches, so when the path breaks silently — NAT rebinding, a
+/// reverse proxy dropping its state — there are no unacked bytes on the socket
+/// for the kernel to retransmit, no TCP error, and the relay's `Close` (it gives
+/// up after 120 s of quiet) falls into the same hole. `stream.next()` then parks
+/// forever on a socket to nobody, `is_connected()` keeps answering `true`, and
+/// the reconnect schedule below — which works fine, it just never gets asked —
+/// is never reached. Only a process restart clears it.
+///
+/// So both halves matter: `ping_every` keeps unacked bytes on the wire (the
+/// relay pongs them back, which also refreshes *its* idle timer), and
+/// `idle_after` turns silence into an `Err` and hands the session to the
+/// reconnect path.
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct Liveness {
+    /// Interval between our outbound `Ping`s.
+    ping_every: Duration,
+    /// Silence tolerated before the session is declared dead.
+    idle_after: Duration,
+}
+
+// Hand-written: a derived `Default` would give a zero `ping_every` (a hot loop)
+// and a zero `idle_after` (every session dead on arrival).
+impl Default for Liveness {
+    fn default() -> Self {
+        Self {
+            ping_every: Duration::from_secs(PING_INTERVAL_SECS),
+            idle_after: Duration::from_secs(IDLE_TIMEOUT_SECS),
+        }
+    }
+}
+
 /// Run the reconnecting WS loop until `cancel` fires (relay-protocol.md §8).
 pub(crate) async fn run_loop(
+    state: Arc<RelayState>,
+    outbound_rx: mpsc::UnboundedReceiver<Vec<u8>>,
+    cancel: CancellationToken,
+) {
+    run_loop_with(state, outbound_rx, cancel, Liveness::default()).await
+}
+
+/// [`run_loop`] with the liveness knobs spelled out (tests use short ones so a
+/// redial is observable in milliseconds).
+async fn run_loop_with(
     state: Arc<RelayState>,
     mut outbound_rx: mpsc::UnboundedReceiver<Vec<u8>>,
     cancel: CancellationToken,
+    liveness: Liveness,
 ) {
     let mut backoff_step: u32 = 0;
     loop {
@@ -39,7 +95,7 @@ pub(crate) async fn run_loop(
             return;
         }
 
-        match connect_once(&state, &mut outbound_rx, &cancel).await {
+        match connect_once(&state, &mut outbound_rx, &cancel, liveness).await {
             Ok(()) => {
                 // Clean disconnect (cancelled or graceful): reset backoff.
                 backoff_step = 0;
@@ -77,6 +133,7 @@ async fn connect_once(
     state: &Arc<RelayState>,
     outbound_rx: &mut mpsc::UnboundedReceiver<Vec<u8>>,
     cancel: &CancellationToken,
+    liveness: Liveness,
 ) -> Result<()> {
     let url = state.relay_url();
     info!(crate_name = "skald-relay-client", %url, "connecting to relay");
@@ -130,7 +187,13 @@ async fn connect_once(
     };
     sink.send(WsMessage::Binary(authorize.encode_to_vec().into())).await?;
 
-    // 5. Main dispatch loop: outbound queue, inbound frames, WS-level Ping/Pong.
+    // 5. Main dispatch loop: outbound queue, inbound frames, WS-level Ping/Pong,
+    // and the liveness probe.
+    let mut ping = tokio::time::interval(liveness.ping_every);
+    ping.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+    ping.tick().await; // consume the immediate first tick — we just handshook
+    let mut last_seen = Instant::now();
+
     loop {
         tokio::select! {
             _ = cancel.cancelled() => {
@@ -138,6 +201,20 @@ async fn connect_once(
                 return Ok(());
             }
 
+            // Liveness (see `Liveness`): probe the socket, and give up on a
+            // session that has gone quiet. Returning `Err` is what puts us back
+            // on the reconnect schedule instead of parking here forever.
+            _ = ping.tick() => {
+                let quiet = last_seen.elapsed();
+                if quiet > liveness.idle_after {
+                    return Err(anyhow!(
+                        "relay silent for {}s (no frame, not even a pong); redialing",
+                        quiet.as_secs()
+                    ));
+                }
+                sink.send(WsMessage::Ping(Vec::new().into())).await?;
+            }
+
             // Outbound: already-encoded protobuf frames queued by pairing / send
             // / revoke. The channel carries `Vec<u8>` ready to be shipped as a
             // binary WS frame.
@@ -151,6 +228,9 @@ async fn connect_once(
             // Inbound: relay → agent frames.
             maybe = stream.next() => {
                 let Some(msg) = maybe else { return Ok(()) }; // stream ended
+                // Any frame at all — data, Ping, Pong — proves the path is
+                // still there, which is the whole question the probe asks.
+                last_seen = Instant::now();
                 match msg? {
                     WsMessage::Binary(data) => {
                         handle_incoming(state, &data).await;
@@ -360,3 +440,155 @@ mod tests {
         }
     }
 }
+
+/// The liveness probe against a **silent** relay — the shape a black-holed path
+/// leaves behind, where no `Close` and no TCP error ever arrive. The fake relay
+/// completes the v2 handshake and then never speaks again; the agent has to work
+/// out on its own that the session is dead, drop it, and redial. Before the
+/// probe existed this parked forever and only a process restart cleared it.
+#[cfg(test)]
+mod net_tests {
+    use super::*;
+    use std::net::SocketAddr;
+
+    use skald_relay_common::proto::v2::{AuthOk, Challenge};
+    use sqlx::SqlitePool;
+    use tokio::io::AsyncReadExt;
+    use tokio::net::{TcpListener, TcpStream};
+
+    use crate::db;
+    use crate::identity::Identity;
+    use crate::state::StateConfig;
+
+    /// Same seed on both sides so the `AuthOk` carries the namespace the agent
+    /// expects (a mismatch is a different failure than the one under test).
+    const SEED: [u8; 32] = [0x42; 32];
+
+    /// What the harness reports about the agent's dialling behaviour.
+    #[derive(Debug)]
+    enum Event {
+        /// A TCP connection was accepted.
+        Accepted,
+        /// That connection reached EOF — i.e. the agent hung up.
+        HungUp,
+    }
+
+    /// A relay that handshakes and then goes mute.
+    async fn spawn_silent_relay(ns_raw: [u8; 32]) -> (String, mpsc::UnboundedReceiver<Event>) {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr: SocketAddr = listener.local_addr().unwrap();
+        let (tx, rx) = mpsc::unbounded_channel();
+        tokio::spawn(async move {
+            while let Ok((tcp, _)) = listener.accept().await {
+                let tx = tx.clone();
+                let _ = tx.send(Event::Accepted);
+                tokio::spawn(async move {
+                    silent_session(tcp, ns_raw).await;
+                    let _ = tx.send(Event::HungUp);
+                });
+            }
+        });
+        (format!("ws://{addr}/v1/ws"), rx)
+    }
+
+    /// Challenge → read the agent's `Auth` → `AuthOk` → total silence, until the
+    /// agent closes the socket.
+    ///
+    /// The silence is why the tail reads the **raw TCP stream** instead of
+    /// `ws.next()`: tungstenite answers an inbound `Ping` with an automatic
+    /// `Pong` flushed on the next read, which would keep the agent's `last_seen`
+    /// fresh and defeat the very condition being simulated. For the same reason
+    /// this asserts nothing about the probe frames themselves — the observable
+    /// contract is that the agent gives up and comes back.
+    async fn silent_session(tcp: TcpStream, ns_raw: [u8; 32]) {
+        let mut ws = tokio_tungstenite::accept_async(tcp).await.expect("ws accept");
+
+        let challenge = RelayFrame {
+            frame: Some(Frame::Challenge(Challenge {
+                nonce: prost::bytes::Bytes::from(vec![0x5A; 32]),
+            })),
+        };
+        ws.send(WsMessage::Binary(challenge.encode_to_vec().into())).await.unwrap();
+
+        // The agent's `Auth` is the next binary frame. It signs a nonce we chose
+        // ourselves, so there is nothing here worth verifying.
+        while let Some(Ok(msg)) = ws.next().await {
+            if matches!(msg, WsMessage::Binary(_)) {
+                break;
+            }
+        }
+
+        let ok = RelayFrame {
+            frame: Some(Frame::AuthOk(AuthOk {
+                namespace_id: prost::bytes::Bytes::copy_from_slice(&ns_raw),
+            })),
+        };
+        ws.send(WsMessage::Binary(ok.encode_to_vec().into())).await.unwrap();
+
+        // From here on we are a black hole: drain bytes, answer nothing.
+        let tcp = ws.get_mut();
+        let mut scratch = [0u8; 1024];
+        while let Ok(n) = tcp.read(&mut scratch).await {
+            if n == 0 {
+                break; // agent hung up
+            }
+        }
+    }
+
+    async fn make_state(relay_url: String) -> Arc<RelayState> {
+        let path = std::env::temp_dir()
+            .join(format!("relay-cli-liveness-{}.db", std::process::id()));
+        let pool = SqlitePool::connect(&format!("sqlite://{}?mode=rwc", path.display()))
+            .await
+            .unwrap();
+        db::init(&pool).await.unwrap();
+        let (events_tx, _) = tokio::sync::broadcast::channel(16);
+        Arc::new(RelayState::new(
+            Identity::from_seed(&SEED),
+            Arc::new(pool),
+            StateConfig { relay_url, pairing_ttl: 300 },
+            events_tx,
+        ))
+    }
+
+    async fn next(rx: &mut mpsc::UnboundedReceiver<Event>) -> Event {
+        tokio::time::timeout(Duration::from_secs(10), rx.recv())
+            .await
+            .expect("timed out waiting on the relay harness")
+            .expect("relay harness gone")
+    }
+
+    #[tokio::test]
+    async fn silent_relay_is_dropped_and_redialed() {
+        let ns_raw = Identity::from_seed(&SEED).namespace_id_raw();
+        let (url, mut events) = spawn_silent_relay(ns_raw).await;
+        let state = make_state(url).await;
+
+        let (out_tx, out_rx) = mpsc::unbounded_channel::<Vec<u8>>();
+        state.set_outbound(out_tx);
+        let cancel = CancellationToken::new();
+        // Production values scaled down ~200×; the ratio is what matters.
+        let liveness = Liveness {
+            ping_every: Duration::from_millis(100),
+            idle_after: Duration::from_millis(400),
+        };
+        let task = {
+            let state = Arc::clone(&state);
+            let cancel = cancel.clone();
+            tokio::spawn(async move { run_loop_with(state, out_rx, cancel, liveness).await })
+        };
+
+        assert!(matches!(next(&mut events).await, Event::Accepted), "agent should dial");
+        assert!(
+            matches!(next(&mut events).await, Event::HungUp),
+            "agent parked on a mute socket instead of giving up on it",
+        );
+        assert!(
+            matches!(next(&mut events).await, Event::Accepted),
+            "agent dropped the dead session but never dialled again",
+        );
+
+        cancel.cancel();
+        let _ = tokio::time::timeout(Duration::from_secs(5), task).await;
+    }
+}