From dc7da2bb58f3866c628dcaa972104a7173700e55 Mon Sep 17 00:00:00 2001 From: Lann Martin Date: Sun, 23 Aug 2026 15:49:05 -0400 Subject: [PATCH] A silent relay wire is detected: quiet wires are pinged, unanswered pings retire the connection MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A relay that completed the handshake and then went silent-but-open was undetectable (#96): the pump retired and redialed only on a websocket error, and a stalled wire errors nothing — silent home-death for the home relay, black-holing pool entries for foreign ones. The client answered server pings but never initiated its own. The probe, upstream's mechanism (iroh-1.0.3 relay actor): - RelayConn stamps every inbound frame (last-inbound) and offers probe_step/send_ping — mechanism in core, policy in the pump. - The pump's tick sweeps the pool: a wire quiet for RELAY_PING_INTERVAL (15s, upstream PING_INTERVAL, reset-on-inbound semantics) is pinged; a ping unanswered for RELAY_PING_TIMEOUT (5s, upstream ping_tracker PING_TIMEOUT) declares the wire dead — the home relay retires into the redial, a foreign relay leaves the pool. - Divergences recorded at the definition sites: any inbound frame counts as the answer (upstream correlates pong payloads; equal power against a silent wire), and the pong bound is fixed (upstream shrinks it by measured RTT; we track none). - Ping sends are detached (spawn_local): a backpressured wire cannot pin the pump, and a ping that never leaves still reaches the pong deadline — the correct verdict for that wire either way. The gate is exam scenario 9: a mute-relay stub completes the relay handshake (immediate confirms-auth) and then never replies while staying OPEN — no socket error is ever available. The endpoint must ping it (observed on the stub), retire and redial the dead wire (a second accept on the still-open stub), and recover once the real relay returns. Measured: first ping at 15.2s, redial 4.9s after it. Falsified against the unfixed endpoint: scenario 9 fails with "the client's liveness probe reached the mute wire: 0 PING(s)". Gates: just check, just build, just exam-polyengine (10/10), just matrix (13/13). Fixes #96 --- core/src/relay.rs | 75 +++++++- core/src/relay_frames.rs | 13 +- endpoint/src/endpoint_impl.rs | 57 ++++++- host-polyengine/README.md | 3 +- host-polyengine/src/run-endpoint.ts | 255 +++++++++++++++++++++++++++- 5 files changed, 397 insertions(+), 6 deletions(-) diff --git a/core/src/relay.rs b/core/src/relay.rs index 239ec25..7fa2f68 100644 --- a/core/src/relay.rs +++ b/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>, + /// When this connection last saw any inbound websocket frame. + last_inbound: Cell, + /// When the outstanding liveness ping was sent, if one is unanswered. + probe: Cell>, + /// The liveness ping payload counter. + probe_seq: Cell, } 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) { diff --git a/core/src/relay_frames.rs b/core/src/relay_frames.rs index 3ddc221..016e66e 100644 --- a/core/src/relay_frames.rs +++ b/core/src/relay_frames.rs @@ -102,6 +102,15 @@ pub fn encode_pong(payload: &[u8; 8]) -> Vec { frame } +/// Encode a `ping` frame carrying `payload`; the relay answers with a +/// `pong` echoing it. +pub fn encode_ping(payload: &[u8; 8]) -> Vec { + 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> { @@ -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]); } diff --git a/endpoint/src/endpoint_impl.rs b/endpoint/src/endpoint_impl.rs index bc5d749..fc4cc5c 100644 --- a/endpoint/src/endpoint_impl.rs +++ b/endpoint/src/endpoint_impl.rs @@ -64,7 +64,7 @@ use crate::identity::IdentityRes; 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 @@ const REDIAL_ESTABLISHED: Duration = Duration::from_secs(10); /// 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, /// The relay-authenticated peer the channel was signaled with; @@ -812,6 +825,12 @@ fn on_event( /// 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>, home_url: String, identity: Rc) { 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 @@ async fn pump(shared: Shared, udp: Option>, home_url: String, identi 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)> = 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 => {} + } + } } } } diff --git a/host-polyengine/README.md b/host-polyengine/README.md index e8f64f8..a5c554a 100644 --- a/host-polyengine/README.md +++ b/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. diff --git a/host-polyengine/src/run-endpoint.ts b/host-polyengine/src/run-endpoint.ts index fb88869..85063a8 100644 --- a/host-polyengine/src/run-endpoint.ts +++ b/host-polyengine/src/run-endpoint.ts @@ -145,6 +145,25 @@ const DIAL_TIMEOUT_SLACK_HIGH_MS = 10_000; // out and the redial slot survived it. const STALL_REDIAL_WAIT_MS = 35_000; +// The guest's liveness-probe budgets for a silent-but-open relay wire +// (issue #96): after this much inbound silence a relay PING frame goes +// out (endpoint_impl.rs RELAY_PING_INTERVAL), and if nothing arrives +// within the timeout after that the wire is declared dead +// (endpoint_impl.rs RELAY_PING_TIMEOUT). +const RELAY_PING_INTERVAL_MS = 15_000; +const RELAY_PING_TIMEOUT_MS = 5_000; + +// How long the mute-relay probe waits for its first observed PING: the +// wire goes quiet at the handshake, so the bound is the probe interval +// plus generous slack for process/test scheduling. +const MUTE_PING_WAIT_MS = RELAY_PING_INTERVAL_MS + 10_000; + +// How long the mute-relay probe waits, after the first PING, for the +// redial to land back on the stub: the unanswered ping declares the +// wire dead at RELAY_PING_TIMEOUT, and the connection lived >= 10s so +// the redial is immediate. +const MUTE_REACCEPT_WAIT_MS = RELAY_PING_TIMEOUT_MS + 10_000; + /** * Bounded retries around the RefCell borrow hazard (see the header). The * budget is per-shape because the shapes lose the race at very different @@ -930,6 +949,184 @@ async function stallProbeOnce(control: RelayControl): Promise { } } +// --- the mute-relay probe (issue #96) ---------------------------------------- + +/** A relay stub that completes the handshake and then falls silent: it + * never replies to anything, including the client's liveness PING — + * the shape that made a silent-but-open wire indistinguishable from a + * healthy one before issue #96's fix. The stub SKIPS AUTHENTICATION + * entirely: it is a test fixture, never a relay implementation. */ +interface MuteRelay { + accepts(): number; + pings(): number; + close(): Promise; +} + +/** + * One relay frame is one BINARY websocket message, first byte a + * QUIC-varint frame tag (single byte for tags < 64) — mirrored from + * `core/src/relay_frames.rs`. SERVER_CONFIRMS_AUTH = 2, PING = 9. The + * client's connect loop accepts an immediate confirms-auth (no + * challenge required) and ignores its payload, so the stub's entire + * handshake is sending the single byte 0x02 after the upgrade. + */ +async function startMuteRelay(port: number): Promise { + let accepts = 0; + let pings = 0; + const sockets = new Set(); + + const server = Deno.serve( + { hostname: "127.0.0.1", port, onListen: () => {} }, + (req) => { + if (new URL(req.url).pathname !== "/relay") { + return new Response("not found", { status: 404 }); + } + // The client offers ["iroh-relay-v2", "iroh-relay-v1"] and its + // connect guard requires the server to select one of them. + const offered = req.headers.get("sec-websocket-protocol") ?? ""; + const protocol = offered.split(",")[0]?.trim(); + const { socket, response } = Deno.upgradeWebSocket(req, { protocol }); + socket.binaryType = "arraybuffer"; + socket.onopen = () => { + accepts++; + sockets.add(socket); + socket.send(new Uint8Array([2])); // SERVER_CONFIRMS_AUTH + }; + socket.onmessage = (ev) => { + // Never reply — that is the whole point of this stub. + const data = ev.data as ArrayBuffer; + const tag = new Uint8Array(data)[0]; + if (tag === 9) pings++; // PING + }; + socket.onclose = () => sockets.delete(socket); + return response; + }, + ); + + return { + accepts: () => accepts, + pings: () => pings, + close: async () => { + for (const s of sockets) { + try { + s.close(); + } catch { /* already closed */ } + } + try { + await server.shutdown(); + } catch { /* already shut down */ } + }, + }; +} + +interface MuteReport { + readonly wentDown: boolean; + readonly pings: number; + readonly pingMs: number; + readonly accepts: number; + readonly reacceptMs: number; + readonly echoed: string; +} + +/** + * One mute-relay probe (issue #96): a relay that completes the + * handshake and then goes silent must be detected by the client's own + * liveness PING, retired, and redialed — with no help from a socket + * error, since the stub's connection never closes on its own. Gathers + * data only — every assertion runs once, after the retry loop, against + * the returned report (scenarios 7/8's shape). + */ +async function muteProbeOnce(control: RelayControl): Promise { + const server = await newEndpointInstance({ label: "mute-server" }); + let sep: Endpoint | undefined; + let cep: Endpoint | undefined; + let stub: MuteRelay | undefined; + try { + sep = await deadline( + bindEndpoint(server, { alpns: [ALPN], relayUrl: control.url(), webrtc: false }), + 30_000, + "server bind", + ); + const serverId = await sep.id(); + + await control.stop(); + let wentDown = false; + for (let i = 0; i < 100; i++) { + if (!await portListening(RELAY_PORT)) { + wentDown = true; + break; + } + await settle(100); + } + + // The endpoint's home redial will connect to this stub and complete + // the mute handshake (accept #1). + stub = await startMuteRelay(RELAY_PORT); + const stubStarted = performance.now(); + + let pingMs = -1; + const pingDeadline = stubStarted + MUTE_PING_WAIT_MS; + while (performance.now() < pingDeadline) { + if (stub.pings() >= 1) { + pingMs = performance.now() - stubStarted; + break; + } + await settle(200); + } + + let reacceptMs = -1; + const reacceptStarted = performance.now(); + const reacceptDeadline = reacceptStarted + MUTE_REACCEPT_WAIT_MS; + while (performance.now() < reacceptDeadline) { + if (stub.accepts() >= 2) { + reacceptMs = performance.now() - reacceptStarted; + break; + } + await settle(200); + } + + const pings = stub.pings(); + const accepts = stub.accepts(); + + // Swap back: the endpoint's live stub connection errors when the + // stub closes (the existing error path), and the next redial + // reaches the real relay. + await stub.close(); + stub = undefined; + await control.start(); + + // Recovery: a fresh client instance, dialed against the now-real + // relay, proves the endpoint is still usable. + const client = await newEndpointInstance({ label: "mute-client" }); + cep = await deadline( + bindEndpoint(client, { alpns: [ALPN], relayUrl: control.url(), webrtc: false }), + 30_000, + "client bind", + ); + const addrs: TransportAddr[] = [{ kind: "relay", value: control.url() }]; + const conn = await deadline( + cep.connect({ endpointId: serverId, addrs }, ALPN), + 60_000, + "recovery connect", + ); + const sconn = await deadline(sep.accept(), 60_000, "recovery accept"); + const echoed = await echoRoundtrip(conn, sconn, "post-mute recovery echo"); + await conn.close(CLOSE_CODE, CLOSE_REASON); + await deadline(conn.waitClosed(), 30_000, "client wait-closed"); + + return { wentDown, pings, pingMs, accepts, reacceptMs, echoed }; + } catch (err) { + // Any throw must still leave the real relay running for later + // scenarios, mirroring scenarios 7/8's restore-on-failure care. + if (stub) await stub.close(); + if (!await portListening(RELAY_PORT)) await control.start(); + throw err; + } finally { + if (cep) await closeQuietly(cep, "client close after mute probe"); + if (sep) await closeQuietly(sep, "server close after mute probe"); + } +} + async function main(): Promise { installPanicWatchdog(); console.log("iroh endpoint exam (polyengine / stock Deno)"); @@ -1410,9 +1607,65 @@ async function main(): Promise { ); // -- 9 ------------------------------------------------------------------- + await scenario( + 9, + "mute relay: a silent-but-open home wire is detected, retired, and redialed (issue #96)", + async (v) => { + if (!relay.owned) { + v.status = "BLOCKED"; + v.detail = "the relay was pre-existing and adopted, so this run cannot stop it"; + return; + } + let r: MuteReport | undefined; + let lastError = ""; + // Two attempts only, as scenarios 7/8: each costs the full + // ping/timeout budget in wall time. + const attempts = 2; + for (let attempt = 1; attempt <= attempts && !r; attempt++) { + takeGuestPanics(); + try { + r = await muteProbeOnce(relayControl); + } catch (err) { + lastError = describeError(err); + console.log(` attempt ${attempt}/${attempts} failed: ${lastError}`); + if (!await portListening(RELAY_PORT)) await relayControl.start(); + await settle(100); + } + } + if (!r) throw new Error(`no attempt completed; last: ${lastError}`); + await settle(); + const panics = takeGuestPanics(); + + check(v, r.wentDown, `the relay stopped accepting on ${RELAY_PORT}`); + check( + v, + r.pings >= 1, + `the client's liveness probe reached the mute wire: ${r.pings} PING(s) ` + + `(first at ${r.pingMs.toFixed(0)} ms)`, + ); + check( + v, + r.accepts >= 2, + `the dead wire was retired and redialed while the stub stayed OPEN — ` + + `${r.accepts} accept(s), no socket error was available, the liveness ` + + `probe alone drove it (redial at ${r.reacceptMs.toFixed(0)} ms after the ping)`, + ); + check(v, r.echoed === MESSAGE.toUpperCase(), "an echo completed after recovery"); + check( + v, + panics.length === 0, + `no guest trap across the mute-relay detection (${panics.join("; ")})`, + ); + v.detail = `ping at ${r.pingMs.toFixed(0)} ms, redial at ${ + r.reacceptMs.toFixed(0) + } ms after it, recovery echo OK`; + }, + ); + + // -- 10 ------------------------------------------------------------------ // Last by necessity: this scenario stops the relay every later // scenario would need. - await scenario(9, "teardown: close + wait-closed, relay reaped", async (v) => { + await scenario(10, "teardown: close + wait-closed, relay reaped", async (v) => { const inst = await newEndpointInstance({ label: "teardown" }); const ep = await deadline( bindEndpoint(inst, { alpns: [ALPN], relayUrl: relay.url, webrtc: false }),