//! The agent's sync loop: join the controller's document, reconcile on change,
//! and report status back.
//!
//! Flow:
//! 1. [`Node::import_doc`] joins the doc from a ticket and yields a live event
//! stream.
//! 2. On startup and on every relevant remote change, [`reconcile_once`] reads
//! this host's manifest, pulls the referenced file bodies from the local
//! (already-synced) blob store, runs the pure [`crate::reconcile`] engine on
//! a blocking thread, and writes a [`Status`] back under the host's key.
//!
//! We defer a reconcile if a manifest references a file body that hasn't synced
//! yet — the manifest entry can arrive before its content. A later
//! `ContentReady` event will retrigger us, so no change is lost.
use std::collections::HashMap;
use std::io;
use std::path::PathBuf;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use anyhow::{Context, Result};
use devendra_node::{DocHandle, DocTicket, LiveEvent, Node, StreamExt};
use devendra_proto::{
ApplyResult, FileResource, FileState, HostId, HostMetrics, Manifest, PROTOCOL_VERSION,
ServiceHealth, ServiceReport, ServiceRuntime, ServiceState, Status, keys,
};
use tracing::{debug, error, info, warn};
use crate::reconcile::{
ContentSource, RealFs, Reconciler, SystemPackageManager, SystemctlController,
SystemdController, systemd::Scope,
};
/// Everything the agent needs to run.
pub struct AgentConfig {
/// This host's stable identifier; selects which manifest to apply.
pub host_id: HostId,
/// Where to keep the iroh blob/doc stores.
pub data_dir: PathBuf,
/// Join ticket for the controller's document.
pub ticket: DocTicket,
/// systemd management scope (system vs --user).
pub scope: Scope,
/// Whether to undo (prune) files/services that drop out of the desired
/// state — e.g. when a fragment is removed from a host.
pub prune: bool,
}
/// Run the agent until the event stream ends, reconciling on every change.
/// The embeddable entrypoint for a host program that manages its own lifetime.
pub async fn run(config: AgentConfig) -> Result<()> {
run_until(config, std::future::pending::<()>()).await
}
/// Like [`run`], but also stops (and shuts the node down cleanly) when
/// `shutdown` resolves — e.g. a Ctrl-C future. Clean shutdown flushes the blob
/// store so the agent's last status write reaches disk and its peers.
pub async fn run_until(config: AgentConfig, shutdown: impl std::future::Future) -> Result<()> {
let node = Node::spawn(&config.data_dir, None).await?;
info!(node_id = %node.node_id(), host = %config.host_id, "agent node up");
let (doc, mut events) = node.import_doc(config.ticket.clone()).await?;
info!(doc = %doc.id(), "joined controller document, syncing");
// The generation we've already converged to. Gating reconciles on this is
// what keeps the agent quiet at steady state: without it, our own status
// write triggers a ContentReady event, which would trigger another
// reconcile, which writes status again — a busy-loop that floods logs and
// flash. `None` until the first successful (non-deferred) apply.
let mut applied_generation: Option<u64> = None;
// Last live service states we published, so the periodic poll only writes a
// new report when something actually changed.
let mut last_service_report: Option<Vec<ServiceRuntime>> = None;
// Best-effort reconcile on startup — the persistent store may already hold
// a manifest from a previous run before any live event arrives.
if let Err(e) = reconcile_once(
&node,
&doc,
&config,
&mut applied_generation,
&mut last_service_report,
)
.await
{
warn!("startup reconcile failed: {e:#}");
}
// Publish metrics once up front so the badge appears without waiting a beat.
if let Err(e) = report_metrics(&node, &doc, &config).await {
warn!("metrics report failed: {e:#}");
}
// Connection health. A periodic heartbeat nudges a re-sync (recovering from
// a dropped controller connection) and, crucially, makes a *sustained* sync
// break loud in the logs instead of silent — the failure mode where an
// agent quietly stops receiving updates. `last_sync` is refreshed whenever a
// sync round completes with a peer.
let heartbeat_period = Duration::from_secs(
std::env::var("DEVENDRA_HEARTBEAT_SECS")
.ok()
.and_then(|v| v.parse().ok())
.filter(|&s| s > 0)
.unwrap_or(60),
);
let stale_after = heartbeat_period * 3;
let mut heartbeat = tokio::time::interval(heartbeat_period);
heartbeat.tick().await; // consume the immediate first tick
let started = Instant::now();
let mut last_sync: Option<Instant> = None;
tokio::pin!(shutdown);
loop {
tokio::select! {
maybe_event = events.next() => match maybe_event {
Some(Ok(event)) => {
// Any completed sync round is a sign of a live connection.
if matches!(event, LiveEvent::SyncFinished(_)) {
last_sync = Some(Instant::now());
}
if event_touches_host(&event, &config.host_id)
&& let Err(e) = reconcile_once(
&node, &doc, &config,
&mut applied_generation, &mut last_service_report,
).await
{
error!("reconcile failed: {e:#}");
}
}
Some(Err(e)) => error!("event stream error: {e:#}"),
None => {
warn!("event stream ended");
break;
}
},
_ = heartbeat.tick() => {
// Nudge sync; on a reachable controller this drives a fresh
// SyncFinished that refreshes `last_sync`.
if let Err(e) = node.resync(&doc, &config.ticket).await {
warn!("re-sync nudge failed: {e:#}");
}
assess_health(&last_sync, applied_generation, started, stale_after);
// Refresh live service states (writes only on change).
if let Some(manifest) = load_applied_manifest(&config).await
&& let Err(e) = report_service_states(
&node, &doc, &config, &manifest, &mut last_service_report,
).await
{
warn!("service state report failed: {e:#}");
}
// Refresh host metrics (uptime / load average).
if let Err(e) = report_metrics(&node, &doc, &config).await {
warn!("metrics report failed: {e:#}");
}
}
_ = &mut shutdown => {
info!("shutdown requested");
break;
}
}
}
node.shutdown().await?;
info!("agent stopped");
Ok(())
}
/// Emit a health line only when something is wrong: silent (debug) when sync is
/// healthy, WARN when the controller has gone unreachable. Keeps flash-limited
/// devices quiet in steady state while making a sync break obvious.
fn assess_health(
last_sync: &Option<Instant>,
applied_generation: Option<u64>,
started: Instant,
stale_after: Duration,
) {
let synced_recently = last_sync
.map(|t| t.elapsed() < stale_after)
.unwrap_or(false);
if synced_recently {
debug!(generation = ?applied_generation, "sync healthy");
} else if last_sync.is_some() {
warn!(
generation = ?applied_generation,
"no sync with the controller in {}s — it may be unreachable",
stale_after.as_secs()
);
} else if started.elapsed() >= stale_after {
// Never synced at all since startup — almost always a bad ticket, a
// firewalled iroh port, or an offline controller.
warn!(
"no contact with the controller after {}s — check the ticket, network, and that the controller's iroh port is reachable",
started.elapsed().as_secs()
);
} else {
debug!("waiting for initial sync with controller");
}
}
/// Whether an event concerns this host's keyspace (manifest or a file body).
/// Status writes are ours, so we ignore them to avoid self-triggering.
fn event_touches_host(event: &LiveEvent, host: &HostId) -> bool {
let manifest_key = keys::manifest(host);
let files_prefix = keys::files_prefix(host);
match event {
LiveEvent::InsertRemote { entry, .. } => {
let key = entry.key();
key == manifest_key.as_bytes() || key.starts_with(files_prefix.as_bytes())
}
// Content for a previously-deferred file body may now be available.
LiveEvent::ContentReady { .. } | LiveEvent::PendingContentReady => true,
_ => false,
}
}
/// Read the current manifest, gather file bodies, reconcile, and report.
///
/// `applied_generation` tracks the last generation we fully applied; a reconcile
/// whose generation matches it is a no-op and returns immediately, so repeated
/// events (including the ones our own status write produces) don't re-run the
/// reconciler or emit logs.
async fn reconcile_once(
node: &Node,
doc: &DocHandle,
config: &AgentConfig,
applied_generation: &mut Option<u64>,
service_last: &mut Option<Vec<ServiceRuntime>>,
) -> Result<()> {
let manifest_key = keys::manifest(&config.host_id);
let Some(bytes) = node.read(doc, &manifest_key).await? else {
debug!("no manifest published for host yet");
return Ok(());
};
let manifest: Manifest = serde_json::from_slice(&bytes).context("parsing manifest JSON")?;
if manifest.protocol_version != PROTOCOL_VERSION {
anyhow::bail!(
"manifest protocol version {} != agent {}",
manifest.protocol_version,
PROTOCOL_VERSION
);
}
// Gate: nothing to do if we've already applied this generation.
if *applied_generation == Some(manifest.generation) {
return Ok(());
}
// Pull every present file's body from the local blob store. If any is not
// yet synced, defer: a ContentReady event will bring us back (and we leave
// `applied_generation` unchanged so that retrigger isn't gated out).
let mut bodies: HashMap<String, Vec<u8>> = HashMap::new();
for file in &manifest.files {
if file.state != FileState::Present {
continue;
}
let key = keys::file(&config.host_id, &file.path);
match node.read(doc, &key).await? {
Some(body) => {
bodies.insert(file.path.clone(), body);
}
None => {
debug!(path = %file.path, "file body not synced yet; deferring reconcile");
return Ok(());
}
}
}
let generation = manifest.generation;
let scope = config.scope;
let content = MapContentSource(bodies);
// Load the manifest we last applied so the reconciler can prune resources
// that have since dropped out of the desired state (e.g. a removed
// fragment). Absent/unreadable => no prior => no pruning (safe on upgrade).
let applied_path = applied_manifest_path(config);
let prior: Option<Manifest> = if config.prune {
tokio::fs::read(&applied_path)
.await
.ok()
.and_then(|b| serde_json::from_slice(&b).ok())
} else {
None
};
// The reconciler is synchronous and shells out to systemctl; run it off the
// async runtime so it can't stall other tasks.
let apply_manifest = manifest.clone();
let report = tokio::task::spawn_blocking(move || {
let mut fs = RealFs;
let mut systemd = SystemctlController::new(scope);
let mut packages = SystemPackageManager::detect();
Reconciler {
fs: &mut fs,
systemd: &mut systemd,
packages: &mut packages,
content: &content,
}
.apply_with_prior(&apply_manifest, prior.as_ref())
})
.await
.context("reconcile task panicked")?;
// Record what we just applied so the next reconcile can diff against it.
if let Ok(json) = serde_json::to_vec(&manifest)
&& let Err(e) = tokio::fs::write(&applied_path, json).await
{
warn!("could not persist applied manifest: {e}");
}
log_report(generation, &report);
let status = Status {
protocol_version: PROTOCOL_VERSION,
generation: report.generation,
result: report.result,
changes: report.changes,
message: summarize(report.result),
applied_at_ms: now_ms(),
};
let status_json = serde_json::to_vec(&status).context("serializing status")?;
node.write(doc, &keys::status(&config.host_id), status_json)
.await
.context("writing status back to doc")?;
// Mark this generation applied (even on failure) so the status write we just
// made — and any other event at this generation — doesn't re-trigger us.
// A retry requires the controller to publish a new generation.
*applied_generation = Some(report.generation);
// Report the freshly-applied services' live state right away; the heartbeat
// keeps it refreshed afterwards.
if let Err(e) = report_service_states(node, doc, config, &manifest, service_last).await {
warn!("service state report failed: {e:#}");
}
Ok(())
}
fn log_report(generation: u64, report: &crate::reconcile::Report) {
let changed = report
.changes
.iter()
.filter(|c| c.action.is_change())
.count();
match report.result {
// No-change applies happen at most once per generation now (the gate
// stops repeats), and on flash-constrained devices even that is noise —
// keep it at debug so steady state writes nothing.
ApplyResult::Ok => debug!(generation, "reconcile ok (no changes)"),
ApplyResult::Changed => info!(generation, changed, "reconcile applied changes"),
ApplyResult::Failed => {
for c in &report.changes {
if c.action == devendra_proto::ChangeAction::Failed {
error!(target = %c.target, detail = %c.detail, "resource failed");
}
}
warn!(generation, "reconcile completed with failures");
}
}
}
fn summarize(result: ApplyResult) -> String {
match result {
ApplyResult::Ok => "converged; no changes".into(),
ApplyResult::Changed => "converged; changes applied".into(),
ApplyResult::Failed => "one or more resources failed".into(),
}
}
fn now_ms() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0)
}
/// Where the agent records the last manifest it applied, for prune diffing.
fn applied_manifest_path(config: &AgentConfig) -> PathBuf {
config.data_dir.join("applied-manifest.json")
}
/// Load the last-applied manifest from disk (used by the heartbeat to know which
/// services to poll).
async fn load_applied_manifest(config: &AgentConfig) -> Option<Manifest> {
tokio::fs::read(applied_manifest_path(config))
.await
.ok()
.and_then(|b| serde_json::from_slice(&b).ok())
}
/// Sample host uptime and load average from `/proc`. Returns `None` if the
/// files can't be read/parsed (non-Linux, unusual sandbox).
fn collect_host_metrics() -> Option<HostMetrics> {
let uptime_secs = std::fs::read_to_string("/proc/uptime")
.ok()?
.split_whitespace()
.next()?
.parse::<f64>()
.ok()? as u64;
let loadavg = std::fs::read_to_string("/proc/loadavg").ok()?;
let mut fields = loadavg.split_whitespace();
let load1 = fields.next()?.parse().ok()?;
let load5 = fields.next()?.parse().ok()?;
let load15 = fields.next()?.parse().ok()?;
Some(HostMetrics {
protocol_version: PROTOCOL_VERSION,
reported_at_ms: now_ms(),
uptime_secs,
load1,
load5,
load15,
})
}
/// Publish the current host metrics. Small and cheap; written each heartbeat.
async fn report_metrics(node: &Node, doc: &DocHandle, config: &AgentConfig) -> Result<()> {
let Some(metrics) = collect_host_metrics() else {
return Ok(());
};
node.write(
doc,
&keys::metrics(&config.host_id),
serde_json::to_vec(&metrics).context("serializing metrics")?,
)
.await
.context("writing metrics")?;
Ok(())
}
/// Query the live systemd state of every service the manifest wants
/// [`ServiceState::Started`], mapping each to a traffic-light health. Runs
/// synchronously (shells out to systemctl); call via `spawn_blocking`.
fn poll_service_states(manifest: &Manifest, scope: Scope) -> Vec<ServiceRuntime> {
let sd = SystemctlController::new(scope);
manifest
.services
.iter()
.filter(|s| s.state == Some(ServiceState::Started))
.map(|s| {
let (health, detail) = match sd.active_state(&s.name) {
Ok(state) => (ServiceHealth::from_active_state(&state), state),
Err(e) => (ServiceHealth::Unknown, format!("query failed: {e}")),
};
ServiceRuntime {
name: s.name.clone(),
health,
detail,
}
})
.collect()
}
/// Poll monitored services and, only if their states changed since the last
/// report, publish a fresh [`ServiceReport`]. Skipping unchanged writes keeps
/// flash and network quiet.
async fn report_service_states(
node: &Node,
doc: &DocHandle,
config: &AgentConfig,
manifest: &Manifest,
last: &mut Option<Vec<ServiceRuntime>>,
) -> Result<()> {
if !manifest
.services
.iter()
.any(|s| s.state == Some(ServiceState::Started))
{
return Ok(()); // no started services to monitor
}
let scope = config.scope;
let manifest = manifest.clone();
let states = tokio::task::spawn_blocking(move || poll_service_states(&manifest, scope))
.await
.context("service poll task panicked")?;
if last.as_deref() == Some(states.as_slice()) {
return Ok(()); // unchanged — nothing to write
}
let report = ServiceReport {
protocol_version: PROTOCOL_VERSION,
reported_at_ms: now_ms(),
services: states.clone(),
};
node.write(
doc,
&keys::service_report(&config.host_id),
serde_json::to_vec(&report).context("serializing service report")?,
)
.await
.context("writing service report")?;
debug!(count = states.len(), "reported live service states");
*last = Some(states);
Ok(())
}
/// A [`ContentSource`] backed by file bodies pre-fetched from the blob store,
/// keyed by destination path.
struct MapContentSource(HashMap<String, Vec<u8>>);
impl ContentSource for MapContentSource {
fn fetch(&self, resource: &FileResource) -> io::Result<Vec<u8>> {
self.0.get(&resource.path).cloned().ok_or_else(|| {
io::Error::new(
io::ErrorKind::NotFound,
format!("no synced body for {}", resource.path),
)
})
}
}
neiam
/devendra
public · Issues · Pulls · Labels · Forks · Compare · Actions failure · Badges · Packages
⭐
Log in to mark this repository.
19.1 KiB · text
History
d69022d