Rebuild the client endpoint after repeated reconnect failures

crates/flextunnel-cli/src/client_session.rs

@@ -19,9 +19,7 @@ use flextunnel_core::forwards::{
 };
 use flextunnel_core::iroh::SecretKey;
 use flextunnel_core::proxy::{ClientAuth, ClientConfig, ProxyClient, reserved};
-use flextunnel_core::transport::endpoint::{
-    RelayConfig, create_client_endpoint, create_quick_client_endpoint,
-};
+use flextunnel_core::transport::endpoint::{ClientEndpoint, RelayConfig};
 use flextunnel_core::transport::paths::{ConnPath, ConnPathKind};
 use flextunnel_core::{app, auth, config};
 
@@ -157,7 +155,7 @@ pub async fn run_quick(r: config::ResolvedClient, client_secret: SecretKey) -> R
 /// endpoint, the proxy client and its live routes, the bound proxy listeners,
 /// the forward manager + set, and the status/mutation state.
 struct SessionRuntime {
-    endpoint: flextunnel_core::iroh::Endpoint,
+    endpoint: ClientEndpoint,
     client: std::sync::Arc<ProxyClient>,
     routes: std::sync::Arc<std::sync::Mutex<flextunnel_core::proxy::TunnelRoutes>>,
     socks_listener: Option<tokio::net::TcpListener>,
@@ -199,11 +197,11 @@ async fn build_session(
         .context("Invalid relay configuration")?;
     let (endpoint, auth) = match auth {
         SessionAuth::Key(client_key) => (
-            create_client_endpoint(&relay_config).await,
+            ClientEndpoint::create(&relay_config).await,
             ClientAuth::Key(Box::new(client_key)),
         ),
         SessionAuth::Quick(secret) => (
-            create_quick_client_endpoint(&relay_config, secret).await,
+            ClientEndpoint::create_quick(&relay_config, secret).await,
             ClientAuth::QuickAllowlisted,
         ),
     };
@@ -353,6 +351,10 @@ async fn drive_session(
                     fwd_mgr.apply(&forwards);
                 }
                 state.observe_connection(routes.lock().map(|r| r.connected).unwrap_or(false));
+                // The reconnect loop rebuilds the endpoint after repeated
+                // failures, which changes the (ephemeral) node id — keep the
+                // status display current.
+                state.client_node_id = endpoint.id().to_string();
             }
             cmd = ipc_rx.recv() => match cmd {
                 Some(IpcCmd::Status(reply)) => {
@@ -420,7 +422,7 @@ async fn drive_session(
         }
         IpcSink::Panel(panel) => Some(panel),
     };
-    crate::close_endpoint_or_exit(&endpoint).await;
+    crate::close_endpoint_or_exit(&endpoint.endpoint()).await;
     if let Some(panel) = panel {
         let _ = panel.await;
     }

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

@@ -8,13 +8,16 @@
 //!
 //! [`BRIDGE_ALPN`]: crate::transport::BRIDGE_ALPN
 //!
-//! The connect/auth/heartbeat machinery mirrors [`super::client`], with one
-//! deliberate difference in reconnect policy: a bridge retries **forever** (no
+//! The connect/auth/heartbeat machinery mirrors [`super::client`], with two
+//! deliberate differences in reconnect policy: a bridge retries **forever** (no
 //! fail-fast first connect, no attempt cap). The peer server may simply not be
 //! up yet, and a server daemon must not exit — or stop serving its other
 //! routes — because a peer is down. While the upstream is down, matching
 //! streams fail with host-unreachable (see `route_to_bridge` in
-//! [`super::server`]).
+//! [`super::server`]). And a bridge never escalates to the client's endpoint
+//! rebuild: it dials on the **server's own endpoint**, which is also accepting
+//! inbound clients on its persistent identity — rebuilding it to unwedge one
+//! upstream would sever every connected client.
 
 use crate::error::{ProxyError, ProxyResult};
 use crate::proxy::client::{calculate_backoff, client_heartbeat_loop, connect_with_timeout};

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

@@ -5,7 +5,7 @@
 use crate::error::{ProxyError, ProxyResult};
 use crate::proxy::signaling::{self, ControlMsg, Hello, Target};
 use crate::proxy::{dial, http, reserved, socks5, RoutedSet};
-use crate::transport::endpoint::RelayConfig;
+use crate::transport::endpoint::{ClientEndpoint, RelayConfig};
 use crate::transport::paths::{connection_paths, ConnPath, ConnectionSnapshot};
 use crate::transport::{HEARTBEAT_INTERVAL, HEARTBEAT_INTERVAL_IDLE, liveness_window};
 use anyhow::Result;
@@ -27,6 +27,16 @@ use tokio::sync::{Semaphore, watch};
 
 /// Reconnect backoff: base 1s, doubling per attempt, capped at 60s.
 const RECONNECT_BACKOFF_MAX: u64 = 60;
+/// Escalate to a full endpoint rebuild every this many consecutive failed
+/// reconnect attempts. The early attempts get the cheap `network_change()`
+/// nudge, which repairs dead UDP sockets; a wedge that survives the nudge plus
+/// two full connect timeouts is endpoint state a rebind cannot fix — a relay
+/// link lost to a ping timeout and never re-established, stale cached paths
+/// for the server — which only a fresh endpoint repairs (observed as "restart
+/// the client process and it connects instantly"; the rebuild is that restart,
+/// in-process). Rebuilding every Nth attempt (not once) keeps a long outage
+/// retrying from fresh state without paying the rebuild on every backoff.
+const REBUILD_ENDPOINT_ATTEMPTS: u32 = 3;
 /// Max jitter (ms) added to each backoff to avoid thundering reconnects.
 const RECONNECT_JITTER_MAX_MS: u64 = 500;
 /// Deadline for the server's handshake response. The QUIC keep-alive keeps the
@@ -448,8 +458,9 @@ impl ProxyClient {
     /// `--no-auto-reconnect` is set). The listeners stay bound across reconnects:
     /// off-list targets keep connecting directly, while on-list requests are held
     /// for the reconnect (failing with network-unreachable only after
-    /// [`TUNNEL_RECOVERY_HOLD`]).
-    pub async fn run(&self, endpoint: &Endpoint) -> ProxyResult<()> {
+    /// [`TUNNEL_RECOVERY_HOLD`]). Reconnects that keep failing escalate to a
+    /// full endpoint rebuild every [`REBUILD_ENDPOINT_ATTEMPTS`] attempts.
+    pub async fn run(&self, endpoint: &ClientEndpoint) -> ProxyResult<()> {
         let socks = match self.config.socks_listen {
             Some(addr) => Some(TcpListener::bind(addr).await?),
             None => None,
@@ -469,7 +480,7 @@ impl ProxyClient {
     /// This path never enables the HTTP front-end.
     pub async fn run_with_listener(
         &self,
-        endpoint: &Endpoint,
+        endpoint: &ClientEndpoint,
         listener: TcpListener,
     ) -> ProxyResult<()> {
         self.run_with_listeners(endpoint, listener, None).await
@@ -483,7 +494,7 @@ impl ProxyClient {
     /// proxy listener.
     pub async fn run_with_listeners(
         &self,
-        endpoint: &Endpoint,
+        endpoint: &ClientEndpoint,
         socks_listener: TcpListener,
         http_listener: Option<TcpListener>,
     ) -> ProxyResult<()> {
@@ -499,7 +510,7 @@ impl ProxyClient {
     /// connection. Both may be absent for a forwarding-only GUI session.
     pub async fn run_with_optional_listeners(
         &self,
-        endpoint: &Endpoint,
+        endpoint: &ClientEndpoint,
         socks_listener: Option<TcpListener>,
         http_listener: Option<TcpListener>,
     ) -> ProxyResult<()> {
@@ -526,7 +537,7 @@ impl ProxyClient {
     /// [`run_with_listeners`].
     pub async fn run_with_listeners_ext(
         &self,
-        endpoint: &Endpoint,
+        endpoint: &ClientEndpoint,
         socks_listener: TcpListener,
         http_listener: Option<TcpListener>,
         #[cfg(unix)] unix_listener: Option<UnixListener>,
@@ -543,7 +554,7 @@ impl ProxyClient {
 
     async fn run_with_optional_listeners_ext(
         &self,
-        endpoint: &Endpoint,
+        endpoint: &ClientEndpoint,
         socks_listener: Option<TcpListener>,
         http_listener: Option<TcpListener>,
         #[cfg(unix)] unix_listener: Option<UnixListener>,
@@ -628,7 +639,7 @@ impl ProxyClient {
     /// must succeed (fail fast); once connected, transient drops are retried.
     async fn manage_connection(
         &self,
-        endpoint: &Endpoint,
+        endpoint: &ClientEndpoint,
         current: &SharedConn,
         routed_set_shared: &SharedRoutedSet,
     ) -> ProxyResult<()> {
@@ -644,20 +655,34 @@ impl ProxyClient {
             self.set_connected(false);
             current.send_replace(None);
 
-            // Retrying after a failure: the endpoint's UDP sockets may be dead
-            // underneath it (iOS defuncts them while the process is suspended;
-            // a sleeping laptop can do the same) and iroh cannot always detect
-            // that by itself, leaving reconnects wedged forever. Nudging it
-            // re-checks and rebinds the transports — harmless when nothing
-            // actually changed.
-            if attempt > 0 || path_returned {
-                endpoint.network_change().await;
+            if attempt > 0 && attempt.is_multiple_of(REBUILD_ENDPOINT_ATTEMPTS) {
+                // Escalation: the nudge below wasn't enough — rebuild the
+                // endpoint from scratch (see [`REBUILD_ENDPOINT_ATTEMPTS`]).
+                // On a rebuild failure (e.g. no route to bind on a dead
+                // network) the current endpoint stays in place and this
+                // attempt proceeds with it — the next multiple retries the
+                // rebuild.
+                log::warn!("Reconnect still failing after {attempt} attempts; rebuilding the endpoint from scratch");
+                if let Err(e) = endpoint.rebuild().await {
+                    log::warn!("Endpoint rebuild failed ({e:#}); retrying with the current endpoint");
+                }
+            } else if attempt > 0 || path_returned {
+                // Retrying after a failure: the endpoint's UDP sockets may be dead
+                // underneath it (iOS defuncts them while the process is suspended;
+                // a sleeping laptop can do the same) and iroh cannot always detect
+                // that by itself, leaving reconnects wedged forever. Nudging it
+                // re-checks and rebinds the transports — harmless when nothing
+                // actually changed.
+                endpoint.endpoint().network_change().await;
             }
 
             // Establish (connect + auth). The handshake also learns the server's
             // tunnel set (drives split-tunneling) and returns the control-stream
-            // halves kept open for heartbeats.
-            let (connection, routed_set, ctrl_send, ctrl_recv) = match self.establish(endpoint).await
+            // halves kept open for heartbeats. The endpoint handle is taken
+            // fresh per attempt — a rebuild above swapped it.
+            let (connection, routed_set, ctrl_send, ctrl_recv) = match self
+                .establish(&endpoint.endpoint())
+                .await
             {
                 Ok(established) => {
                     ever_connected = true;

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

@@ -22,7 +22,7 @@ use crate::proxy::{
     BridgeUpstream, BridgeUpstreamConfig, ClientAuth, ClientConfig, ForwardManager, ForwardSpec,
     ProxyClient, ProxyServer, ProxyServerParams, RoutedSet, ServerForwarder,
 };
-use crate::transport::endpoint::{AllowlistHook, EndpointAllowlists};
+use crate::transport::endpoint::{AllowlistHook, ClientEndpoint, EndpointAllowlists};
 use crate::transport::{ALPN, BRIDGE_ALPN, QUICK_ALPN, build_quic_transport_config};
 use iroh::address_lookup::MemoryLookup;
 use iroh::endpoint::{presets, Connection, RecvStream, SendStream};
@@ -465,6 +465,101 @@ async fn server_direct_forward_relays_and_server_rejects_off_list_target() {
     let _ = std::fs::remove_file(bl_path);
 }
 
+/// Endpoint-rebuild escalation: when reconnect attempts keep failing on an
+/// endpoint that is broken beyond repair (here: closed out from under the
+/// session — the loopback stand-in for a wedged endpoint whose relay link or
+/// path state never recovers), the reconnect loop must escalate to rebuilding
+/// the endpoint via its factory and reconnect on the fresh one. The old
+/// endpoint can never connect again, so recovery itself proves the rebuild
+/// ran; the changed node id confirms it.
+#[tokio::test]
+async fn reconnect_rebuilds_a_dead_endpoint() {
+    let server_ep = loopback_endpoint(SecretKey::generate(), true).await;
+    let server_id = server_ep.id();
+    let server_addr = EndpointAddr::new(server_id).with_ip_addr(server_ep.bound_sockets()[0]);
+    let bl_path = temp_blocklist("rebuild-endpoint");
+    let (routed_set, routed_cidrs) = loopback_cidr_set();
+    spawn_server_params(
+        server_ep.clone(),
+        ProxyServerParams {
+            routed_set,
+            routed_cidrs,
+            ..base_params(server_id, bl_path.clone())
+        },
+    );
+
+    let lookup = MemoryLookup::from_endpoint_info(vec![server_addr]);
+    let first_ep = loopback_endpoint_full(
+        SecretKey::generate(),
+        false,
+        lookup.clone(),
+        EndpointAllowlists::default(),
+    )
+    .await;
+    let client_ep = ClientEndpoint::from_parts(first_ep.clone(), {
+        let lookup = lookup.clone();
+        Arc::new(move || {
+            let lookup = lookup.clone();
+            Box::pin(async move {
+                Ok(loopback_endpoint_full(
+                    SecretKey::generate(),
+                    false,
+                    lookup,
+                    EndpointAllowlists::default(),
+                )
+                .await)
+            })
+        })
+    });
+    let first_id = client_ep.id();
+
+    let client = Arc::new(ProxyClient::new(ClientConfig {
+        server_node_id: server_id.to_string(),
+        auth: ClientAuth::Key(Box::new(test_client_key().clone())),
+        socks_listen: None,
+        http_listen: None,
+        relay_urls: Vec::new(),
+        relay_auth_token: None,
+        auto_reconnect: true,
+        max_reconnect_attempts: None,
+    }));
+    let socks_listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    {
+        let (client, ep) = (client.clone(), client_ep.clone());
+        tokio::spawn(async move {
+            if let Err(e) = client.run_with_listener(&ep, socks_listener).await {
+                eprintln!("e2e rebuild test client session ended: {e}");
+            }
+        });
+    }
+    let connected = || client.routes().lock().unwrap().connected;
+    wait_until("client to connect", connected).await;
+
+    // Kill the client's own endpoint. Every reconnect attempt on it fails
+    // immediately, so the backoff series reaches the rebuild escalation well
+    // inside the wait window.
+    first_ep.close().await;
+    wait_until("client to notice the drop", || !connected()).await;
+    // Own, longer bound: reaching the rebuild takes the full early backoff
+    // series (1s + 2s + 4s plus jitter), which crowds `wait_until`'s 10s.
+    let start = Instant::now();
+    while !connected() {
+        assert!(
+            start.elapsed() < Duration::from_secs(20),
+            "timed out waiting for client to reconnect on a rebuilt endpoint"
+        );
+        tokio::time::sleep(Duration::from_millis(25)).await;
+    }
+
+    assert_ne!(
+        client_ep.id(),
+        first_id,
+        "recovery must have swapped in a freshly built endpoint"
+    );
+
+    let _ = std::fs::remove_file(bl_path);
+}
+
 /// Deploy-style connection holding: a SOCKS request for an on-list target
 /// arriving while the tunnel link is down is *held* for the client's own
 /// reconnect and then proceeds transparently on the fresh connection, instead
@@ -502,6 +597,24 @@ async fn on_list_request_is_held_across_reconnect() {
         EndpointAllowlists::default(),
     )
     .await;
+    // Hermetic rebuild recipe: should the session escalate to an endpoint
+    // rebuild mid-test, the replacement is another loopback endpoint sharing
+    // the same externally-held lookup.
+    let client_ep = ClientEndpoint::from_parts(client_ep, {
+        let lookup = lookup.clone();
+        Arc::new(move || {
+            let lookup = lookup.clone();
+            Box::pin(async move {
+                Ok(loopback_endpoint_full(
+                    SecretKey::generate(),
+                    false,
+                    lookup,
+                    EndpointAllowlists::default(),
+                )
+                .await)
+            })
+        })
+    });
 
     // The full reconnecting client session with a real SOCKS front-end.
     let client = Arc::new(ProxyClient::new(ClientConfig {