A silent relay wire is detected: quiet wires are pinged, unanswered pings retire the connection

core/src/relay.rs

@@ -6,21 +6,43 @@
 //! path browsers use, since the TLS keying-material shortcut needs an
 //! exporter the WebSocket surface does not carry. The challenge signature
 //! is produced by the webcrypto identity handle.
+//!
+//! Liveness is a mechanism here and a policy elsewhere: the connection
+//! stamps every inbound frame and offers [`RelayConn::probe_step`] and
+//! [`RelayConn::send_ping`], while the cadence and the deadline belong
+//! to the caller driving the steps.
 
-use std::cell::RefCell;
+use std::cell::{Cell, RefCell};
 use std::collections::VecDeque;
+use std::time::{Duration, Instant};
 
 use crate::bindings::polymorph::websocket::connections::Websocket;
 use crate::bindings::polymorph::websocket::types::Message as WsMessage;
 use crate::crypto::sign::Identity;
 use crate::relay_frames::{self as frames, tag};
 
+/// The verdict of one [`RelayConn::probe_step`].
+pub enum ProbeStep {
+    /// The wire is either busy or waiting on an in-flight ping.
+    Healthy,
+    /// The wire is quiet; send a ping carrying this payload.
+    Ping([u8; 8]),
+    /// A ping went unanswered past its deadline; the wire is dead.
+    Dead,
+}
+
 /// A connected, authenticated relay client.
 pub struct RelayConn {
     ws: Websocket,
     /// Datagrams decoded but not yet delivered (a batch frame carries
     /// several).
     pending: RefCell<VecDeque<frames::Datagram>>,
+    /// When this connection last saw any inbound websocket frame.
+    last_inbound: Cell<Instant>,
+    /// When the outstanding liveness ping was sent, if one is unanswered.
+    probe: Cell<Option<Instant>>,
+    /// The liveness ping payload counter.
+    probe_seq: Cell<u64>,
 }
 
 impl RelayConn {
@@ -72,6 +94,9 @@ impl RelayConn {
         Ok(Self {
             ws,
             pending: RefCell::new(VecDeque::new()),
+            last_inbound: Cell::new(Instant::now()),
+            probe: Cell::new(None),
+            probe_seq: Cell::new(0),
         })
     }
 
@@ -91,7 +116,14 @@ impl RelayConn {
             if let Some(datagram) = self.pending.borrow_mut().pop_front() {
                 return Ok(datagram);
             }
-            let frame = match self.ws.receive().await {
+            let received = self.ws.receive().await;
+            // Any frame at all proves the relay's software end is alive,
+            // whatever it carries; the liveness stamp precedes the tag
+            // dispatch and the frames the happy path ignores.
+            if received.is_ok() {
+                self.last_inbound.set(Instant::now());
+            }
+            let frame = match received {
                 Ok(WsMessage::Binary(frame)) => frame,
                 Ok(WsMessage::String(_)) => continue,
                 Err(e) => return Err(format!("relay: {e:?}")),
@@ -122,6 +154,45 @@ impl RelayConn {
         }
     }
 
+    /// One liveness-probe step at `now`: answered probes are cleared (any
+    /// frame since the ping counts — upstream matches pong payloads, but for
+    /// detecting a silent wire any inbound frame has the same power, so this
+    /// client deliberately does not correlate pongs or track RTT), a wire
+    /// quiet for `quiet_after` gets a ping, and a probe unanswered for
+    /// `deadline` declares the wire dead.
+    pub fn probe_step(&self, now: Instant, quiet_after: Duration, deadline: Duration) -> ProbeStep {
+        if let Some(sent) = self.probe.get() {
+            if self.last_inbound.get() > sent {
+                self.probe.set(None);
+            } else if now.saturating_duration_since(sent) >= deadline {
+                // The probe stays set: a dead verdict is terminal for
+                // this connection, and repeating it is harmless.
+                return ProbeStep::Dead;
+            } else {
+                return ProbeStep::Healthy;
+            }
+        }
+        if now.saturating_duration_since(self.last_inbound.get()) >= quiet_after {
+            self.probe.set(Some(now));
+            let seq = self.probe_seq.get().wrapping_add(1);
+            self.probe_seq.set(seq);
+            ProbeStep::Ping(seq.to_le_bytes())
+        } else {
+            ProbeStep::Healthy
+        }
+    }
+
+    /// Send one liveness ping carrying `payload`. The payload is a
+    /// counter, not a correlator: [`RelayConn::probe_step`] accepts any
+    /// inbound frame as the answer.
+    pub async fn send_ping(&self, payload: &[u8; 8]) -> Result<(), String> {
+        let frame = frames::encode_ping(payload);
+        self.ws
+            .send(WsMessage::Binary(frame))
+            .await
+            .map_err(|e| format!("relay ping: {e:?}"))
+    }
+
     /// Initiate the connection's close (idempotent); a pending
     /// `recv_datagram` then resolves with its closed error.
     pub fn close(&self) {

core/src/relay_frames.rs

@@ -102,6 +102,15 @@ pub fn encode_pong(payload: &[u8; 8]) -> Vec<u8> {
     frame
 }
 
+/// Encode a `ping` frame carrying `payload`; the relay answers with a
+/// `pong` echoing it.
+pub fn encode_ping(payload: &[u8; 8]) -> Vec<u8> {
+    let mut frame = Vec::with_capacity(1 + 8);
+    frame.push(tag::PING as u8);
+    frame.extend_from_slice(payload);
+    frame
+}
+
 /// Decode a `relay-to-client-datagram` or `-batch` payload (the bytes
 /// after the frame type) into individual datagrams.
 pub fn decode_relay_datagrams(payload: &[u8], batch: bool) -> Option<Vec<Datagram>> {
@@ -213,7 +222,9 @@ mod tests {
             vec![0x0a, 42, 42, 42, 42, 42, 42, 42, 42]
         );
         let ping = [0x09, 42, 42, 42, 42, 42, 42, 42, 42];
-        let (tag, payload) = split_tag(&ping).unwrap();
+        let encoded = encode_ping(&[42; 8]);
+        assert_eq!(encoded, ping.to_vec());
+        let (tag, payload) = split_tag(&encoded).unwrap();
         assert_eq!(tag, tag::PING);
         assert_eq!(payload, [42; 8]);
     }

endpoint/src/endpoint_impl.rs

@@ -64,7 +64,7 @@
 use crate::udp::UdpWire;
 use crate::webrtc::{self, ChannelWire, SIGNAL_PREFIX};
 use crate::Component;
-use iroh_endpoint_core::relay::RelayConn;
+use iroh_endpoint_core::relay::{ProbeStep, RelayConn};
 use wit_bindgen::rt::async_support::{FutureReader, StreamReader};
 
 /// The pump's tick: noq's deadlines, the waiters' deadline re-check
@@ -245,6 +245,19 @@
 /// resolves its waiters before they give up.
 const DIAL_TIMEOUT: Duration = Duration::from_secs(10);
 
+/// How much inbound silence on a relay wire elicits a liveness ping
+/// (issue #96): matches upstream's ping cadence (iroh-1.0.3
+/// src/socket/transports/relay/actor.rs, PING_INTERVAL, reset on any
+/// inbound message), chosen there as half QUIC's default 30s
+/// max-idle-timeout so a dead home wire is caught with time to recover.
+const RELAY_PING_INTERVAL: Duration = Duration::from_secs(15);
+
+/// How long an unanswered liveness ping is allowed before the wire is
+/// declared dead: upstream's maximum pong bound (iroh-relay-1.0.3
+/// src/ping_tracker.rs, PING_TIMEOUT). Upstream shrinks the bound by
+/// measured RTT; this client tracks no RTT and uses the cap.
+const RELAY_PING_TIMEOUT: Duration = Duration::from_secs(5);
+
 struct ChannelEntry {
     wire: Rc<ChannelWire>,
     /// The relay-authenticated peer the channel was signaled with;
@@ -812,6 +825,12 @@
 /// redial; the pump ends only on close, never on a wire failure.
 /// `home_url` and `identity` are what a redial re-runs `RelayConn::connect`
 /// with.
+///
+/// The tick also probes relay liveness: a wire quiet for
+/// [`RELAY_PING_INTERVAL`] is pinged, and a ping unanswered for
+/// [`RELAY_PING_TIMEOUT`] retires its connection — the home relay into
+/// the redial, a foreign relay out of the pool — so a relay that goes
+/// silent without erroring the websocket is still detected.
 async fn pump(shared: Shared, udp: Option<Rc<UdpWire>>, home_url: String, identity: Rc<Identity>) {
     let mut udp_recv = pin!(udp_receive(udp.clone()).fuse());
     let mut tick = pin!(monotonic_clock::wait_for(TICK_NS).fuse());
@@ -1051,6 +1070,42 @@
                 tick.set(monotonic_clock::wait_for(TICK_NS).fuse());
                 shared.borrow_mut().handle_timeouts();
                 force_wake = true;
+                // Liveness sweep (issue #96). The pool is copied out of
+                // one short borrow, and the steps below take none across
+                // an await — the sends are detached, not awaited here.
+                let pool: Vec<(u32, Rc<RelayConn>)> = shared
+                    .borrow()
+                    .relay_pool
+                    .iter()
+                    .map(|(key, conn)| (*key, conn.clone()))
+                    .collect();
+                for (key, conn) in pool {
+                    match conn.probe_step(Instant::now(), RELAY_PING_INTERVAL, RELAY_PING_TIMEOUT) {
+                        ProbeStep::Ping(data) => {
+                            // Detached so a backpressured wire cannot pin
+                            // the pump: a ping that never leaves still
+                            // reaches the pong deadline, which is the
+                            // correct verdict for that wire either way.
+                            wit_bindgen::spawn_local(async move {
+                                let _ = conn.send_ping(&data).await;
+                            });
+                        }
+                        ProbeStep::Dead if key == HOME_RELAY => {
+                            if retire_home(&shared, &conn) {
+                                let delay = next_redial_delay(&mut redial_attempt, &mut home_since);
+                                redial = redial_home(home_url.clone(), identity.clone(), delay)
+                                    .boxed_local()
+                                    .fuse();
+                            }
+                        }
+                        ProbeStep::Dead => {
+                            if let Some(conn) = shared.borrow_mut().retire_relay(key) {
+                                conn.close();
+                            }
+                        }
+                        ProbeStep::Healthy => {}
+                    }
+                }
             }
         }
     }
@@ -2298,7 +2353,7 @@
             // ever will be, and the stream did not reach its FIN — a
             // connection close must not read as one (issue #13,
             // finding A2).
             Err(ReadError::Blocked) => match conn_failure {
                 Some(err) => Some(Err(err)),
                 None => None,
             },

host-polyengine/README.md

@@ -33,7 +33,8 @@ pinned module graph + the `node-datachannel` addon
 `src/run-endpoint.ts` — the endpoint lifecycle (bind + identity through
 idempotent teardown), the relay and WebRTC wires, the issue #10
 concurrency rows, and the liveness/recovery rows (idle survival, relay
-outage, stalling-relay dial deadlines). Each scenario names its
+outage, stalling-relay dial deadlines, mute-relay detection). Each
+scenario names its
 assertions where it lives; the exam's
 summary line is the inventory.