perf(async): remove blocking std::process::Command from async paths
Every production process spawn reachable from a tokio worker now uses tokio::process: the install path's podman-port probe, the dependencies disk check, factory-reset restart, config host-IP detection, the orchestrator's host-facts helpers (resolve_dynamic_env and its call sites made async to carry it through), and AutoRuntime's podman/docker probes. The FIPS transport probe is the special case: is_available() is a sync trait method called from async route(), so instead of blocking ~50ms on systemctl per stale-cache hit it now serves the cached value and refreshes on a background thread (stale-while-revalidate) — bounded staleness, zero stalled workers. §C of the 1.8.0 hardening plan; container/transport/config/package suites green. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Fable 5
parent
01cbec27ed
commit
4c75bb3d38
@@ -23,26 +23,36 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
||||
/// TTL keeps the result responsive to daemon flaps without pounding DBus.
|
||||
const AVAILABILITY_CACHE_TTL: Duration = Duration::from_secs(10);
|
||||
|
||||
/// Availability cache shared with the background probe thread, so the
|
||||
/// sync `is_available()` hot path never blocks on `systemctl`.
|
||||
struct AvailabilityCache {
|
||||
available: AtomicBool,
|
||||
probed_at_ms: AtomicU64,
|
||||
probe_in_flight: AtomicBool,
|
||||
}
|
||||
|
||||
pub struct FipsTransport {
|
||||
identity_dir: PathBuf,
|
||||
available_cached: AtomicBool,
|
||||
available_cached_at_ms: AtomicU64,
|
||||
availability: std::sync::Arc<AvailabilityCache>,
|
||||
}
|
||||
|
||||
impl FipsTransport {
|
||||
pub fn new(identity_dir: &Path) -> Self {
|
||||
Self {
|
||||
identity_dir: identity_dir.to_path_buf(),
|
||||
available_cached: AtomicBool::new(false),
|
||||
available_cached_at_ms: AtomicU64::new(0),
|
||||
availability: std::sync::Arc::new(AvailabilityCache {
|
||||
available: AtomicBool::new(false),
|
||||
probed_at_ms: AtomicU64::new(0),
|
||||
probe_in_flight: AtomicBool::new(false),
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
fn probe_daemon_active() -> bool {
|
||||
// Cheap blocking probe: spawn `systemctl is-active` synchronously.
|
||||
// Short-circuit if either the archipelago-managed unit or the
|
||||
// upstream fips.service is active — legacy/dev nodes run only the
|
||||
// upstream unit.
|
||||
// Blocking probe — only ever run on a dedicated background thread
|
||||
// (see is_available), never on a tokio worker. Short-circuit if
|
||||
// either the archipelago-managed unit or the upstream fips.service
|
||||
// is active — legacy/dev nodes run only the upstream unit.
|
||||
for unit in [
|
||||
crate::fips::SERVICE_UNIT,
|
||||
crate::fips::UPSTREAM_SERVICE_UNIT,
|
||||
@@ -70,14 +80,30 @@ impl NodeTransport for FipsTransport {
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.map(|d| d.as_millis() as u64)
|
||||
.unwrap_or(0);
|
||||
let cached_at = self.available_cached_at_ms.load(Ordering::Relaxed);
|
||||
let cached_at = self.availability.probed_at_ms.load(Ordering::Relaxed);
|
||||
let cached = self.availability.available.load(Ordering::Relaxed);
|
||||
if now_ms.saturating_sub(cached_at) < AVAILABILITY_CACHE_TTL.as_millis() as u64 {
|
||||
return self.available_cached.load(Ordering::Relaxed);
|
||||
return cached;
|
||||
}
|
||||
let val = Self::probe_daemon_active();
|
||||
self.available_cached.store(val, Ordering::Relaxed);
|
||||
self.available_cached_at_ms.store(now_ms, Ordering::Relaxed);
|
||||
val
|
||||
// Cache is stale. This sync trait method is called from async
|
||||
// route(), so running the ~50ms systemctl probe inline stalls a
|
||||
// tokio worker. Serve the stale value and refresh on a background
|
||||
// thread instead — the transport supervisor's warm loop keeps this
|
||||
// fresh in steady state, so staleness is bounded to one probe round.
|
||||
let cache = std::sync::Arc::clone(&self.availability);
|
||||
if !cache.probe_in_flight.swap(true, Ordering::Relaxed) {
|
||||
std::thread::spawn(move || {
|
||||
let val = Self::probe_daemon_active();
|
||||
let probed_ms = SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.map(|d| d.as_millis() as u64)
|
||||
.unwrap_or(0);
|
||||
cache.available.store(val, Ordering::Relaxed);
|
||||
cache.probed_at_ms.store(probed_ms, Ordering::Relaxed);
|
||||
cache.probe_in_flight.store(false, Ordering::Relaxed);
|
||||
});
|
||||
}
|
||||
cached
|
||||
}
|
||||
|
||||
fn send<'a>(
|
||||
|
||||
Reference in New Issue
Block a user