diff --git a/crates/skald-relay-client/src/ws.rs b/crates/skald-relay-client/src/ws.rs index c9cecd9..8631b13 100644 --- a/crates/skald-relay-client/src/ws.rs +++ b/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, + outbound_rx: mpsc::UnboundedReceiver>, + 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, mut outbound_rx: mpsc::UnboundedReceiver>, 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, outbound_rx: &mut mpsc::UnboundedReceiver>, 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` 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) { + 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 { + 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 { + 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::>(); + 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; + } +}