Add a server-side home-relay watchdog with in-process endpoint rebuild

Cargo.toml

@@ -10,7 +10,7 @@ default-members = [
 ]
 
 [workspace.package]
-version = "0.0.69"
+version = "0.0.70"
 edition = "2024"
 description = "SOCKS5/HTTP-proxy-over-QUIC split tunnel via iroh P2P"
 
@@ -33,6 +33,9 @@ iroh-mdns-address-lookup = "0.4.0"
 libc = "0.2"
 log = "0.4"
 n0-future = "0.3"
+# Test double for iroh's `Watcher`-based status APIs (the relay watchdog tests
+# drive a plain `Watchable`); the same crate iroh itself re-exports `Watcher` from.
+n0-watcher = "1"
 rand = "0.9"
 ratatui = "0.30"
 # Custom-relay `/healthz` checks (see `transport::paths`). `rustls-no-provider`

README.md

@@ -619,6 +619,13 @@ Auto-reconnect is **enabled by default** (`auto_reconnect = true`); pass
   connecting directly; on-list requests are held for the reconnect (up to 45s)
   and only then fail with a network-unreachable reply.
 
+A **server** with custom relays watches its own home-relay registration: if it
+has no connected home relay for 60s it re-checks the network, and if that
+has not helped by 180s it rebuilds its endpoint in place (same server id) —
+the in-process equivalent of a restart, so relay-only clients (the iOS app,
+anything off the LAN) are not stranded until someone restarts the service.
+See [`docs/architecture.md`](docs/architecture.md#relay-watchdog-server-custom-relays).
+
 ## Logging
 
 Logging uses `env_logger`. The default is `info` with iroh/tracing quieted to

crates/flextunnel-cli/src/main.rs

@@ -13,6 +13,7 @@ use clap::{Parser, Subcommand};
 use std::io::IsTerminal;
 use std::num::NonZeroU32;
 use std::path::PathBuf;
+use std::pin::pin;
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::Notify;
@@ -33,7 +34,9 @@ use flextunnel_core::proxy::{
 };
 use flextunnel_core::transport::endpoint::{
     EndpointAllowlists, RelayConfig, create_server_endpoint, secret_to_endpoint_id,
+    server_rebuild_factory,
 };
+use flextunnel_core::transport::relay_watchdog;
 use flextunnel_core::{auth, config, secret};
 
 #[derive(Parser)]
@@ -614,6 +617,12 @@ struct QuickServer {
 /// relay/connection teardown must never leave the process unkillable.
 const SHUTDOWN_CLOSE_TIMEOUT: Duration = Duration::from_secs(5);
 
+/// Pause between attempts to bind a replacement endpoint after the relay
+/// watchdog retired the old one and the rebuild itself failed (e.g. no route
+/// to bind on). The server has no endpoint at all during this wait, so it is
+/// short — there is nothing to lose by trying again soon.
+const REBUILD_RETRY: Duration = Duration::from_secs(30);
+
 /// Build the ephemeral `ServerConfig` for `server start --quick`: a full-tunnel
 /// routed set (`routed_domains = ["*"]`, `routed_cidrs = ["0.0.0.0/0", "::/0"]`)
 /// plus a freshly generated in-memory identity, returned *alongside* the config —
@@ -798,9 +807,16 @@ async fn run_server(
             .map(|q| std::collections::HashSet::from([q.client_id]))
             .unwrap_or_default(),
     };
-    let endpoint = create_server_endpoint(&relay_config, secret_key, allowlists)
+    let endpoint = create_server_endpoint(&relay_config, secret_key.clone(), allowlists.clone())
         .await
         .context("Failed to create iroh endpoint")?;
+    // The relay watchdog's remedy of last resort: a fresh endpoint with the
+    // same identity and allowlists (see the serve loop below).
+    let rebuild = server_rebuild_factory(relay_config.clone(), secret_key, allowlists);
+    // Only a custom-relay server hangs its reachability on one home-relay
+    // registration (n0 discovery is off, clients dial by relay hint), so the
+    // watchdog is armed for custom relays only.
+    let relay_watchdog_armed = relay_config.is_custom();
 
     log::info!("flextunnel server Node ID: {}", endpoint.id());
     match &quick {
@@ -844,20 +860,13 @@ async fn run_server(
         blocklist,
         first_client: first_client.clone(),
     });
-    let run = async {
-        server
-            .run(&endpoint)
-            .await
-            .map_err(|e| anyhow::anyhow!("Server error: {e}"))
-    };
-
     // Quick-mode grace window: fire after `QUICK_IDLE_TIMEOUT` unless a client
     // connects first (which resolves `notified()` and parks this future
     // forever). `first_client` is `Some` exactly in quick mode; a normal server
     // parks here immediately, so the arm never fires. `notify_one` stores a
     // permit, so a client that connects before this future is first polled is
-    // not missed.
-    let grace = async {
+    // not missed. Pinned outside the serve loop so it spans endpoint rebuilds.
+    let mut grace = pin!(async {
         match &first_client {
             Some(notify) => {
                 tokio::select! {
@@ -867,19 +876,88 @@ async fn run_server(
             }
             None => std::future::pending::<()>().await,
         }
-    };
+    });
+    // One signal listener for the whole serve loop: re-registering it per pass
+    // could drop a signal delivered while an endpoint is being rebuilt.
+    let mut shutdown = pin!(app::shutdown_signal());
+
+    /// How one pass of the serve loop ended.
+    enum Pass {
+        /// The server is done (clean or failed): close the endpoint and return.
+        Exit(Result<()>),
+        /// The relay watchdog gave up on the endpoint after an outage this long.
+        Rebuild(Duration),
+    }
+
+    // Serve loop. A pass serves on the current endpoint until the server ends,
+    // a shutdown signal arrives, the quick-mode grace expires, or — custom
+    // relays only — the relay watchdog reports the endpoint has lost its home
+    // relay for good. That last case is the in-process equivalent of the
+    // process restart known to fix it: close the wedged endpoint, bind a fresh
+    // one with the same identity, and serve again. The `ProxyServer` (its
+    // registries, blocklist, status state) carries over; the old endpoint's
+    // connections and bridge tasks end with it.
+    let mut endpoint = endpoint;
+    let res = loop {
+        let pass = {
+            let run = Arc::clone(&server).run(&endpoint);
+            let outage = async {
+                if relay_watchdog_armed {
+                    relay_watchdog::watch_home_relay(&endpoint).await
+                } else {
+                    std::future::pending().await
+                }
+            };
+            tokio::select! {
+                res = run => Pass::Exit(res.map_err(|e| anyhow::anyhow!("Server error: {e}"))),
+                sig = &mut shutdown => Pass::Exit(sig.map(|()| {
+                    log::info!("Received shutdown signal, stopping server");
+                })),
+                _ = &mut grace => {
+                    log::warn!("Quick mode: no client connected within 5 minutes — exiting");
+                    Pass::Exit(Ok(()))
+                }
+                outage = outage => Pass::Rebuild(outage),
+            }
+        };
+        let outage = match pass {
+            Pass::Exit(res) => break res,
+            Pass::Rebuild(outage) => outage,
+        };
 
-    let res = tokio::select! {
-        res = run => res,
-        sig = app::shutdown_signal() => {
-            sig?;
-            log::info!("Received shutdown signal, stopping server");
-            Ok(())
-        }
-        _ = grace => {
-            log::warn!("Quick mode: no client connected within 5 minutes — exiting");
-            Ok(())
-        }
+        log::error!(
+            "No connected home relay for {:.0}s despite a network re-check; rebuilding the \
+             endpoint from scratch (server id stays {})",
+            outage.as_secs_f64(),
+            endpoint.id()
+        );
+        close_endpoint_or_exit(&endpoint).await;
+        endpoint = loop {
+            match rebuild().await {
+                Ok(fresh) => break fresh,
+                Err(e) => {
+                    log::error!(
+                        "Endpoint rebuild failed: {e:#}; retrying in {}s",
+                        REBUILD_RETRY.as_secs()
+                    );
+                    // Nothing is bound while waiting here, so both exits
+                    // below return directly: there is no endpoint to close.
+                    tokio::select! {
+                        _ = tokio::time::sleep(REBUILD_RETRY) => {}
+                        sig = &mut shutdown => {
+                            sig?;
+                            log::info!("Received shutdown signal, stopping server");
+                            return Ok(());
+                        }
+                        _ = &mut grace => {
+                            log::warn!("Quick mode: no client connected within 5 minutes — exiting");
+                            return Ok(());
+                        }
+                    }
+                }
+            }
+        };
+        log::warn!("Endpoint rebuilt; serving again as {}", endpoint.id());
     };
 
     close_endpoint_or_exit(&endpoint).await;

crates/flextunnel-core/Cargo.toml

@@ -41,4 +41,5 @@ libc.workspace = true
 iroh-mdns-address-lookup.workspace = true
 
 [dev-dependencies]
+n0-watcher.workspace = true
 tempfile.workspace = true

crates/flextunnel-core/src/proxy/bridge.rs

@@ -106,6 +106,10 @@ impl BridgeUpstream {
     /// only when `endpoint` closes underneath it, failing each retry.
     pub async fn run(self: Arc<Self>, endpoint: Endpoint) {
         let name = &self.config.name;
+        // A previous run (on a since-closed endpoint, see the server's relay
+        // watchdog rebuild) may have been aborted while connected; its stale
+        // connection must not read as live until this run establishes its own.
+        *self.conn.lock().expect("bridge conn lock") = None;
         let mut attempt: u32 = 0;
         loop {
             if attempt > 0 {

crates/flextunnel-core/src/proxy/server.rs

@@ -307,14 +307,22 @@ impl ProxyServer {
     }
 
     /// Accept connections until the endpoint closes or the server self-blocks.
+    ///
+    /// May be called again on a *fresh* endpoint after the previous one was
+    /// closed (the relay watchdog's rebuild): the registries and blocklist
+    /// carry over, the old endpoint's connection handlers end as its
+    /// connections close, and its bridge tasks are aborted when the previous
+    /// `run` future is dropped.
     pub async fn run(self: Arc<Self>, endpoint: &Endpoint) -> ProxyResult<()> {
-        // Maintain the outbound bridge upstreams for the life of the process.
-        // The bridging side dials out on this same server endpoint, so the TLS
+        // Maintain the outbound bridge upstreams for the life of this run. The
+        // bridging side dials out on this same server endpoint, so the TLS
         // identity it presents is this server's persistent id — what the target
-        // server's allowlist matches. The tasks retry forever and die with the
-        // endpoint.
+        // server's allowlist matches. The tasks retry forever; owning them in a
+        // `JoinSet` aborts them with this future, so a rebuild never leaves
+        // bridges retrying on a closed endpoint.
+        let mut bridge_tasks = tokio::task::JoinSet::new();
         for bridge in &self.bridges {
-            tokio::spawn(bridge.clone().run(endpoint.clone()));
+            bridge_tasks.spawn(bridge.clone().run(endpoint.clone()));
         }
 
         let conn_limit = Arc::new(Semaphore::new(MAX_CONCURRENT_CONNECTIONS));

crates/flextunnel-core/src/transport/endpoint.rs

@@ -30,7 +30,7 @@ pub const CLOSE_NOT_ALLOWLISTED: u32 = 3;
 /// The server's per-ALPN endpoint-id allowlists, enforced natively at the TLS
 /// handshake by [`AllowlistHook`]. An empty set disables its ALPN entirely —
 /// the allowlist is the sole and mandatory credential on both.
-#[derive(Debug, Default)]
+#[derive(Debug, Default, Clone)]
 pub struct EndpointAllowlists {
     /// Servers allowed to bridge into this server over [`BRIDGE_ALPN`].
     pub bridge_servers: HashSet<EndpointId>,
@@ -436,15 +436,7 @@ pub async fn create_server_endpoint(
     // no-op for the default relays.
     probe_custom_relays(relay_config).await?;
 
-    let builder = create_endpoint_builder(relay_config, Some(&secret))?
-        .alpns(vec![ALPN.to_vec(), BRIDGE_ALPN.to_vec(), QUICK_ALPN.to_vec()])
-        .hooks(AllowlistHook::new(allowlists))
-        .secret_key(secret);
-
-    let endpoint = builder
-        .bind()
-        .await
-        .context("Failed to create iroh endpoint")?;
+    let endpoint = bind_server_endpoint(relay_config, secret, allowlists).await?;
 
     if let Err(e) = wait_online(&endpoint).await {
         // Close before propagating: dropping a bound endpoint without
@@ -455,6 +447,52 @@ pub async fn create_server_endpoint(
     Ok(endpoint)
 }
 
+/// Bind a server endpoint: persistent identity, all three server ALPNs, the
+/// allowlist hook. No relay probe, no online wait — [`create_server_endpoint`]
+/// and [`server_rebuild_factory`] layer their own policy over this.
+async fn bind_server_endpoint(
+    relay_config: &RelayConfig,
+    secret: SecretKey,
+    allowlists: EndpointAllowlists,
+) -> Result<Endpoint> {
+    let builder = create_endpoint_builder(relay_config, Some(&secret))?
+        .alpns(vec![ALPN.to_vec(), BRIDGE_ALPN.to_vec(), QUICK_ALPN.to_vec()])
+        .hooks(AllowlistHook::new(allowlists))
+        .secret_key(secret);
+    builder.bind().await.context("Failed to create iroh endpoint")
+}
+
+/// The rebuild recipe for the server endpoint, used when the relay watchdog
+/// (`transport::relay_watchdog`) gives up on the current one. Same identity
+/// and allowlists as the original, so the server's id — what clients dial —
+/// never changes. Differs from first creation the same way the client's
+/// rebuild does:
+///
+/// - **No per-relay probe.** Creation fails fast if *any* relay is down
+///   (configuration validation); mid-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 clients can still find it over
+///   mDNS — and the watchdog trips again if the relays stay unreachable.
+pub fn server_rebuild_factory(
+    relay_config: RelayConfig,
+    secret: SecretKey,
+    allowlists: EndpointAllowlists,
+) -> EndpointFactory {
+    Arc::new(move || {
+        let relay_config = relay_config.clone();
+        let secret = secret.clone();
+        let allowlists = allowlists.clone();
+        Box::pin(async move {
+            let endpoint = bind_server_endpoint(&relay_config, secret, allowlists).await?;
+            if let Err(e) = wait_online(&endpoint).await {
+                log::warn!("Rebuilt endpoint: {e:#}; continuing (LAN discovery still works)");
+            }
+            Ok(endpoint)
+        })
+    })
+}
+
 /// Bind a client endpoint: no relay probe, no online wait — callers layer
 /// their own creation-vs-rebuild policy over this.
 async fn bind_client_endpoint(

crates/flextunnel-core/src/transport/mod.rs

@@ -7,6 +7,7 @@
 
 pub mod endpoint;
 pub mod paths;
+pub mod relay_watchdog;
 
 use anyhow::{Context, Result};
 use iroh::endpoint::QuicTransportConfig;