feat(fips): resilience — connectivity watcher with immediate anchor re-apply, rebindable peer listener, cached service probe, warm-path union

Phase A3 of docs/FIPS-UPTIME-AND-UI-STATE-PLAN.md (RC5), measured against
the A2 dial_stats baseline:

- 25s connectivity watcher in the fips supervisor: re-applies seed anchors
  immediately on an anchor-link drop, on startup-disconnected, AND on
  silent data-path death (connect_fails growing with zero fips_ok — the
  live .198 failure where the daemon reported "connected" while every
  dial blackholed and the 300s tick never healed it). Bounded to one
  re-apply per 60s.
- anchors::apply is now concurrent with a 15s per-connect cap — the old
  serial loop waited unbounded on each `sudo fipsctl connect`, so one
  hung subprocess stalled the whole periodic tick.
- rebindable peer listener: the accept loop returns after persistent
  accept errors (was: continue forever = inbound-dead until restart) and
  peer_late_bind_loop rebinds — also on fips0 ULA change.
- is_service_active gets a 10s TTL cache (was up to 2 systemctl spawns
  per dial attempt and per warm-tick peer).
- the warm tick now warms the union of federation peers + configured
  seed anchors (direct anchor links used to go cold between 300s ticks),
  skipping the redundant per-peer service check.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

core/archipelago/src/fips/anchors.rs

@@ -232,26 +232,32 @@ pub async fn remove(data_dir: &Path, npub: &str) -> Result<Vec<SeedAnchor>> {
 /// leaving `anchor_connected=false` and every peer dial falling back to
 /// a slow Tor timeout.
 pub async fn apply(anchors: &[SeedAnchor]) -> Vec<ApplyResult> {
-    let mut results = Vec::with_capacity(anchors.len());
-    for anchor in anchors {
-        let out = Command::new("sudo")
-            .args([
-                "-n",
-                "fipsctl",
-                "connect",
-                &anchor.npub,
-                &anchor.address,
-                &anchor.transport,
-            ])
-            .output()
-            .await;
+    // Concurrent, each connect hard-capped: the old serial loop waited
+    // unbounded on every `sudo fipsctl connect`, so one hung subprocess
+    // stalled the whole apply — and the periodic anchor tick behind it,
+    // which is exactly when a wedged daemon most needs the re-apply.
+    let futs = anchors.iter().cloned().map(|anchor| async move {
+        let out = tokio::time::timeout(
+            std::time::Duration::from_secs(15),
+            Command::new("sudo")
+                .args([
+                    "-n",
+                    "fipsctl",
+                    "connect",
+                    &anchor.npub,
+                    &anchor.address,
+                    &anchor.transport,
+                ])
+                .output(),
+        )
+        .await;
         let result = match out {
-            Ok(o) if o.status.success() => ApplyResult {
+            Ok(Ok(o)) if o.status.success() => ApplyResult {
                 npub: anchor.npub.clone(),
                 ok: true,
                 message: String::from_utf8_lossy(&o.stdout).trim().to_string(),
             },
-            Ok(o) => ApplyResult {
+            Ok(Ok(o)) => ApplyResult {
                 npub: anchor.npub.clone(),
                 ok: false,
                 message: format!(
@@ -260,11 +266,16 @@ pub async fn apply(anchors: &[SeedAnchor]) -> Vec<ApplyResult> {
                     String::from_utf8_lossy(&o.stderr).trim()
                 ),
             },
-            Err(e) => ApplyResult {
+            Ok(Err(e)) => ApplyResult {
                 npub: anchor.npub.clone(),
                 ok: false,
                 message: format!("sudo fipsctl launch failed: {}", e),
             },
+            Err(_) => ApplyResult {
+                npub: anchor.npub.clone(),
+                ok: false,
+                message: "sudo fipsctl connect timed out after 15s".to_string(),
+            },
         };
         if result.ok {
             tracing::debug!(npub = %result.npub, "Seed anchor applied");
@@ -275,9 +286,9 @@ pub async fn apply(anchors: &[SeedAnchor]) -> Vec<ApplyResult> {
                 "Seed anchor apply failed (non-fatal)"
             );
         }
-        results.push(result);
-    }
-    results
+        result
+    });
+    futures_util::future::join_all(futs).await
 }
 
 /// Outcome of a single `fipsctl connect` call.

core/archipelago/src/fips/dial.rs

@@ -151,6 +151,12 @@ pub async fn warm_path(npub: &str) {
     if !is_service_active().await {
         return;
     }
+    warm_path_unchecked(npub).await
+}
+
+/// [`warm_path`] without the service-active check — for callers (the warm
+/// tick) that already verified the daemon once for the whole batch.
+pub async fn warm_path_unchecked(npub: &str) {
     let Ok(base) = peer_base_url(npub).await else {
         return;
     };
@@ -277,21 +283,39 @@ pub fn as_ip_addr(v6: Ipv6Addr) -> IpAddr {
 
 // ── High-level peer request helpers ────────────────────────────────────
 
+/// TTL for the [`is_service_active`] cache. Every FIPS dial attempt and
+/// every warm-tick peer used to spawn up to two `systemctl` subprocesses;
+/// service state changes on human timescales, so 10s staleness is free.
+const SERVICE_ACTIVE_TTL_MS: u64 = 10_000;
+static SERVICE_ACTIVE: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);
+static SERVICE_PROBED_AT_MS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
+
 /// Quick poll: is the FIPS daemon (archipelago-supervised OR upstream)
-/// currently `systemctl is-active`? Async wrapper intended for the
-/// migration call sites; unlike `FipsTransport::is_available` this does
-/// not maintain a cache, so callers that poll frequently should cache
-/// themselves.
+/// currently `systemctl is-active`? Cached for [`SERVICE_ACTIVE_TTL_MS`];
+/// concurrent refreshes are harmless (idempotent probe, last write wins).
 pub async fn is_service_active() -> bool {
+    use std::sync::atomic::Ordering;
+    let now_ms = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_millis() as u64)
+        .unwrap_or(0);
+    let probed_at = SERVICE_PROBED_AT_MS.load(Ordering::Relaxed);
+    if probed_at != 0 && now_ms.saturating_sub(probed_at) < SERVICE_ACTIVE_TTL_MS {
+        return SERVICE_ACTIVE.load(Ordering::Relaxed);
+    }
+    let mut active = false;
     for unit in [
         crate::fips::SERVICE_UNIT,
         crate::fips::UPSTREAM_SERVICE_UNIT,
     ] {
         if crate::fips::service::unit_state(unit).await == "active" {
-            return true;
+            active = true;
+            break;
         }
     }
-    false
+    SERVICE_ACTIVE.store(active, Ordering::Relaxed);
+    SERVICE_PROBED_AT_MS.store(now_ms, Ordering::Relaxed);
+    active
 }
 
 /// Builder for a peer request that may be sent over FIPS (preferred) or

core/archipelago/src/fips/mod.rs

@@ -80,25 +80,73 @@ pub async fn ensure_activated(data_dir: &std::path::Path) {
 pub fn spawn_fips_supervisor(data_dir: std::path::PathBuf) {
     tokio::spawn(async move {
         let mut tick = tokio::time::interval(std::time::Duration::from_secs(25));
+        // Connectivity watcher state: re-apply seed anchors the moment the
+        // anchor link drops (edge) or the data path degrades (dials keep
+        // failing with zero successes), instead of waiting for the 300s
+        // anchor tick. Bounded: at most one re-apply per RE_APPLY_BACKOFF.
+        const RE_APPLY_BACKOFF: std::time::Duration = std::time::Duration::from_secs(60);
+        let mut prev_connected: Option<bool> = None;
+        let mut prev_totals = telemetry::totals();
+        let mut last_apply: Option<std::time::Instant> = None;
         loop {
             tick.tick().await;
             // Bring FIPS up on its own once onboarding has materialised the key.
             ensure_activated(&data_dir).await;
             if !dial::is_service_active().await {
+                prev_connected = None; // daemon restart = fresh edge detection
                 continue;
             }
+
+            // ── Warm the union of federation peers + configured seed
+            // anchors. Warming only federation npubs left the direct
+            // anchors (vps2, LAN peers) to go cold between 300s ticks.
             let nodes = crate::federation::load_nodes(&data_dir)
                 .await
                 .unwrap_or_default();
+            let seed = anchors::load(&data_dir).await.unwrap_or_default();
+            let mut warm_npubs: std::collections::BTreeSet<String> = nodes
+                .iter()
+                .filter_map(|n| n.fips_npub.clone())
+                .collect();
+            warm_npubs.extend(seed.iter().map(|a| a.npub.clone()));
             let mut handles = Vec::new();
-            for node in nodes {
-                if let Some(npub) = node.fips_npub.clone() {
-                    handles.push(tokio::spawn(async move { dial::warm_path(&npub).await }));
-                }
+            for npub in warm_npubs {
+                // Service-active was checked once above for the whole batch.
+                handles.push(tokio::spawn(
+                    async move { dial::warm_path_unchecked(&npub).await },
+                ));
             }
             for h in handles {
                 let _ = h.await;
             }
+
+            // ── Connectivity watcher: detect anchor-link loss AND silent
+            // data-path death (daemon reports "connected" but every dial
+            // connect-fails — observed live on .198, 2026-07-27, where the
+            // 300s tick never healed it).
+            let mut anchor_npubs = vec![service::PUBLIC_ANCHOR_NPUB.to_string()];
+            anchor_npubs.extend(seed.iter().map(|a| a.npub.clone()));
+            let (_, connected) = service::peer_connectivity_summary(&anchor_npubs).await;
+            let totals = telemetry::totals();
+            let link_dropped = prev_connected == Some(true) && !connected;
+            let never_connected = prev_connected.is_none() && !connected;
+            let data_path_dead =
+                totals.1.saturating_sub(prev_totals.1) >= 5 && totals.0 == prev_totals.0;
+            prev_connected = Some(connected);
+            prev_totals = totals;
+
+            let backoff_ok = last_apply.is_none_or(|t| t.elapsed() >= RE_APPLY_BACKOFF);
+            if (link_dropped || never_connected || data_path_dead) && backoff_ok && !seed.is_empty()
+            {
+                tracing::info!(
+                    link_dropped,
+                    never_connected,
+                    data_path_dead,
+                    "FIPS connectivity degraded — re-applying seed anchors now"
+                );
+                last_apply = Some(std::time::Instant::now());
+                let _ = anchors::apply(&seed).await;
+            }
         }
     });
 }

core/archipelago/src/fips/telemetry.rs

@@ -76,6 +76,17 @@ pub fn record_fallback(reason: FallbackReason) {
     counter(reason).fetch_add(1, Ordering::Relaxed);
 }
 
+/// `(fips_ok, connect_fail)` totals for the connectivity watcher: a window
+/// where connect_fail grows while fips_ok doesn't is a degraded data path —
+/// including the "daemon says connected but packets blackhole" failure the
+/// link-state check alone can't see (observed live 2026-07-27 on .198).
+pub fn totals() -> (u64, u64) {
+    (
+        FIPS_OK.load(Ordering::Relaxed),
+        CONNECT_FAIL.load(Ordering::Relaxed),
+    )
+}
+
 /// Snapshot for `fips.status` (`dial_stats`). Process-lifetime counts.
 pub fn snapshot() -> serde_json::Value {
     let f1 = NO_NPUB.load(Ordering::Relaxed);

core/archipelago/src/server.rs

@@ -1217,18 +1217,36 @@ async fn peer_late_bind_loop(
                     }
                 };
                 info!("FIPS peer listener bound {}", addr);
-                // Once bound, serve until shutdown fires. accept_loop
-                // returns on shutdown, which also ends this outer loop.
-                accept_loop(
-                    handler,
-                    listener,
-                    active_connections,
-                    true, // peer listener: apply path filter
-                    shutdown_rx,
-                    addr,
-                )
-                .await;
-                return;
+                // Serve until shutdown, a persistent accept failure, or a
+                // fips0 ULA change. The listener must be REBINDABLE: a
+                // daemon re-key tears fips0 down and brings it back with a
+                // (possibly different) ULA, and the old one-shot bind left
+                // the node inbound-dead over FIPS until process restart.
+                tokio::select! {
+                    _ = accept_loop(
+                        handler.clone(),
+                        listener,
+                        active_connections.clone(),
+                        true, // peer listener: apply path filter
+                        shutdown_rx.clone(),
+                        addr,
+                    ) => {
+                        if *shutdown_rx.borrow() { return; }
+                        warn!("FIPS peer accept loop ended — rebinding");
+                    }
+                    _ = async {
+                        loop {
+                            tokio::time::sleep(std::time::Duration::from_secs(30)).await;
+                            if crate::fips::iface::fips0_ula() != Some(ip) {
+                                break;
+                            }
+                        }
+                    } => {
+                        info!("fips0 ULA changed — rebinding FIPS peer listener");
+                        // Dropping the select arm cancels accept_loop and
+                        // frees the socket; the outer loop rebinds fresh.
+                    }
+                }
             }
             _ = shutdown_rx.changed() => {
                 if *shutdown_rx.borrow() { return; }
@@ -1280,13 +1298,26 @@ async fn accept_loop(
     mut shutdown_rx: tokio::sync::watch::Receiver<bool>,
     local_addr: SocketAddr,
 ) {
+    // Consecutive accept-error tracking: a fips0 teardown/re-key leaves the
+    // peer listener's socket permanently broken — `continue`-ing forever
+    // made the node inbound-dead over FIPS until process restart. After a
+    // burst of consecutive errors the peer accept loop returns so its
+    // caller (peer_late_bind_loop) can rebind on the current ULA.
+    let mut consecutive_errors: u32 = 0;
     loop {
         tokio::select! {
             result = listener.accept() => {
                 let (stream, peer_addr) = match result {
-                    Ok(c) => c,
+                    Ok(c) => { consecutive_errors = 0; c }
                     Err(e) => {
                         error!("{} accept error: {}", local_addr, e);
+                        consecutive_errors += 1;
+                        if peer_only && consecutive_errors >= 10 {
+                            warn!("{} accept failing persistently — returning for rebind", local_addr);
+                            return;
+                        }
+                        // Don't hot-loop on a dead socket.
+                        tokio::time::sleep(std::time::Duration::from_millis(200)).await;
                         continue;
                     }
                 };