From 66382efce267e0ae9a02578db94351f17d7c9cd3 Mon Sep 17 00:00:00 2001 From: Andrew Chen Date: Fri, 4 Sep 2026 20:10:53 -0700 Subject: [PATCH 1/3] Replace the home-relay watchdog with in-place relay failover The watchdog's two steps were a nudge that did nothing on a stable host (`network_change()` only asks netmon to re-read interfaces, which reports "no change") and a rebuild that dropped every connection. Both are gone. `relay_failover::fail_over_home_relay` watches the home-relay status and, after 60 s without a connected home relay, removes the wedged relay from the endpoint's relay map. The relay-map change forces a full net report, which can only prefer a relay still in the map, so the endpoint homes on another configured relay while its identity, direct paths and established connections stay untouched. The removed relay is probed every 90 s and put back once it is connectable again, so a relay that answers HTTP but refuses relay connections cannot be re-selected and fail again in a loop. Custom relay sets must now hold at least two distinct relays, since one relay leaves nothing to fail over to. The startup probe fails only when no relay is reachable and warns per unreachable relay, so a client restarting during a relay outage still starts on the surviving relay. `rebuild_endpoint`, `EndpointFactory` and `RebuildableEndpoint` are removed with the rebuild step. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01HnozJMiCj3v5Xfdb9gB5S8 --- Cargo.lock | 2 +- Cargo.toml | 6 +- README.md | 12 +- src/endpoint.rs | 228 +-------------- src/lib.rs | 16 +- src/relay.rs | 104 +++++-- src/relay_failover.rs | 634 ++++++++++++++++++++++++++++++++++++++++++ src/relay_watchdog.rs | 366 ------------------------ 8 files changed, 736 insertions(+), 632 deletions(-) create mode 100644 src/relay_failover.rs delete mode 100644 src/relay_watchdog.rs diff --git a/Cargo.lock b/Cargo.lock index f1ee2e3..5c05f62 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -687,7 +687,7 @@ checksum = "3e0f1c7c3a72c66fd80abe965175f7523475c0489a87d3ff9d6e8c87d87a9d2d" [[package]] name = "flexaccess-iroh" -version = "0.0.3" +version = "0.0.7" dependencies = [ "anyhow", "base64", diff --git a/Cargo.toml b/Cargo.toml index 4f812e0..7e85634 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,8 +1,8 @@ [package] name = "flexaccess-iroh" -version = "0.0.3" +version = "0.0.7" edition = "2024" -description = "Shared iroh transport layer for FlexAccess applications: relay configuration and probing, endpoint building and rebuilding, the home-relay watchdog, and the endpoint-bound public-key auth transcript" +description = "Shared iroh transport layer for FlexAccess applications: relay configuration and probing, endpoint building, in-place home-relay failover, and the endpoint-bound public-key auth transcript" repository = "https://github.com/flexaccessdev/flexaccess-iroh" [features] @@ -34,7 +34,7 @@ tokio = { version = "1", features = ["macros", "rt", "sync", "time"] } iroh-mdns-address-lookup = { version = "0.5", optional = true } [dev-dependencies] -# Test double for iroh's `Watcher`-based status APIs (the relay watchdog tests +# Test double for iroh's `Watcher`-based status APIs (the relay failover tests # drive a plain `Watchable`); the same crate iroh itself re-exports `Watcher` from. n0-watcher = "1" tokio = { version = "1", features = ["full", "test-util"] } diff --git a/README.md b/README.md index bcad002..af7ec11 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,7 @@ Shared iroh transport layer for FlexAccess applications, as a Rust crate. The programs built on iroh in this org — [tunnel-rs], [ezvpn], [flextunnel] — share one transport foundation. Its design is documented once in [iroh-common-architecture]; this crate is that design as code, so a fix to the -relay watchdog or the relay probe lands here once instead of being ported by +relay failover or the relay probe lands here once instead of being ported by hand into every repo. [tunnel-rs]: https://github.com/flexaccessdev/tunnel-rs @@ -17,9 +17,9 @@ hand into every repo. | Module | Contents | |---|---| -| `relay` | `RelayConfig` (default vs custom relays, which also decides whether n0 internet discovery is on), the shared relay auth token, the strict per-relay startup probe | -| `endpoint` | the common endpoint builder, `create_endpoint` (strict first creation) vs `rebuild_endpoint` (tolerant mid-run replacement), `RebuildableEndpoint` | -| `relay_watchdog` | the server-side home-relay watchdog: nudge with `network_change()`, then ask for a rebuild | +| `relay` | `RelayConfig` (default vs custom relays, which also decides whether n0 internet discovery is on; custom relays must number at least two), the shared relay auth token, the per-relay startup probe (fails only when no relay is reachable) | +| `endpoint` | the common endpoint builder and `create_endpoint` | +| `relay_failover` | the server-side home-relay failover: after 60 s without a connected home relay, take the wedged relay out of the relay map so the forced net report homes the endpoint on another configured relay, in place; put it back once a probe shows it connectable | | `auth` | the endpoint-bound public-key auth transcript over the [flexaccess-keys] format; each application passes its own domain-separation context | Deliberately **not** in it: ALPNs, handshake wire formats, QUIC transport @@ -35,9 +35,9 @@ takes the resulting `iroh::SecretKey` / `flexaccess_keys` values. ```toml [dependencies] -flexaccess-iroh = { git = "https://github.com/flexaccessdev/flexaccess-iroh", tag = "v0.0.3" } +flexaccess-iroh = { git = "https://github.com/flexaccessdev/flexaccess-iroh", tag = "v0.0.7" } # or, with mDNS local-network discovery on every endpoint (compiled out on iOS): -flexaccess-iroh = { git = "...", tag = "v0.0.3", features = ["mdns"] } +flexaccess-iroh = { git = "...", tag = "v0.0.7", features = ["mdns"] } ``` The `flexaccess_keys` crate is re-exported so a consumer signs and verifies diff --git a/src/endpoint.rs b/src/endpoint.rs index 5692d5d..9abcacc 100644 --- a/src/endpoint.rs +++ b/src/endpoint.rs @@ -1,5 +1,5 @@ //! Endpoint construction shared by every FlexAccess program: the common -//! builder, the creation-vs-rebuild policy, and a rebuildable endpoint handle. +//! builder and the bind-and-come-online policy. //! //! Identity is the application's: it reads and decodes its own secret-key //! file (or generates an ephemeral key) and binds the resulting @@ -7,20 +7,17 @@ //! //! Applications layer their own ALPNs, hooks, identity, and QUIC transport //! tuning onto the [`iroh::endpoint::Builder`] returned by -//! [`endpoint_builder`], then hand it to [`create_endpoint`] (first creation: -//! strict) or [`rebuild_endpoint`] (mid-run replacement: tolerant). +//! [`endpoint_builder`], then hand it to [`create_endpoint`]. use crate::relay::{RELAY_CONNECT_TIMEOUT, RelayConfig, probe_custom_relays}; use anyhow::{Context, Result}; -use futures::future::BoxFuture; use iroh::{ - Endpoint, EndpointId, + Endpoint, address_lookup::{DnsAddressLookup, PkarrPublisher}, endpoint::{Builder as EndpointBuilder, QuicTransportConfig, presets}, }; use log::info; use std::sync::Arc; -use std::time::Duration; /// What an application decides about every endpoint it builds. #[derive(Debug, Clone)] @@ -96,9 +93,8 @@ pub fn endpoint_builder(relay_config: &RelayConfig, options: EndpointOptions) -> /// Wait for a freshly bound endpoint to come online (relay/discovery ready), /// bounded by [`RELAY_CONNECT_TIMEOUT`]. Does not close the endpoint on -/// failure; the caller decides (creation closes and fails, a rebuild carries -/// on). -pub async fn wait_online(endpoint: &Endpoint) -> Result<()> { +/// failure; the caller decides. +async fn wait_online(endpoint: &Endpoint) -> Result<()> { info!( "Waiting for endpoint to come online (timeout: {}s)...", RELAY_CONNECT_TIMEOUT.as_secs() @@ -112,10 +108,10 @@ pub async fn wait_online(endpoint: &Endpoint) -> Result<()> { } } -/// First creation of an endpoint: log the relay setup, probe every custom -/// relay (fail if any is unreachable — configuration validation), bind, and -/// require the endpoint to come online. On failure after binding the endpoint -/// is closed before the error propagates (dropping a bound endpoint without +/// Create an endpoint: log the relay setup, probe every custom relay (fail +/// only if none is reachable; see [`probe_custom_relays`]), bind, and require +/// the endpoint to come online. On failure after binding the endpoint is +/// closed before the error propagates (dropping a bound endpoint without /// `close()` is fatal under `panic = "abort"`). pub async fn create_endpoint(relay_config: &RelayConfig, builder: EndpointBuilder) -> Result { relay_config.log_status(); @@ -127,209 +123,3 @@ pub async fn create_endpoint(relay_config: &RelayConfig, builder: EndpointBuilde } Ok(endpoint) } - -/// Mid-run replacement of an endpoint, the recipe behind an -/// [`EndpointFactory`]. Differs from [`create_endpoint`] deliberately: -/// -/// - **No per-relay probe.** At creation the probe validates the configuration -/// (fail fast if *any* relay is down); during an outage that strictness -/// would block recovery through the one relay that still answers. -/// - **The online wait is tolerated failing.** A fresh endpoint is no worse -/// than the wedged one it replaces — LAN peers can still find it over mDNS — -/// and whatever tripped the rebuild (the relay watchdog, a client's -/// reconnect escalation) trips again if the relays stay unreachable. -pub async fn rebuild_endpoint(builder: EndpointBuilder) -> Result { - let endpoint = builder.bind().await.context("Failed to create iroh endpoint")?; - if let Err(e) = wait_online(&endpoint).await { - log::warn!("Rebuilt endpoint: {e:#}; continuing (local discovery may still work)"); - } - Ok(endpoint) -} - -/// Recipe producing a fresh, fully bound endpoint — how a -/// [`RebuildableEndpoint`] replaces itself mid-session, or how a server -/// replaces a wedged endpoint when the relay watchdog gives up on it. -pub type EndpointFactory = Arc BoxFuture<'static, Result> + Send + Sync>; - -/// Bound wait on the old endpoint's graceful close during a rebuild. The close -/// runs as its own task and is never cancelled (dropping a bound endpoint -/// without `close()` is fatal under panic=abort); the bound only keeps the -/// caller's reconnect loop from stalling behind it, letting a slow close -/// finish in the background. -const REBUILD_CLOSE_TIMEOUT: Duration = Duration::from_secs(5); - -/// An endpoint handle that can be **rebuilt** from scratch mid-session. -/// -/// `Endpoint::network_change()` re-binds dead UDP transports, but a wedged -/// endpoint can be broken beyond what a rebind repairs: a relay link lost to a -/// ping timeout that never re-establishes, stale cached paths for the peer, -/// dead discovery state. A process restart always recovers because it builds a -/// brand-new endpoint; [`Self::rebuild`] gives a reconnect loop that same -/// remedy in-process — fresh sockets, fresh relay connections, fresh discovery -/// — without dropping anything else the process holds (bound listeners, a -/// control socket). -/// -/// The handle is `Clone` and shared: a reconnect loop escalates to -/// [`Self::rebuild`] after repeated failures, while the embedder logs -/// [`Self::id`] and [`Self::close`]s whatever endpoint is current at teardown. -#[derive(Clone)] -pub struct RebuildableEndpoint { - /// The live endpoint, swapped by [`Self::rebuild`]. Std lock: accessors - /// clone the handle out synchronously and never hold it across an await. - current: Arc>, - factory: EndpointFactory, - /// Serializes [`Self::rebuild`]'s build-and-swap: the handle is shared, - /// and two callers noticing the same outage must not each build an - /// endpoint and have the second discard (and close) the first's good one. - rebuilding: Arc>, -} - -/// The installed endpoint plus how many rebuilds produced it, so a rebuild -/// caller can tell whether one already happened while it waited its turn. -struct Current { - generation: u64, - endpoint: Endpoint, -} - -impl RebuildableEndpoint { - /// Wrap a bound endpoint with the recipe that rebuilds it. - pub fn from_parts(endpoint: Endpoint, factory: EndpointFactory) -> Self { - Self { - current: Arc::new(std::sync::RwLock::new(Current { - generation: 0, - endpoint, - })), - factory, - rebuilding: Arc::new(tokio::sync::Mutex::new(())), - } - } - - /// A clone of the current endpoint handle. Take it fresh per use: a handle - /// held across a [`Self::rebuild`] keeps pointing at the old, closed - /// endpoint. - pub fn endpoint(&self) -> Endpoint { - self.current.read().expect("endpoint lock").endpoint.clone() - } - - /// The current endpoint id. Changes on rebuild for an ephemeral identity; - /// stable when the factory binds a fixed secret. - pub fn id(&self) -> EndpointId { - self.endpoint().id() - } - - fn generation(&self) -> u64 { - self.current.read().expect("endpoint lock").generation - } - - /// Swap in a freshly built endpoint and close the old one. On error the - /// current endpoint stays in place, so the caller can simply retry with it. - /// - /// Concurrent calls coalesce: a caller that arrives while another rebuild - /// is in flight waits for it and, if it installed a fresh endpoint, - /// returns `Ok` without building another — its trigger was the same dead - /// endpoint, and [`Self::endpoint`] now yields the replacement. Only if - /// the in-flight rebuild failed does the waiter build one itself. - pub async fn rebuild(&self) -> Result<()> { - let seen = self.generation(); - let old = { - let _serialized = self.rebuilding.lock().await; - if self.generation() != seen { - info!( - "Endpoint already rebuilt by a concurrent caller; endpoint id: {}", - self.id() - ); - return Ok(()); - } - let fresh = (self.factory)().await?; - let mut current = self.current.write().expect("endpoint lock"); - current.generation += 1; - std::mem::replace(&mut current.endpoint, fresh) - }; - // Graceful close on its own task: bounded wait here, but the task is - // never cancelled (see [`REBUILD_CLOSE_TIMEOUT`]). - let mut close = tokio::task::spawn(async move { old.close().await }); - if tokio::time::timeout(REBUILD_CLOSE_TIMEOUT, &mut close) - .await - .is_err() - { - log::warn!("Old endpoint's close is slow; leaving it to finish in the background"); - } - info!("Endpoint rebuilt; endpoint id: {}", self.id()); - Ok(()) - } - - /// Close the current endpoint gracefully (session teardown). - pub async fn close(&self) { - self.endpoint().close().await; - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[tokio::test] - async fn rebuildable_endpoint_swaps_and_closes_the_old_one() { - // Hermetic: loopback-only endpoints, no relays, no discovery. - fn loopback() -> EndpointBuilder { - Endpoint::builder(presets::Empty) - .relay_mode(iroh::RelayMode::Disabled) - .crypto_provider(Arc::new(rustls::crypto::ring::default_provider())) - } - let first = loopback().bind().await.unwrap(); - let first_id = first.id(); - let handle = RebuildableEndpoint::from_parts( - first.clone(), - Arc::new(|| Box::pin(async { loopback().bind().await.map_err(Into::into) })), - ); - assert_eq!(handle.id(), first_id); - - handle.rebuild().await.unwrap(); - assert_ne!(handle.id(), first_id, "an ephemeral rebuild gets a new id"); - assert!(first.is_closed(), "the replaced endpoint is closed"); - handle.close().await; - assert!(handle.endpoint().is_closed()); - } - - #[tokio::test] - async fn concurrent_rebuilds_coalesce_into_one() { - use std::sync::atomic::{AtomicUsize, Ordering}; - - fn loopback() -> EndpointBuilder { - Endpoint::builder(presets::Empty) - .relay_mode(iroh::RelayMode::Disabled) - .crypto_provider(Arc::new(rustls::crypto::ring::default_provider())) - } - let builds = Arc::new(AtomicUsize::new(0)); - let first = loopback().bind().await.unwrap(); - let handle = RebuildableEndpoint::from_parts(first.clone(), { - let builds = builds.clone(); - Arc::new(move || { - builds.fetch_add(1, Ordering::SeqCst); - Box::pin(async { - // Hold the build long enough for the second caller to - // queue behind it. - tokio::time::sleep(std::time::Duration::from_millis(200)).await; - loopback().bind().await.map_err(Into::into) - }) - }) - }); - - let a = handle.clone(); - let b = handle.clone(); - let (ra, rb) = tokio::join!(a.rebuild(), async { - tokio::time::sleep(std::time::Duration::from_millis(50)).await; - b.rebuild().await - }); - ra.unwrap(); - rb.unwrap(); - assert_eq!(builds.load(Ordering::SeqCst), 1, "the second caller joined the first"); - assert!(first.is_closed()); - assert!(!handle.endpoint().is_closed(), "the one fresh endpoint is live"); - - // A rebuild after the coalesced one is a new outage: it builds again. - handle.rebuild().await.unwrap(); - assert_eq!(builds.load(Ordering::SeqCst), 2); - handle.close().await; - } -} diff --git a/src/lib.rs b/src/lib.rs index b07f96d..fe732d3 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -7,13 +7,13 @@ //! across repos: //! //! - [`relay`]: the default-vs-custom [`relay::RelayConfig`] (which also decides -//! whether n0 internet discovery is on), the shared relay auth token, and the -//! strict per-relay startup probe. -//! - [`endpoint`]: the common endpoint builder, the bind-and-come-online policy -//! for first creation versus a mid-run rebuild, and a -//! [`endpoint::RebuildableEndpoint`] handle. -//! - [`relay_watchdog`]: the server-side home-relay watchdog that nudges, then -//! asks for a rebuild, when iroh silently loses its home relay. +//! whether n0 internet discovery is on, and requires at least two custom +//! relays), the shared relay auth token, and the per-relay startup probe. +//! - [`endpoint`]: the common endpoint builder and the bind-and-come-online +//! policy. +//! - [`relay_failover`]: the server-side home-relay failover that moves the +//! endpoint onto another configured relay, in place, when its home relay is +//! lost and iroh does not recover it on its own. //! - [`auth`]: the endpoint-bound public-key authentication transcript over the //! shared [`flexaccess_keys`] format; each application supplies its own //! domain-separation context. @@ -32,6 +32,6 @@ pub mod auth; pub mod endpoint; pub mod relay; -pub mod relay_watchdog; +pub mod relay_failover; pub use flexaccess_keys; diff --git a/src/relay.rs b/src/relay.rs index d04d640..67832d4 100644 --- a/src/relay.rs +++ b/src/relay.rs @@ -8,7 +8,7 @@ use anyhow::{Context, Result}; use futures::future::join_all; use iroh::{Endpoint, RelayMap, RelayMode, RelayUrl, endpoint::presets}; -use log::info; +use log::{info, warn}; use std::fmt; use std::sync::Arc; use std::time::Duration; @@ -17,6 +17,11 @@ use std::time::Duration; /// online before that is treated as a relay connectivity failure. pub const RELAY_CONNECT_TIMEOUT: Duration = Duration::from_secs(10); +/// The fewest distinct custom relays a configuration may name. Relay failover +/// needs somewhere to fail over *to*; the default relay map is n0's and is +/// not subject to this. +pub const MIN_CUSTOM_RELAYS: usize = 2; + /// Relay configuration, resolved once from the raw config strings. /// /// This is the single source of the default-vs-custom distinction. It selects @@ -32,7 +37,11 @@ pub enum RelayConfig { /// iroh's default relay map, with n0 address lookup. #[default] Default, - /// Custom relay set (parsed, deduped, in configured order). Never empty. + /// Custom relay set (parsed, deduped, in configured order). Never fewer + /// than [`MIN_CUSTOM_RELAYS`] distinct relays: a server keeps working + /// through a relay outage only by moving onto another configured relay, + /// and with one relay there is nothing to move to (see + /// [`crate::relay_failover`]). /// /// The configured order is kept because it is meaningful to a relay-only /// dialer, which tries the relays one at a time: the first URL is the @@ -86,7 +95,10 @@ impl RelayConfig { /// /// Empty input selects the default relays. Parsing fails on the first /// malformed URL, so config typos surface at resolve time instead of at each - /// use site. + /// use site. Custom relays must number at least [`MIN_CUSTOM_RELAYS`] + /// after deduplication: one relay leaves a server nothing to fail over to + /// when it stops working, which is rejected up front rather than + /// discovered during an outage. /// /// The token is normalized (blank/whitespace-only becomes `None`) and is /// **strictly gated to custom relays**: a non-empty token with no custom @@ -114,6 +126,14 @@ impl RelayConfig { parsed.push(url); } } + if parsed.len() < MIN_CUSTOM_RELAYS { + anyhow::bail!( + "custom relays need at least {MIN_CUSTOM_RELAYS} distinct relay_urls (got {}): a \ + server rides out a relay outage by moving onto another configured relay, and \ + with one relay there is nothing to fail over to", + parsed.len() + ); + } Ok(Self::Custom { urls: parsed, auth_token, @@ -169,7 +189,6 @@ impl RelayConfig { }; match self.custom_urls().len() { 0 => {} - 1 => info!("Using custom relay server{auth}"), n => info!("Using {n} custom relay servers with failover{auth}"), } } @@ -208,7 +227,7 @@ fn probe_endpoint_builder( /// it to come online, bounded by [`RELAY_CONNECT_TIMEOUT`]. `Ok(())` means the /// relay connected (and accepted the auth token, if any); otherwise the error /// describes the failure. The probe endpoint is always closed before returning. -async fn probe_relay(relay_url: &RelayUrl, auth_token: Option<&str>) -> Result<()> { +pub(crate) async fn probe_relay(relay_url: &RelayUrl, auth_token: Option<&str>) -> Result<()> { let endpoint = probe_endpoint_builder(relay_url, auth_token) .bind() .await @@ -223,24 +242,26 @@ async fn probe_relay(relay_url: &RelayUrl, auth_token: Option<&str>) -> Result<( }) } -/// Probe every configured custom relay individually (in parallel) and fail if -/// **any** relay is unreachable. +/// Probe every configured custom relay individually (in parallel). Startup +/// fails only if **every** relay is unreachable; each relay that does not +/// come online is reported as a warning, since the endpoint starts with that +/// much less failover headroom. /// -/// This is stricter than a single endpoint-wide `online()` wait, which only -/// proves that *one* relay in the set (the home relay) connected and so -/// reports a misleading all-clear when a backup relay is down. Default relays +/// Probing each relay on its own is what makes the warnings possible: a +/// single endpoint-wide `online()` wait proves only that *one* relay (the +/// home relay) connected and says nothing about the others. Default relays /// are not probed (returns `Ok(())` immediately). /// -/// Run this at first creation only: it validates the configuration (fail fast -/// if *any* relay is down). During an outage that strictness would block a -/// rebuild's recovery through the one relay that still answers, so -/// [`crate::endpoint::rebuild_endpoint`] deliberately skips it. +/// A relay that is down at startup must not stop the process: with at least +/// [`MIN_CUSTOM_RELAYS`] relays configured, the remaining ones carry it, and +/// refusing to start would turn a survivable relay outage into an outage of +/// every client that restarts during it. pub async fn probe_custom_relays(relay_config: &RelayConfig) -> Result<()> { let RelayConfig::Custom { urls, auth_token } = relay_config else { return Ok(()); }; let token = auth_token.as_deref(); - info!("Probing {} custom relay(s) for reachability...", urls.len()); + info!("Probing {} custom relays for reachability...", urls.len()); let results = join_all( urls.iter() .map(|url| async move { (url, probe_relay(url, token).await) }), @@ -250,9 +271,17 @@ pub async fn probe_custom_relays(relay_config: &RelayConfig) -> Result<()> { .into_iter() .filter_map(|(url, res)| res.err().map(|e| format!("{url}: {e}"))) .collect(); - if !failures.is_empty() { + if failures.len() == urls.len() { anyhow::bail!( - "{} of {} custom relay(s) failed to come online:\n {}", + "all {} custom relays failed to come online:\n {}", + urls.len(), + failures.join("\n ") + ); + } + if !failures.is_empty() { + warn!( + "{} of {} custom relays failed to come online; continuing with the rest, but a \ + further relay failure now has less to fail over to:\n {}", failures.len(), urls.len(), failures.join("\n ") @@ -266,6 +295,11 @@ mod tests { use super::*; const RELAY: &str = "https://relay.example.com./"; + const RELAY2: &str = "https://relay2.example.com./"; + + fn two() -> [String; 2] { + [RELAY.to_string(), RELAY2.to_string()] + } #[test] fn empty_urls_no_token_is_default() { @@ -304,11 +338,28 @@ mod tests { ); } + #[test] + fn a_single_custom_url_is_rejected() { + let err = RelayConfig::from_urls(&[RELAY.to_string()]) + .expect_err("one custom relay leaves nothing to fail over to"); + assert!( + err.to_string().contains("at least 2 distinct relay_urls (got 1)"), + "unexpected error: {err}" + ); + // Repeats of one URL are still one relay. + let err = RelayConfig::from_urls(&[RELAY.to_string(), RELAY.to_string()]) + .expect_err("a duplicated relay is still one relay"); + assert!( + err.to_string().contains("(got 1)"), + "unexpected error: {err}" + ); + } + #[test] fn custom_urls_without_token() { - let cfg = RelayConfig::from_urls_with_token(&[RELAY.to_string()], None).unwrap(); + let cfg = RelayConfig::from_urls_with_token(&two(), None).unwrap(); assert!(cfg.is_custom()); - assert_eq!(cfg.custom_urls().len(), 1); + assert_eq!(cfg.custom_urls().len(), 2); assert_eq!(cfg.relay_auth_token(), None); assert!(matches!(cfg.relay_mode(), RelayMode::Custom(_))); } @@ -329,9 +380,7 @@ mod tests { #[test] fn custom_urls_with_token_trimmed() { - let cfg = - RelayConfig::from_urls_with_token(&[RELAY.to_string()], Some(" secret\n".to_string())) - .unwrap(); + let cfg = RelayConfig::from_urls_with_token(&two(), Some(" secret\n".to_string())).unwrap(); assert!(cfg.is_custom()); assert_eq!(cfg.relay_auth_token(), Some("secret")); assert!(matches!(cfg.relay_mode(), RelayMode::Custom(_))); @@ -340,17 +389,14 @@ mod tests { #[test] fn token_is_trimmed_to_none_with_custom_urls() { // A blank token alongside custom relays is simply no token, not an error. - let cfg = RelayConfig::from_urls_with_token(&[RELAY.to_string()], Some(" ".to_string())) - .unwrap(); + let cfg = RelayConfig::from_urls_with_token(&two(), Some(" ".to_string())).unwrap(); assert!(cfg.is_custom()); assert_eq!(cfg.relay_auth_token(), None); } #[test] fn debug_output_redacts_auth_token() { - let cfg = - RelayConfig::from_urls_with_token(&[RELAY.to_string()], Some("secret".to_string())) - .unwrap(); + let cfg = RelayConfig::from_urls_with_token(&two(), Some("secret".to_string())).unwrap(); let dbg = format!("{cfg:?}"); assert!( !dbg.contains("secret"), @@ -359,14 +405,14 @@ mod tests { assert!(dbg.contains(""), "unexpected Debug output: {dbg}"); assert!(dbg.contains(RELAY), "urls missing from Debug output: {dbg}"); - let no_token = RelayConfig::from_urls(&[RELAY.to_string()]).unwrap(); + let no_token = RelayConfig::from_urls(&two()).unwrap(); assert!(format!("{no_token:?}").contains("auth_token: None")); assert_eq!(format!("{:?}", RelayConfig::Default), "Default"); } #[test] fn from_urls_carries_no_token() { - let cfg = RelayConfig::from_urls(&[RELAY.to_string()]).unwrap(); + let cfg = RelayConfig::from_urls(&two()).unwrap(); assert_eq!(cfg.relay_auth_token(), None); } diff --git a/src/relay_failover.rs b/src/relay_failover.rs new file mode 100644 index 0000000..438eed1 --- /dev/null +++ b/src/relay_failover.rs @@ -0,0 +1,634 @@ +//! Server-side home-relay failover. +//! +//! A server configured with custom relays is reachable to off-LAN clients +//! *only* through its home relay: with n0 discovery off, clients dial with +//! relay hints, and a relay forwards QUIC Initials only to endpoints currently +//! registered on it. iroh keeps that registration alive on its own and, when +//! the home relay stops answering its net-report probe, moves to another +//! configured relay within a re-probe cycle (20–26 s). What it does **not** +//! recover from on its own is a home relay that still answers probes while +//! the relay connection cannot be (re-)established: every net report keeps +//! preferring that relay, the relay actor keeps failing to connect to it, and +//! the server is dialable from nowhere. That is the shape of the incident +//! observed on iroh 1.0.3 (relays behind Cloudflare Tunnels that reset idle +//! WebSockets roughly hourly): no registration on any relay until the process +//! was restarted, while LAN clients that find it over mDNS kept working and +//! masked the outage. +//! +//! [`fail_over_home_relay`] observes [`Endpoint::home_relay_status`] and, +//! after [`RELAY_OUTAGE_FAILOVER`] without a connected home relay, takes the +//! wedged relay **out of the endpoint's relay map** +//! ([`Endpoint::remove_relay`]). A relay-map change forces a full net report, +//! the report can only prefer a relay that is still in the map, the endpoint +//! homes there, and clients dialing with the full relay list as hints reach +//! it through that relay. Nothing is torn down: the endpoint, its identity, +//! its direct paths, and every established connection stay as they are; a +//! connection whose only path ran through the dead relay times out on its own +//! and that client redials. This is why a custom relay set is required to hold +//! at least [`crate::relay::MIN_CUSTOM_RELAYS`] relays. +//! +//! The removed relay is put back once a probe (the same relay-only probe used +//! at startup) shows it connectable again, checked every +//! [`RELAY_RESTORE_INTERVAL`]. Putting it back unprobed would let a relay that +//! answers HTTP but refuses relay connections be re-selected — and fail +//! again — every few minutes. iroh drops a demoted relay's actor after 60 s +//! idle, so a relay that comes back is dialed by a fresh actor. +//! +//! When no home relay is selected at all (a net report that found none), +//! there is nothing to remove; the endpoint is nudged with a no-op relay-map +//! change, which forces a fresh report the same way. +//! +//! Only the *home* relay matters: non-home relays are connected on demand and +//! dropped after a minute idle, which is normal and not an outage. With the +//! default relays the future never resolves and never acts: there +//! reachability rests on n0 publishing and resolution, not on one relay +//! registration. + +use crate::relay::{RelayConfig, probe_relay}; +use iroh::endpoint::RelayStatus; +use iroh::{Endpoint, RelayMap, RelayUrl, Watcher}; +use std::time::Duration; +use tokio::time::Instant; + +/// How long the endpoint may go without a connected home relay before the +/// wedged relay is taken out of the relay map. Long enough to ride out a +/// routine relay reconnect (iroh's own reconnect backoff caps at 16 s) plus +/// the 20–26 s cadence of its periodic net report, which on its own re-homes +/// the endpoint when the relay is *really* down. +pub const RELAY_OUTAGE_FAILOVER: Duration = Duration::from_secs(60); + +/// How long an outage must last before it is reported. A home-relay change +/// publishes the new relay as connecting before it connects, so a healthy +/// move would otherwise log as a momentary outage. +const RELAY_OUTAGE_LOG_GRACE: Duration = Duration::from_secs(5); + +/// How often a relay that was taken out of the map is probed for +/// restoration. Longer than iroh's 60 s idle cleanup of a demoted relay +/// actor, so a relay that comes back is dialed by a fresh actor rather than +/// the one that was stuck on it. +pub const RELAY_RESTORE_INTERVAL: Duration = Duration::from_secs(90); + +/// Watch `endpoint`'s home relay and fail over in place when it is lost for +/// [`RELAY_OUTAGE_FAILOVER`]; see the module docs. Never resolves: run it +/// alongside the accept loop and drop it with the endpoint. Pending forever +/// with the default relays. +pub async fn fail_over_home_relay(endpoint: &Endpoint, relay_config: &RelayConfig) { + let RelayConfig::Custom { urls, auth_token } = relay_config else { + std::future::pending().await + }; + let mut relays = EndpointRelays { + endpoint, + configured: relay_config.relay_mode().relay_map(), + first: urls[0].clone(), + auth_token: auth_token.clone(), + }; + run_failover( + endpoint.home_relay_status(), + |statuses| describe_statuses(statuses), + &mut relays, + ) + .await +} + +/// What the home-relay status amounts to for failover purposes. +#[derive(Debug, Clone, PartialEq, Eq)] +enum HomeRelay { + /// Some home relay is connected. + Connected(RelayUrl), + /// No home relay is connected. `home` is the relay the endpoint is + /// trying to use (`None` when it has not selected one). + Down { + home: Option, + reason: String, + }, +} + +/// Classify a home-relay status vector for the failover loop. +fn describe_statuses(statuses: &[RelayStatus]) -> HomeRelay { + if let Some(connected) = statuses.iter().find(|s| s.is_connected()) { + return HomeRelay::Connected(connected.url().clone()); + } + let Some(home) = statuses.first() else { + return HomeRelay::Down { + home: None, + reason: "no home relay selected".into(), + }; + }; + let reason = statuses + .iter() + .map(|s| match s.last_error() { + Some(e) => format!("{} disconnected ({e:#})", s.url()), + None => format!("{} not connected", s.url()), + }) + .collect::>() + .join("; "); + HomeRelay::Down { + home: Some(home.url().clone()), + reason, + } +} + +/// The relay-map operations the failover loop performs, abstracted so the +/// loop can be driven by a test double. +trait FailoverRelays { + /// Take the wedged home relay out of the map so the next net report must + /// choose another; with no home selected, just force a fresh report. + /// Returns the URL that is now out of the map, if any. + async fn fail_over(&mut self, home: Option) -> Option; + + /// Put a removed relay back if it is connectable again. Returns whether + /// it was restored. + async fn restore(&mut self, url: &RelayUrl) -> bool; +} + +/// The real endpoint's relay map. +struct EndpointRelays<'a> { + endpoint: &'a Endpoint, + /// The configured relay map, kept as the source of each relay's + /// configuration (URL, auth token) for re-insertion. + configured: RelayMap, + /// The first configured relay: re-inserted unchanged to force a net + /// report when there is nothing to remove. + first: RelayUrl, + auth_token: Option, +} + +impl EndpointRelays<'_> { + async fn force_net_report(&self) { + if let Some(config) = self.configured.get(&self.first) { + log::warn!( + "Re-inserting {} into the relay map unchanged to force a fresh net report", + self.first + ); + self.endpoint.insert_relay(self.first.clone(), config).await; + } + } +} + +impl FailoverRelays for EndpointRelays<'_> { + async fn fail_over(&mut self, home: Option) -> Option { + let Some(url) = home else { + self.force_net_report().await; + return None; + }; + if self.endpoint.remove_relay(&url).await.is_none() { + log::warn!("{url} is not in the relay map; forcing a fresh net report instead"); + self.force_net_report().await; + return None; + } + log::warn!( + "Removed {url} from the relay map so the next net report homes this endpoint on \ + another configured relay" + ); + Some(url) + } + + async fn restore(&mut self, url: &RelayUrl) -> bool { + match probe_relay(url, self.auth_token.as_deref()).await { + Ok(()) => { + if let Some(config) = self.configured.get(url) { + self.endpoint.insert_relay(url.clone(), config).await; + } + true + } + Err(e) => { + log::warn!( + "{url} is still not connectable ({e:#}); leaving it out of the relay map, \ + next check in {}s", + RELAY_RESTORE_INTERVAL.as_secs() + ); + false + } + } + } +} + +/// Outage bookkeeping shared by the loop and its status observer. +#[derive(Default)] +struct Outage { + /// When the current outage began; `None` while a home relay is connected. + since: Option, + /// When to report the outage, unless it ends first; `None` once reported. + report_at: Option, + /// Why the home relay is down, for the report. + reason: String, + /// When to fail over if the outage lasts; `None` while connected or + /// while a failover is in progress. + fail_over_at: Option, + /// The last home relay seen connected, to log only changes. + last_connected: Option, +} + +impl Outage { + /// Log and record a status observation. + fn observe(&mut self, state: &HomeRelay) { + match state { + HomeRelay::Connected(url) => { + match self.since.take() { + // Reported outages end loudly; a blip that ended within + // the grace period is just a home change. + Some(since) if self.report_at.is_none() => log::info!( + "Home relay connection restored on {url} after {:.0}s", + since.elapsed().as_secs_f64() + ), + _ if self.last_connected.as_ref() != Some(url) => { + log::info!("Home relay: {url} connected") + } + _ => {} + } + self.report_at = None; + self.fail_over_at = None; + self.last_connected = Some(url.clone()); + } + HomeRelay::Down { reason, .. } => { + if self.since.is_none() { + let now = Instant::now(); + self.since = Some(now); + self.report_at = Some(now + RELAY_OUTAGE_LOG_GRACE); + self.fail_over_at = Some(now + RELAY_OUTAGE_FAILOVER); + } + self.reason = reason.clone(); + } + } + } + + /// Report the outage once it has outlasted the grace period. + fn report(&mut self) { + self.report_at = None; + log::warn!( + "No connected home relay ({}); off-LAN clients cannot reach this server until \ + it reconnects. Failing over to another relay in {}s unless it does", + self.reason, + RELAY_OUTAGE_FAILOVER.as_secs() + ); + } +} + +/// Wait for the next status value; pending forever once the watcher is +/// disconnected (the endpoint is gone). +async fn next_value(watcher: &mut W) -> W::Value { + match watcher.updated().await { + Ok(value) => value, + Err(_disconnected) => std::future::pending().await, + } +} + +/// The failover loop proper, generic over the status source and the relay +/// map so tests can drive it with doubles. +async fn run_failover(mut watcher: W, describe: D, relays: &mut R) +where + W: Watcher, + D: Fn(&W::Value) -> HomeRelay, + R: FailoverRelays, +{ + let mut outage = Outage::default(); + let mut value = watcher.get(); + loop { + let state = describe(&value); + outage.observe(&state); + + let Some(fail_over_at) = outage.fail_over_at else { + value = next_value(&mut watcher).await; + continue; + }; + if let Some(report_at) = outage.report_at { + tokio::select! { + () = tokio::time::sleep_until(report_at) => { + outage.report(); + continue; + } + next = next_value(&mut watcher) => { + value = next; + continue; + } + } + } + tokio::select! { + () = tokio::time::sleep_until(fail_over_at) => {} + next = next_value(&mut watcher) => { + value = next; + continue; + } + } + + let HomeRelay::Down { home, .. } = state else { + continue; + }; + outage.fail_over_at = None; + log::warn!( + "Still no connected home relay after {:.0}s; failing over", + outage + .since + .map(|since| since.elapsed().as_secs_f64()) + .unwrap_or_default() + ); + let removed = relays.fail_over(home).await; + + // Keep the relay out of the map until it is connectable again, still + // reporting status changes (the expected one being the home relay + // coming up elsewhere) while waiting between checks. + if let Some(url) = removed { + loop { + let check_at = Instant::now() + RELAY_RESTORE_INTERVAL; + loop { + let report_at = outage.report_at.unwrap_or(check_at); + tokio::select! { + () = tokio::time::sleep_until(check_at) => break, + () = tokio::time::sleep_until(report_at), if report_at < check_at => { + outage.report(); + } + next = next_value(&mut watcher) => { + value = next; + outage.observe(&describe(&value)); + } + } + } + if relays.restore(&url).await { + log::info!("{url} is connectable again and back in the relay map"); + break; + } + } + } + + // Still down after all that: the next failover is a full window away, + // not immediate. + value = watcher.get(); + outage.observe(&describe(&value)); + if outage.since.is_some() { + outage.fail_over_at = Some(Instant::now() + RELAY_OUTAGE_FAILOVER); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use n0_watcher::Watchable; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::{Arc, Mutex}; + + /// The home-relay status as a test double sees it. + #[derive(Debug, Clone, Copy, PartialEq, Eq)] + enum Status { + /// Connected on relay A. + UpOnA, + /// Connected on relay B. + UpOnB, + /// Home relay A selected but not connected. + DownOnA, + /// No home relay selected at all. + NoHome, + } + + fn relay_a() -> RelayUrl { + "https://a.example.com./".parse().unwrap() + } + + fn relay_b() -> RelayUrl { + "https://b.example.com./".parse().unwrap() + } + + fn describe(status: &Status) -> HomeRelay { + match status { + Status::UpOnA => HomeRelay::Connected(relay_a()), + Status::UpOnB => HomeRelay::Connected(relay_b()), + Status::DownOnA => HomeRelay::Down { + home: Some(relay_a()), + reason: "down".into(), + }, + Status::NoHome => HomeRelay::Down { + home: None, + reason: "no home relay selected".into(), + }, + } + } + + /// What the loop did, with the (paused-clock) time since the test began. + #[derive(Debug, Clone, PartialEq, Eq)] + enum Action { + FailOver { home: Option, at: Duration }, + Restore { url: RelayUrl, restored: bool, at: Duration }, + } + + #[derive(Clone)] + struct FakeRelays { + started: Instant, + actions: Arc>>, + /// Whether a restore probe succeeds. + connectable: Arc, + status: Watchable, + /// What the status flips to once the wedged relay is removed + /// (`None` = stays as it is). + after_fail_over: Option, + } + + impl FakeRelays { + fn new(status: &Watchable, after_fail_over: Option) -> Self { + Self { + started: Instant::now(), + actions: Arc::new(Mutex::new(Vec::new())), + connectable: Arc::new(AtomicBool::new(true)), + status: status.clone(), + after_fail_over, + } + } + + fn actions(&self) -> Vec { + self.actions.lock().unwrap().clone() + } + } + + impl FailoverRelays for FakeRelays { + async fn fail_over(&mut self, home: Option) -> Option { + self.actions.lock().unwrap().push(Action::FailOver { + home: home.clone(), + at: self.started.elapsed(), + }); + if let Some(status) = self.after_fail_over { + self.status.set(status).ok(); + } + home + } + + async fn restore(&mut self, url: &RelayUrl) -> bool { + let restored = self.connectable.load(Ordering::SeqCst); + self.actions.lock().unwrap().push(Action::Restore { + url: url.clone(), + restored, + at: self.started.elapsed(), + }); + restored + } + } + + /// Run the loop against `status` for `bound` of paused time. + async fn run_for(status: &Watchable, relays: &mut FakeRelays, bound: Duration) { + let _ = tokio::time::timeout(bound, run_failover(status.watch(), describe, relays)).await; + } + + const SEC: Duration = Duration::from_secs(1); + + #[tokio::test(start_paused = true)] + async fn healthy_relay_never_acts() { + let status = Watchable::new(Status::UpOnA); + let mut relays = FakeRelays::new(&status, None); + run_for(&status, &mut relays, RELAY_OUTAGE_FAILOVER * 5).await; + assert!(relays.actions().is_empty()); + } + + #[tokio::test(start_paused = true)] + async fn sustained_outage_removes_the_home_relay_and_restores_it_once_connectable() { + let status = Watchable::new(Status::DownOnA); + let mut relays = FakeRelays::new(&status, Some(Status::UpOnB)); + relays.connectable.store(false, Ordering::SeqCst); + let connectable = relays.connectable.clone(); + let flipper = async move { + // The relay comes back between the first and second restore checks. + tokio::time::sleep(RELAY_OUTAGE_FAILOVER + RELAY_RESTORE_INTERVAL + 5 * SEC).await; + connectable.store(true, Ordering::SeqCst); + }; + tokio::join!( + run_for( + &status, + &mut relays, + RELAY_OUTAGE_FAILOVER + RELAY_RESTORE_INTERVAL * 3 + ), + flipper + ); + assert_eq!( + relays.actions(), + vec![ + Action::FailOver { + home: Some(relay_a()), + at: RELAY_OUTAGE_FAILOVER, + }, + Action::Restore { + url: relay_a(), + restored: false, + at: RELAY_OUTAGE_FAILOVER + RELAY_RESTORE_INTERVAL, + }, + Action::Restore { + url: relay_a(), + restored: true, + at: RELAY_OUTAGE_FAILOVER + RELAY_RESTORE_INTERVAL * 2, + }, + ] + ); + } + + #[tokio::test(start_paused = true)] + async fn recovery_before_the_deadline_resets_the_clock() { + let status = Watchable::new(Status::UpOnA); + let mut relays = FakeRelays::new(&status, None); + let flipper = { + let status = status.clone(); + async move { + // Drop out for half the window, recover, drop out again for + // half a window: never long enough to act. + tokio::time::sleep(5 * SEC).await; + status.set(Status::DownOnA).ok(); + tokio::time::sleep(RELAY_OUTAGE_FAILOVER / 2).await; + status.set(Status::UpOnA).ok(); + tokio::time::sleep(RELAY_OUTAGE_FAILOVER / 2 + 5 * SEC).await; + status.set(Status::DownOnA).ok(); + tokio::time::sleep(RELAY_OUTAGE_FAILOVER / 2).await; + status.set(Status::UpOnA).ok(); + } + }; + tokio::join!( + run_for(&status, &mut relays, RELAY_OUTAGE_FAILOVER * 4), + flipper + ); + assert!(relays.actions().is_empty()); + } + + #[tokio::test(start_paused = true)] + async fn a_relay_that_stays_down_is_failed_over_again_a_full_window_after_restore() { + // Failover moves nothing (status stays down): the relay is restored + // when connectable, and the next failover comes a full window later. + let status = Watchable::new(Status::DownOnA); + let mut relays = FakeRelays::new(&status, None); + run_for( + &status, + &mut relays, + (RELAY_OUTAGE_FAILOVER + RELAY_RESTORE_INTERVAL) * 2 + SEC, + ) + .await; + let cycle = RELAY_OUTAGE_FAILOVER + RELAY_RESTORE_INTERVAL; + assert_eq!( + relays.actions(), + vec![ + Action::FailOver { + home: Some(relay_a()), + at: RELAY_OUTAGE_FAILOVER, + }, + Action::Restore { + url: relay_a(), + restored: true, + at: cycle, + }, + Action::FailOver { + home: Some(relay_a()), + at: cycle + RELAY_OUTAGE_FAILOVER, + }, + Action::Restore { + url: relay_a(), + restored: true, + at: cycle * 2, + }, + ] + ); + } + + #[tokio::test(start_paused = true)] + async fn no_home_relay_selected_only_forces_a_report() { + let status = Watchable::new(Status::NoHome); + let mut relays = FakeRelays::new(&status, Some(Status::UpOnB)); + run_for(&status, &mut relays, RELAY_OUTAGE_FAILOVER * 3).await; + assert_eq!( + relays.actions(), + vec![Action::FailOver { + home: None, + at: RELAY_OUTAGE_FAILOVER, + }], + "nothing was removed, so nothing is restored" + ); + } + + #[tokio::test(start_paused = true)] + async fn a_new_outage_after_recovery_gets_its_own_window() { + let status = Watchable::new(Status::DownOnA); + let mut relays = FakeRelays::new(&status, Some(Status::UpOnB)); + let flipper = { + let status = status.clone(); + async move { + // Failover at 60 s homes on B; restore of A at 150 s. Then + // B is lost at 200 s: a fresh window, failover at 260 s. + tokio::time::sleep(200 * SEC).await; + status.set(Status::DownOnA).ok(); + } + }; + tokio::join!( + run_for(&status, &mut relays, 300 * SEC), + flipper + ); + assert_eq!( + relays.actions(), + vec![ + Action::FailOver { + home: Some(relay_a()), + at: 60 * SEC, + }, + Action::Restore { + url: relay_a(), + restored: true, + at: 150 * SEC, + }, + Action::FailOver { + home: Some(relay_a()), + at: 260 * SEC, + }, + ] + ); + } +} diff --git a/src/relay_watchdog.rs b/src/relay_watchdog.rs deleted file mode 100644 index 20da4e5..0000000 --- a/src/relay_watchdog.rs +++ /dev/null @@ -1,366 +0,0 @@ -//! Server-side home-relay watchdog. -//! -//! A server configured with custom relays is reachable to off-LAN clients -//! *only* through its home relay: with n0 discovery off, clients dial with -//! relay hints, and a relay forwards QUIC Initials only to endpoints currently -//! registered on it. iroh keeps that registration alive on its own, but it has -//! been observed (v1.0.3, relays behind Cloudflare tunnels that reset idle -//! WebSockets roughly hourly) to silently lose its home relay for good after -//! one such reset: no dial retries, no warnings, no registration on any relay — -//! the server just stops being dialable until the process is restarted, while -//! LAN clients that find it over mDNS keep working and mask the outage. -//! -//! [`watch_home_relay`] observes [`Endpoint::home_relay_status`] and reacts in -//! two steps, mirroring the client's reconnect escalation: -//! -//! 1. after [`RELAY_OUTAGE_NUDGE`] without a connected home relay it calls -//! [`Endpoint::network_change`], which forces a fresh net report and relay -//! re-selection (enough when only the bookkeeping went stale); -//! 2. after the caller's rebuild deadline ([`RELAY_OUTAGE_REBUILD`] by -//! default) it resolves, telling the caller to replace the endpoint — the -//! in-process equivalent of the restart that is known to fix it. The caller -//! (the server's serve loop) closes the wedged endpoint, binds a fresh one -//! with the same identity (see [`crate::endpoint::rebuild_endpoint`]), and -//! serves on again. -//! -//! The resolution also says whether a home relay was connected at *any* point -//! of the watch ([`RelayOutage::relay_seen`]). A rebuilt endpoint that never -//! registers is a sign the relay itself is unreachable, not that iroh's -//! bookkeeping went stale; rebuilding it again drops every LAN client for -//! nothing, so the caller backs off between such rebuilds by passing a longer -//! deadline. -//! -//! Only the *home* relay matters: non-home relays are connected on demand and -//! dropped after a minute idle, which is normal and not an outage. - -use iroh::endpoint::RelayStatus; -use iroh::{Endpoint, Watcher}; -use std::future::Future; -use std::time::Duration; -use tokio::time::Instant; - -/// How long the endpoint may go without a connected home relay before the -/// watchdog nudges it with `network_change()`. Long enough to ride out a -/// routine relay reconnect (iroh's own reconnect backoff caps at 16s) plus the -/// ~25s cadence of its periodic net report. -pub const RELAY_OUTAGE_NUDGE: Duration = Duration::from_secs(60); - -/// Default for how long from the start of the outage before the watchdog -/// gives up on the endpoint and asks for a rebuild. Leaves the nudge two -/// minutes to take effect (a net report through slow relays can take tens of -/// seconds). -pub const RELAY_OUTAGE_REBUILD: Duration = Duration::from_secs(180); - -/// A tripped watchdog: the endpoint should be replaced. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct RelayOutage { - /// How long the endpoint has had no connected home relay. - pub duration: Duration, - /// Whether a home relay was connected at any point during the watch. - /// `false` means this endpoint never registered at all. - pub relay_seen: bool, -} - -/// Watch `endpoint`'s home-relay status and resolve once it has had no -/// connected home relay for `rebuild_after` (at least -/// [`RELAY_OUTAGE_NUDGE`]; [`RELAY_OUTAGE_REBUILD`] is the usual value), -/// having nudged it with `network_change()` at [`RELAY_OUTAGE_NUDGE`]. Never -/// resolves while the home relay stays connected; a reconnect at any point -/// resets the clock. Pending forever once the endpoint is gone. -pub async fn watch_home_relay(endpoint: &Endpoint, rebuild_after: Duration) -> RelayOutage { - watch_outage( - endpoint.home_relay_status(), - |statuses| describe_statuses(statuses), - || endpoint.network_change(), - rebuild_after, - ) - .await -} - -/// Describe a home-relay status vector for the watchdog: `Ok(())` when some -/// home relay is connected, otherwise `Err(reason)` naming what is wrong. -fn describe_statuses(statuses: &[RelayStatus]) -> Result<(), String> { - if statuses.iter().any(RelayStatus::is_connected) { - return Ok(()); - } - if statuses.is_empty() { - return Err("no home relay selected".into()); - } - let parts: Vec = statuses - .iter() - .map(|s| match s.last_error() { - Some(e) => format!("{} disconnected ({e:#})", s.url()), - None => format!("{} not connected", s.url()), - }) - .collect(); - Err(parts.join("; ")) -} - -/// The watchdog proper, generic over the status source so it can be driven by -/// a plain watchable in tests. `describe` classifies a status value -/// (`Ok` = connected); `nudge` is the first-stage remedy; `rebuild_after` is -/// the outage duration at which the watchdog trips. -async fn watch_outage( - mut watcher: W, - describe: D, - mut nudge: N, - rebuild_after: Duration, -) -> RelayOutage -where - W: Watcher, - D: Fn(&W::Value) -> Result<(), String>, - N: FnMut() -> Fut, - Fut: Future, -{ - let rebuild_after = rebuild_after.max(RELAY_OUTAGE_NUDGE); - let mut outage_since: Option = None; - let mut nudged = false; - let mut relay_seen = false; - let mut value = watcher.get(); - loop { - match describe(&value) { - Ok(()) => { - relay_seen = true; - if let Some(since) = outage_since.take() { - log::info!( - "Home relay connection restored after {:.0}s", - since.elapsed().as_secs_f64() - ); - } - nudged = false; - } - Err(reason) => { - if outage_since.is_none() { - outage_since = Some(Instant::now()); - log::warn!( - "No connected home relay ({reason}); off-LAN clients cannot reach this \ - server until it reconnects" - ); - } - } - } - - let Some(since) = outage_since else { - // Healthy: nothing to time, just wait for the next status change. - value = match watcher.updated().await { - Ok(value) => value, - Err(_disconnected) => std::future::pending().await, - }; - continue; - }; - - let deadline = since - + if nudged { - rebuild_after - } else { - RELAY_OUTAGE_NUDGE - }; - tokio::select! { - _ = tokio::time::sleep_until(deadline) => { - if nudged { - return RelayOutage { - duration: since.elapsed(), - relay_seen, - }; - } - nudged = true; - log::warn!( - "Still no connected home relay after {:.0}s; nudging the endpoint to \ - re-check its network and relays", - since.elapsed().as_secs_f64() - ); - nudge().await; - // The nudge may have already reconnected the relay; re-read - // rather than wait for a change notification we may have - // missed while it ran. - value = watcher.get(); - } - updated = watcher.updated() => { - value = match updated { - Ok(value) => value, - Err(_disconnected) => std::future::pending().await, - }; - } - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - use n0_watcher::Watchable; - use std::sync::Arc; - use std::sync::atomic::{AtomicUsize, Ordering}; - - /// Test double for the home-relay status: `true` = a home relay is - /// connected. - fn describe(connected: &bool) -> Result<(), String> { - if *connected { - Ok(()) - } else { - Err("down".into()) - } - } - - /// Run the watchdog on `status` with the default rebuild deadline, - /// counting nudges. Returns the watchdog future's resolution wrapped in a - /// bounded wait so a test never hangs. - async fn run_for( - status: &Watchable, - nudges: Arc, - bound: Duration, - ) -> Option { - run_with_deadline(status, nudges, bound, RELAY_OUTAGE_REBUILD).await - } - - async fn run_with_deadline( - status: &Watchable, - nudges: Arc, - bound: Duration, - rebuild_after: Duration, - ) -> Option { - let watchdog = watch_outage( - status.watch(), - describe, - || { - let nudges = nudges.clone(); - async move { - nudges.fetch_add(1, Ordering::SeqCst); - } - }, - rebuild_after, - ); - tokio::time::timeout(bound, watchdog).await.ok() - } - - #[tokio::test(start_paused = true)] - async fn healthy_relay_never_trips() { - let status = Watchable::new(true); - let nudges = Arc::new(AtomicUsize::new(0)); - let tripped = run_for(&status, nudges.clone(), RELAY_OUTAGE_REBUILD * 3).await; - assert!( - tripped.is_none(), - "healthy relay must never request a rebuild" - ); - assert_eq!(nudges.load(Ordering::SeqCst), 0); - } - - #[tokio::test(start_paused = true)] - async fn sustained_outage_nudges_then_requests_rebuild() { - let status = Watchable::new(false); - let nudges = Arc::new(AtomicUsize::new(0)); - let outage = run_for(&status, nudges.clone(), RELAY_OUTAGE_REBUILD * 2) - .await - .expect("a sustained outage must request a rebuild"); - assert_eq!( - nudges.load(Ordering::SeqCst), - 1, - "exactly one nudge before the rebuild" - ); - assert!(outage.duration >= RELAY_OUTAGE_REBUILD); - assert!(outage.duration < RELAY_OUTAGE_REBUILD + Duration::from_secs(1)); - assert!( - !outage.relay_seen, - "a relay that was never connected must be reported as never seen" - ); - } - - #[tokio::test(start_paused = true)] - async fn a_longer_rebuild_deadline_delays_the_trip_but_not_the_nudge() { - let status = Watchable::new(false); - let nudges = Arc::new(AtomicUsize::new(0)); - let rebuild_after = RELAY_OUTAGE_REBUILD * 4; - let nudge_count = nudges.clone(); - let (outage, ()) = tokio::join!( - run_with_deadline(&status, nudges.clone(), rebuild_after * 2, rebuild_after), - async move { - // The nudge still comes at the fixed first-stage deadline. - tokio::time::sleep(RELAY_OUTAGE_NUDGE + Duration::from_secs(1)).await; - assert_eq!(nudge_count.load(Ordering::SeqCst), 1); - } - ); - let outage = outage.expect("a sustained outage must request a rebuild"); - assert!(outage.duration >= rebuild_after); - assert!(outage.duration < rebuild_after + Duration::from_secs(1)); - assert_eq!(nudges.load(Ordering::SeqCst), 1); - } - - #[tokio::test(start_paused = true)] - async fn recovery_before_the_nudge_resets_the_clock() { - let status = Watchable::new(true); - let nudges = Arc::new(AtomicUsize::new(0)); - let flipper = { - let status = status.clone(); - async move { - // Drop out for half the nudge window, then recover; the - // watchdog must neither nudge nor trip. - tokio::time::sleep(Duration::from_secs(5)).await; - status.set(false).ok(); - tokio::time::sleep(RELAY_OUTAGE_NUDGE / 2).await; - status.set(true).ok(); - } - }; - let (tripped, ()) = tokio::join!( - run_for(&status, nudges.clone(), RELAY_OUTAGE_REBUILD * 2), - flipper - ); - assert!(tripped.is_none()); - assert_eq!(nudges.load(Ordering::SeqCst), 0); - } - - #[tokio::test(start_paused = true)] - async fn recovery_after_the_nudge_avoids_the_rebuild() { - let status = Watchable::new(false); - let nudges = Arc::new(AtomicUsize::new(0)); - let flipper = { - let status = status.clone(); - async move { - // Recover between the nudge and the rebuild deadline. - tokio::time::sleep(RELAY_OUTAGE_NUDGE + Duration::from_secs(10)).await; - status.set(true).ok(); - } - }; - let (tripped, ()) = tokio::join!( - run_for(&status, nudges.clone(), RELAY_OUTAGE_REBUILD * 2), - flipper - ); - assert!( - tripped.is_none(), - "a relay that came back must not be rebuilt" - ); - assert_eq!(nudges.load(Ordering::SeqCst), 1); - } - - #[tokio::test(start_paused = true)] - async fn a_second_outage_starts_a_fresh_clock() { - let status = Watchable::new(false); - let nudges = Arc::new(AtomicUsize::new(0)); - let flipper = { - let status = status.clone(); - async move { - // First outage: nudged, then recovers. Second outage: must - // get its own nudge and only trip a full window later. - tokio::time::sleep(RELAY_OUTAGE_NUDGE + Duration::from_secs(10)).await; - status.set(true).ok(); - tokio::time::sleep(Duration::from_secs(10)).await; - status.set(false).ok(); - } - }; - let start = Instant::now(); - let (tripped, ()) = tokio::join!( - run_for(&status, nudges.clone(), RELAY_OUTAGE_REBUILD * 3), - flipper - ); - let outage = tripped.expect("second outage must eventually trip"); - assert_eq!(nudges.load(Ordering::SeqCst), 2); - assert!(outage.duration >= RELAY_OUTAGE_REBUILD); - assert!(outage.duration < RELAY_OUTAGE_REBUILD + Duration::from_secs(1)); - assert!( - outage.relay_seen, - "the relay was connected between the outages, so it was seen" - ); - // Second outage began at nudge + 20s; the trip comes a full window after that. - let total = start.elapsed(); - assert!(total >= RELAY_OUTAGE_NUDGE + Duration::from_secs(20) + RELAY_OUTAGE_REBUILD); - } -} From 44d212b12076b9fa0439d7de176b25893302cfa3 Mon Sep 17 00:00:00 2001 From: Andrew Chen Date: Fri, 4 Sep 2026 20:37:31 -0700 Subject: [PATCH 2/3] Document that the two-relay minimum counts distinct relays The custom relay count is taken after deduplication, so say "at least two distinct relays" everywhere the minimum is described; listing the same relay twice is not a valid two-relay configuration. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01HnozJMiCj3v5Xfdb9gB5S8 --- README.md | 2 +- src/lib.rs | 5 +++-- src/relay.rs | 9 +++++---- src/relay_failover.rs | 2 +- 4 files changed, 10 insertions(+), 8 deletions(-) diff --git a/README.md b/README.md index af7ec11..e8ff8b5 100644 --- a/README.md +++ b/README.md @@ -17,7 +17,7 @@ hand into every repo. | Module | Contents | |---|---| -| `relay` | `RelayConfig` (default vs custom relays, which also decides whether n0 internet discovery is on; custom relays must number at least two), the shared relay auth token, the per-relay startup probe (fails only when no relay is reachable) | +| `relay` | `RelayConfig` (default vs custom relays, which also decides whether n0 internet discovery is on; custom relays must number at least two distinct URLs; duplicates are collapsed before the count), the shared relay auth token, the per-relay startup probe (fails only when no relay is reachable) | | `endpoint` | the common endpoint builder and `create_endpoint` | | `relay_failover` | the server-side home-relay failover: after 60 s without a connected home relay, take the wedged relay out of the relay map so the forced net report homes the endpoint on another configured relay, in place; put it back once a probe shows it connectable | | `auth` | the endpoint-bound public-key auth transcript over the [flexaccess-keys] format; each application passes its own domain-separation context | diff --git a/src/lib.rs b/src/lib.rs index fe732d3..f55d0e3 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -7,8 +7,9 @@ //! across repos: //! //! - [`relay`]: the default-vs-custom [`relay::RelayConfig`] (which also decides -//! whether n0 internet discovery is on, and requires at least two custom -//! relays), the shared relay auth token, and the per-relay startup probe. +//! whether n0 internet discovery is on, and requires at least two distinct +//! custom relays, duplicates collapsed), the shared relay auth token, and the +//! per-relay startup probe. //! - [`endpoint`]: the common endpoint builder and the bind-and-come-online //! policy. //! - [`relay_failover`]: the server-side home-relay failover that moves the diff --git a/src/relay.rs b/src/relay.rs index 67832d4..b884800 100644 --- a/src/relay.rs +++ b/src/relay.rs @@ -96,9 +96,10 @@ impl RelayConfig { /// Empty input selects the default relays. Parsing fails on the first /// malformed URL, so config typos surface at resolve time instead of at each /// use site. Custom relays must number at least [`MIN_CUSTOM_RELAYS`] - /// after deduplication: one relay leaves a server nothing to fail over to - /// when it stops working, which is rejected up front rather than - /// discovered during an outage. + /// distinct URLs after deduplication (listing the same relay twice counts + /// once): one relay leaves a server nothing to fail over to when it stops + /// working, which is rejected up front rather than discovered during an + /// outage. /// /// The token is normalized (blank/whitespace-only becomes `None`) and is /// **strictly gated to custom relays**: a non-empty token with no custom @@ -253,7 +254,7 @@ pub(crate) async fn probe_relay(relay_url: &RelayUrl, auth_token: Option<&str>) /// are not probed (returns `Ok(())` immediately). /// /// A relay that is down at startup must not stop the process: with at least -/// [`MIN_CUSTOM_RELAYS`] relays configured, the remaining ones carry it, and +/// [`MIN_CUSTOM_RELAYS`] distinct relays configured, the remaining ones carry it, and /// refusing to start would turn a survivable relay outage into an outage of /// every client that restarts during it. pub async fn probe_custom_relays(relay_config: &RelayConfig) -> Result<()> { diff --git a/src/relay_failover.rs b/src/relay_failover.rs index 438eed1..d65d781 100644 --- a/src/relay_failover.rs +++ b/src/relay_failover.rs @@ -25,7 +25,7 @@ //! its direct paths, and every established connection stay as they are; a //! connection whose only path ran through the dead relay times out on its own //! and that client redials. This is why a custom relay set is required to hold -//! at least [`crate::relay::MIN_CUSTOM_RELAYS`] relays. +//! at least [`crate::relay::MIN_CUSTOM_RELAYS`] distinct relays. //! //! The removed relay is put back once a probe (the same relay-only probe used //! at startup) shows it connectable again, checked every From de88179b5923497f6c629a4c02ed631c0a9aaa0b Mon Sep 17 00:00:00 2001 From: Andrew Chen Date: Fri, 4 Sep 2026 20:48:46 -0700 Subject: [PATCH 3/3] Let failover act on a second wedged relay while the first is still out Review fixes in relay_failover: - The restore wait was a nested loop: while a removed relay stayed unconnectable, a newly wedged home relay was reported but never failed over. The loop is now a single select over the report timer, the failover timer, the per-relay restore deadlines and the status watcher, tracking every removed relay with its own next probe time (each still spaced RELAY_RESTORE_INTERVAL apart so a returning relay gets a fresh actor). The last relay left in the map is never removed; the endpoint is nudged instead, and a nudge re-inserts only a relay that is still in the map, never a removed one. - restore() reports success only after insert_relay ran; a connectable relay missing from the configured map is logged and left out. - fail_over_home_relay no longer indexes urls[0]: an empty custom relay set (constructible from outside the crate) stays pending instead of panicking. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01HnozJMiCj3v5Xfdb9gB5S8 --- src/relay_failover.rs | 390 +++++++++++++++++++++++++++--------------- 1 file changed, 248 insertions(+), 142 deletions(-) diff --git a/src/relay_failover.rs b/src/relay_failover.rs index d65d781..3f8e129 100644 --- a/src/relay_failover.rs +++ b/src/relay_failover.rs @@ -34,9 +34,12 @@ //! again — every few minutes. iroh drops a demoted relay's actor after 60 s //! idle, so a relay that comes back is dialed by a fresh actor. //! -//! When no home relay is selected at all (a net report that found none), -//! there is nothing to remove; the endpoint is nudged with a no-op relay-map -//! change, which forces a fresh report the same way. +//! Several relays can be out of the map at once: if the relay the endpoint +//! moved onto wedges too while the first is still unconnectable, it is removed +//! as well. The last relay left in the map is never removed. When there is +//! nothing to remove (no home relay selected, or the home relay is the last +//! one), the endpoint is nudged with a no-op relay-map change instead, which +//! forces a fresh report the same way. //! //! Only the *home* relay matters: non-home relays are connected on demand and //! dropped after a minute idle, which is normal and not an outage. With the @@ -76,10 +79,14 @@ pub async fn fail_over_home_relay(endpoint: &Endpoint, relay_config: &RelayConfi let RelayConfig::Custom { urls, auth_token } = relay_config else { std::future::pending().await }; + if urls.is_empty() { + log::error!("Custom relay set is empty; home-relay failover has nothing to work with"); + std::future::pending().await + } let mut relays = EndpointRelays { endpoint, configured: relay_config.relay_mode().relay_map(), - first: urls[0].clone(), + urls: urls.clone(), auth_token: auth_token.clone(), }; run_failover( @@ -131,13 +138,19 @@ fn describe_statuses(statuses: &[RelayStatus]) -> HomeRelay { /// The relay-map operations the failover loop performs, abstracted so the /// loop can be driven by a test double. trait FailoverRelays { - /// Take the wedged home relay out of the map so the next net report must - /// choose another; with no home selected, just force a fresh report. - /// Returns the URL that is now out of the map, if any. - async fn fail_over(&mut self, home: Option) -> Option; + /// The configured relays, in configured order. + fn configured(&self) -> &[RelayUrl]; + + /// Take `url` out of the map so the next net report must choose another. + /// Returns whether it was in the map. + async fn remove(&mut self, url: &RelayUrl) -> bool; + + /// Re-insert `url`, which is still in the map, unchanged: a relay-map + /// change forces a fresh net report without changing the choice on offer. + async fn nudge(&mut self, url: &RelayUrl); /// Put a removed relay back if it is connectable again. Returns whether - /// it was restored. + /// it is back in the map. async fn restore(&mut self, url: &RelayUrl) -> bool; } @@ -147,48 +160,48 @@ struct EndpointRelays<'a> { /// The configured relay map, kept as the source of each relay's /// configuration (URL, auth token) for re-insertion. configured: RelayMap, - /// The first configured relay: re-inserted unchanged to force a net - /// report when there is nothing to remove. - first: RelayUrl, + /// The configured relay URLs, in configured order. + urls: Vec, auth_token: Option, } -impl EndpointRelays<'_> { - async fn force_net_report(&self) { - if let Some(config) = self.configured.get(&self.first) { - log::warn!( - "Re-inserting {} into the relay map unchanged to force a fresh net report", - self.first - ); - self.endpoint.insert_relay(self.first.clone(), config).await; - } +impl FailoverRelays for EndpointRelays<'_> { + fn configured(&self) -> &[RelayUrl] { + &self.urls } -} -impl FailoverRelays for EndpointRelays<'_> { - async fn fail_over(&mut self, home: Option) -> Option { - let Some(url) = home else { - self.force_net_report().await; - return None; - }; - if self.endpoint.remove_relay(&url).await.is_none() { - log::warn!("{url} is not in the relay map; forcing a fresh net report instead"); - self.force_net_report().await; - return None; + async fn remove(&mut self, url: &RelayUrl) -> bool { + if self.endpoint.remove_relay(url).await.is_none() { + log::warn!("{url} is not in the relay map"); + return false; } log::warn!( "Removed {url} from the relay map so the next net report homes this endpoint on \ another configured relay" ); - Some(url) + true + } + + async fn nudge(&mut self, url: &RelayUrl) { + let Some(config) = self.configured.get(url) else { + log::error!("{url} is not a configured relay; cannot re-insert it"); + return; + }; + log::warn!("Re-inserting {url} into the relay map unchanged to force a fresh net report"); + self.endpoint.insert_relay(url.clone(), config).await; } async fn restore(&mut self, url: &RelayUrl) -> bool { match probe_relay(url, self.auth_token.as_deref()).await { Ok(()) => { - if let Some(config) = self.configured.get(url) { - self.endpoint.insert_relay(url.clone(), config).await; - } + let Some(config) = self.configured.get(url) else { + log::error!( + "{url} is connectable but not a configured relay; cannot put it back in \ + the relay map" + ); + return false; + }; + self.endpoint.insert_relay(url.clone(), config).await; true } Err(e) => { @@ -273,6 +286,49 @@ async fn next_value(watcher: &mut W) -> W::Value { } } +/// Sleep until `at`; pending forever when there is no deadline. +async fn sleep_until_opt(at: Option) { + match at { + Some(at) => tokio::time::sleep_until(at).await, + None => std::future::pending().await, + } +} + +/// Act on a home relay that has been down for the whole window: remove it +/// from the map, or nudge the endpoint when there is nothing to remove. +/// Returns the URL that is now out of the map, if any. +async fn fail_over( + relays: &mut R, + home: Option, + removed: &[RelayUrl], +) -> Option { + let in_map: Vec = relays + .configured() + .iter() + .filter(|u| !removed.contains(u)) + .cloned() + .collect(); + let home = home.filter(|home| !removed.contains(home)); + match home { + // Never remove the last relay left in the map. + Some(home) if in_map.len() > 1 => { + if relays.remove(&home).await { + return Some(home); + } + if let Some(url) = in_map.first() { + relays.nudge(url).await; + } + } + Some(home) => relays.nudge(&home).await, + None => { + if let Some(url) = in_map.first() { + relays.nudge(url).await; + } + } + } + None +} + /// The failover loop proper, generic over the status source and the relay /// map so tests can drive it with doubles. async fn run_failover(mut watcher: W, describe: D, relays: &mut R) @@ -282,81 +338,53 @@ where R: FailoverRelays, { let mut outage = Outage::default(); + // Relays taken out of the map, each with when it is next probed for + // restoration. + let mut removed: Vec<(RelayUrl, Instant)> = Vec::new(); let mut value = watcher.get(); + outage.observe(&describe(&value)); loop { - let state = describe(&value); - outage.observe(&state); - - let Some(fail_over_at) = outage.fail_over_at else { - value = next_value(&mut watcher).await; - continue; - }; - if let Some(report_at) = outage.report_at { - tokio::select! { - () = tokio::time::sleep_until(report_at) => { - outage.report(); + let check_at = removed.iter().map(|(_, at)| *at).min(); + let out: Vec = removed.iter().map(|(url, _)| url.clone()).collect(); + tokio::select! { + () = sleep_until_opt(outage.report_at) => outage.report(), + () = sleep_until_opt(outage.fail_over_at) => { + outage.fail_over_at = None; + let HomeRelay::Down { home, .. } = describe(&value) else { continue; + }; + log::warn!( + "Still no connected home relay after {:.0}s; failing over", + outage + .since + .map(|since| since.elapsed().as_secs_f64()) + .unwrap_or_default() + ); + if let Some(url) = fail_over(relays, home, &out).await { + removed.push((url, Instant::now() + RELAY_RESTORE_INTERVAL)); } - next = next_value(&mut watcher) => { - value = next; - continue; + // Still down after that (the status watcher clears this if + // not): the next failover is a full window away, not + // immediate. + outage.fail_over_at = Some(Instant::now() + RELAY_OUTAGE_FAILOVER); + } + () = sleep_until_opt(check_at) => { + let now = Instant::now(); + for (url, at) in std::mem::take(&mut removed) { + if at > now { + removed.push((url, at)); + } else if relays.restore(&url).await { + log::info!("{url} is connectable again and back in the relay map"); + } else { + removed.push((url, now + RELAY_RESTORE_INTERVAL)); + } } } - } - tokio::select! { - () = tokio::time::sleep_until(fail_over_at) => {} next = next_value(&mut watcher) => { value = next; - continue; - } - } - - let HomeRelay::Down { home, .. } = state else { - continue; - }; - outage.fail_over_at = None; - log::warn!( - "Still no connected home relay after {:.0}s; failing over", - outage - .since - .map(|since| since.elapsed().as_secs_f64()) - .unwrap_or_default() - ); - let removed = relays.fail_over(home).await; - - // Keep the relay out of the map until it is connectable again, still - // reporting status changes (the expected one being the home relay - // coming up elsewhere) while waiting between checks. - if let Some(url) = removed { - loop { - let check_at = Instant::now() + RELAY_RESTORE_INTERVAL; - loop { - let report_at = outage.report_at.unwrap_or(check_at); - tokio::select! { - () = tokio::time::sleep_until(check_at) => break, - () = tokio::time::sleep_until(report_at), if report_at < check_at => { - outage.report(); - } - next = next_value(&mut watcher) => { - value = next; - outage.observe(&describe(&value)); - } - } - } - if relays.restore(&url).await { - log::info!("{url} is connectable again and back in the relay map"); - break; - } + outage.observe(&describe(&value)); } } - - // Still down after all that: the next failover is a full window away, - // not immediate. - value = watcher.get(); - outage.observe(&describe(&value)); - if outage.since.is_some() { - outage.fail_over_at = Some(Instant::now() + RELAY_OUTAGE_FAILOVER); - } } } @@ -376,6 +404,10 @@ mod tests { UpOnB, /// Home relay A selected but not connected. DownOnA, + /// Home relay B selected but not connected. + DownOnB, + /// Home relay C selected but not connected. + DownOnC, /// No home relay selected at all. NoHome, } @@ -388,14 +420,24 @@ mod tests { "https://b.example.com./".parse().unwrap() } + fn relay_c() -> RelayUrl { + "https://c.example.com./".parse().unwrap() + } + + fn down_on(url: RelayUrl) -> HomeRelay { + HomeRelay::Down { + home: Some(url), + reason: "down".into(), + } + } + fn describe(status: &Status) -> HomeRelay { match status { Status::UpOnA => HomeRelay::Connected(relay_a()), Status::UpOnB => HomeRelay::Connected(relay_b()), - Status::DownOnA => HomeRelay::Down { - home: Some(relay_a()), - reason: "down".into(), - }, + Status::DownOnA => down_on(relay_a()), + Status::DownOnB => down_on(relay_b()), + Status::DownOnC => down_on(relay_c()), Status::NoHome => HomeRelay::Down { home: None, reason: "no home relay selected".into(), @@ -406,19 +448,21 @@ mod tests { /// What the loop did, with the (paused-clock) time since the test began. #[derive(Debug, Clone, PartialEq, Eq)] enum Action { - FailOver { home: Option, at: Duration }, + Remove { url: RelayUrl, at: Duration }, + Nudge { url: RelayUrl, at: Duration }, Restore { url: RelayUrl, restored: bool, at: Duration }, } #[derive(Clone)] struct FakeRelays { started: Instant, + configured: Vec, actions: Arc>>, /// Whether a restore probe succeeds. connectable: Arc, status: Watchable, - /// What the status flips to once the wedged relay is removed - /// (`None` = stays as it is). + /// What the status flips to once the loop has acted on the relay + /// map (`None` = stays as it is). after_fail_over: Option, } @@ -426,6 +470,7 @@ mod tests { fn new(status: &Watchable, after_fail_over: Option) -> Self { Self { started: Instant::now(), + configured: vec![relay_a(), relay_b(), relay_c()], actions: Arc::new(Mutex::new(Vec::new())), connectable: Arc::new(AtomicBool::new(true)), status: status.clone(), @@ -436,18 +481,33 @@ mod tests { fn actions(&self) -> Vec { self.actions.lock().unwrap().clone() } + + fn acted(&self, action: Action) { + self.actions.lock().unwrap().push(action); + if let Some(status) = self.after_fail_over { + self.status.set(status).ok(); + } + } } impl FailoverRelays for FakeRelays { - async fn fail_over(&mut self, home: Option) -> Option { - self.actions.lock().unwrap().push(Action::FailOver { - home: home.clone(), + fn configured(&self) -> &[RelayUrl] { + &self.configured + } + + async fn remove(&mut self, url: &RelayUrl) -> bool { + self.acted(Action::Remove { + url: url.clone(), + at: self.started.elapsed(), + }); + true + } + + async fn nudge(&mut self, url: &RelayUrl) { + self.acted(Action::Nudge { + url: url.clone(), at: self.started.elapsed(), }); - if let Some(status) = self.after_fail_over { - self.status.set(status).ok(); - } - home } async fn restore(&mut self, url: &RelayUrl) -> bool { @@ -498,8 +558,8 @@ mod tests { assert_eq!( relays.actions(), vec![ - Action::FailOver { - home: Some(relay_a()), + Action::Remove { + url: relay_a(), at: RELAY_OUTAGE_FAILOVER, }, Action::Restore { @@ -543,52 +603,55 @@ mod tests { } #[tokio::test(start_paused = true)] - async fn a_relay_that_stays_down_is_failed_over_again_a_full_window_after_restore() { - // Failover moves nothing (status stays down): the relay is restored - // when connectable, and the next failover comes a full window later. + async fn a_relay_that_stays_down_is_nudged_while_out_and_removed_again_after_restore() { + // Removing A changes nothing (status stays down on A): while A is + // out, each further window only nudges a relay still in the map; + // once A is restored, the next window removes it again. let status = Watchable::new(Status::DownOnA); let mut relays = FakeRelays::new(&status, None); - run_for( - &status, - &mut relays, - (RELAY_OUTAGE_FAILOVER + RELAY_RESTORE_INTERVAL) * 2 + SEC, - ) - .await; - let cycle = RELAY_OUTAGE_FAILOVER + RELAY_RESTORE_INTERVAL; + run_for(&status, &mut relays, 290 * SEC).await; assert_eq!( relays.actions(), vec![ - Action::FailOver { - home: Some(relay_a()), - at: RELAY_OUTAGE_FAILOVER, + Action::Remove { + url: relay_a(), + at: 60 * SEC, + }, + Action::Nudge { + url: relay_b(), + at: 120 * SEC, }, Action::Restore { url: relay_a(), restored: true, - at: cycle, + at: 150 * SEC, + }, + Action::Remove { + url: relay_a(), + at: 180 * SEC, }, - Action::FailOver { - home: Some(relay_a()), - at: cycle + RELAY_OUTAGE_FAILOVER, + Action::Nudge { + url: relay_b(), + at: 240 * SEC, }, Action::Restore { url: relay_a(), restored: true, - at: cycle * 2, + at: 270 * SEC, }, ] ); } #[tokio::test(start_paused = true)] - async fn no_home_relay_selected_only_forces_a_report() { + async fn no_home_relay_selected_only_nudges() { let status = Watchable::new(Status::NoHome); let mut relays = FakeRelays::new(&status, Some(Status::UpOnB)); run_for(&status, &mut relays, RELAY_OUTAGE_FAILOVER * 3).await; assert_eq!( relays.actions(), - vec![Action::FailOver { - home: None, + vec![Action::Nudge { + url: relay_a(), at: RELAY_OUTAGE_FAILOVER, }], "nothing was removed, so nothing is restored" @@ -615,8 +678,8 @@ mod tests { assert_eq!( relays.actions(), vec![ - Action::FailOver { - home: Some(relay_a()), + Action::Remove { + url: relay_a(), at: 60 * SEC, }, Action::Restore { @@ -624,11 +687,54 @@ mod tests { restored: true, at: 150 * SEC, }, - Action::FailOver { - home: Some(relay_a()), + Action::Remove { + url: relay_a(), at: 260 * SEC, }, ] ); } + + #[tokio::test(start_paused = true)] + async fn a_second_wedged_relay_is_removed_while_the_first_is_still_out_but_never_the_last() { + // A is removed at 60 s and stays unconnectable. The endpoint moves to + // B, which wedges too: B is removed at 120 s while A is still out. + // Then C wedges, but it is the last relay in the map, so at 180 s it + // is only nudged. + let status = Watchable::new(Status::DownOnA); + let mut relays = FakeRelays::new(&status, None); + relays.connectable.store(false, Ordering::SeqCst); + let flipper = { + let status = status.clone(); + async move { + tokio::time::sleep(61 * SEC).await; + status.set(Status::DownOnB).ok(); + tokio::time::sleep(60 * SEC).await; + status.set(Status::DownOnC).ok(); + } + }; + tokio::join!(run_for(&status, &mut relays, 200 * SEC), flipper); + assert_eq!( + relays.actions(), + vec![ + Action::Remove { + url: relay_a(), + at: 60 * SEC, + }, + Action::Remove { + url: relay_b(), + at: 120 * SEC, + }, + Action::Restore { + url: relay_a(), + restored: false, + at: 150 * SEC, + }, + Action::Nudge { + url: relay_c(), + at: 180 * SEC, + }, + ] + ); + } }