From a3f07d5ac6b990561f24a8553834a4c8558513a9 Mon Sep 17 00:00:00 2001 From: archipelago Date: Tue, 28 Jul 2026 03:57:12 -0400 Subject: [PATCH] =?UTF-8?q?feat(fips):=20resilience=20=E2=80=94=20connecti?= =?UTF-8?q?vity=20watcher=20with=20immediate=20anchor=20re-apply,=20rebind?= =?UTF-8?q?able=20peer=20listener,=20cached=20service=20probe,=20warm-path?= =?UTF-8?q?=20union?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- core/archipelago/src/fips/anchors.rs | 49 +++++++++++++--------- core/archipelago/src/fips/dial.rs | 36 +++++++++++++--- core/archipelago/src/fips/mod.rs | 56 +++++++++++++++++++++++-- core/archipelago/src/fips/telemetry.rs | 11 +++++ core/archipelago/src/server.rs | 57 ++++++++++++++++++++------ 5 files changed, 167 insertions(+), 42 deletions(-) diff --git a/core/archipelago/src/fips/anchors.rs b/core/archipelago/src/fips/anchors.rs index e58746d7..077021d7 100644 --- a/core/archipelago/src/fips/anchors.rs +++ b/core/archipelago/src/fips/anchors.rs @@ -232,26 +232,32 @@ pub async fn remove(data_dir: &Path, npub: &str) -> Result> { /// leaving `anchor_connected=false` and every peer dial falling back to /// a slow Tor timeout. pub async fn apply(anchors: &[SeedAnchor]) -> Vec { - 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 { 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 { "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. diff --git a/core/archipelago/src/fips/dial.rs b/core/archipelago/src/fips/dial.rs index 64c22e80..cb45f9c3 100644 --- a/core/archipelago/src/fips/dial.rs +++ b/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 diff --git a/core/archipelago/src/fips/mod.rs b/core/archipelago/src/fips/mod.rs index a717f917..ae192b5a 100644 --- a/core/archipelago/src/fips/mod.rs +++ b/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 = None; + let mut prev_totals = telemetry::totals(); + let mut last_apply: Option = 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 = 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; + } } }); } diff --git a/core/archipelago/src/fips/telemetry.rs b/core/archipelago/src/fips/telemetry.rs index b0ab0a53..1308b9bd 100644 --- a/core/archipelago/src/fips/telemetry.rs +++ b/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); diff --git a/core/archipelago/src/server.rs b/core/archipelago/src/server.rs index b9dbfef4..fbd0047c 100644 --- a/core/archipelago/src/server.rs +++ b/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, 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; } };