diff --git a/docs/4-implementation/scheduler/coordinator.md b/docs/4-implementation/scheduler/coordinator.md index 992c5ecec..f38d0c5e4 100644 --- a/docs/4-implementation/scheduler/coordinator.md +++ b/docs/4-implementation/scheduler/coordinator.md @@ -321,6 +321,31 @@ operator sees at a glance how much work the boot triggered. `auto_update::run` seeds agents without an intent from observed state during classification, so a boot with no operator action in this process never finds drift of its own. +### Periodic reconcile tick + +Once a minute (first tick one minute after boot), `workers::reconcile_tick` +looks for an agent that the hive wants up and whose container is down. It reads +the power intent, the container's unit state, the pause marker and +`Coordinator::is_updating`, then queues the same `Reconcile` node the boot power +path uses, through `JobQueue::insert_reconcile_if_idle`. The tick never stops +anything. + +It leaves an agent alone when: + +- its intent is unknown or `Offline`, or its container runs; +- the agent has its pause marker, an update in progress, or a container state + the tick can't read in time; +- any queued or running node targets it (update, create, destroy, power op or + an earlier reconcile). The busy check and the insert share one lock. + +An agent still down on the tick after the tick queued a restart has failed to +come up, so the tick spaces restarts of one agent 1, 2, 4, 8, then 15 ticks +apart (a 15-minute cap). Seeing the agent running, or no longer wanted up, +resets the spacing. Each restart logs `reconcile tick: restarting agent` with +the agent, `observed=stopped`, `wanted=up` and the attempt number, and bumps +`hyperhive.hive.reconcile_tick.restarts`. `crash_watch` keeps raising its +alert; the tick only converges. + ## Meta flake Agents build from the swarm's `internal/meta` flake, which swarm-controller diff --git a/docs/4-implementation/scheduler/observability.md b/docs/4-implementation/scheduler/observability.md index 74e44bf93..c5e2d9e04 100644 --- a/docs/4-implementation/scheduler/observability.md +++ b/docs/4-implementation/scheduler/observability.md @@ -533,14 +533,15 @@ distinguishable from "this hive is gone." Select them with -| Metric | Unit | Kind | Meaning | -| -------------------------------------- | ---- | ------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `process.uptime` | `s` | gauge | seconds since this hive's `hive-c0re` started exporting; a restart reads as a drop to ~0 | -| `hyperhive.hive.degraded` | `1` | gauge | `1` while the hive reports itself unhealthy — the same verdict `/health/ready` gives and the swarm status view shows | -| `hyperhive.hive.warnings` | `1` | gauge | how many warnings are currently raised, split by a `level` attribute (`warn`, `crit`) | -| `hyperhive.hive.warning` | `1` | gauge | `1` per raised warning, with `kind` and `level` attributes; a cleared warning stops reporting | -| `hyperhive.hive.nix_store.utilization` | `1` | gauge | used fraction (0–1) of the filesystem that holds the nix store; no sample when the probe fails | -| `hyperhive.hive.pressure.stall` | `us` | counter | host-wide `/proc/pressure/{cpu,memory,io}` `total` in microseconds, split by `resource` (`cpu`, `memory`, `io`) and `kind` (`some`, `full`); a rate is the stalled share of time; no sample without PSI | +| Metric | Unit | Kind | Meaning | +| ---------------------------------------- | ----------- | ------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `process.uptime` | `s` | gauge | seconds since this hive's `hive-c0re` started exporting; a restart reads as a drop to ~0 | +| `hyperhive.hive.degraded` | `1` | gauge | `1` while the hive reports itself unhealthy — the same verdict `/health/ready` gives and the swarm status view shows | +| `hyperhive.hive.warnings` | `1` | gauge | how many warnings are currently raised, split by a `level` attribute (`warn`, `crit`) | +| `hyperhive.hive.warning` | `1` | gauge | `1` per raised warning, with `kind` and `level` attributes; a cleared warning stops reporting | +| `hyperhive.hive.nix_store.utilization` | `1` | gauge | used fraction (0–1) of the filesystem that holds the nix store; no sample when the probe fails | +| `hyperhive.hive.pressure.stall` | `us` | counter | host-wide `/proc/pressure/{cpu,memory,io}` `total` in microseconds, split by `resource` (`cpu`, `memory`, `io`) and `kind` (`some`, `full`); a rate is the stalled share of time; no sample without PSI | +| `hyperhive.hive.reconcile_tick.restarts` | `{restart}` | counter | restarts the periodic reconcile tick has queued since `hive-c0re` started, for agents declared up and found down; the agent is in the `reconcile tick: restarting agent` log line | hive-c0re reports both levels of `hyperhive.hive.warnings` every cycle, `0` included, so a healthy hive is visible as zeros rather than as missing diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 92d76661b..75b2ad21b 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -224,6 +224,36 @@ impl JobQueue { Ok(Some(named)) } + /// Queue a lone `Reconcile` for `agent` unless any queued or running node + /// already targets it. + /// + /// `None` means the agent is busy: an update, create, destroy, power op or + /// earlier reconcile is live, and the work in flight converges it (or + /// decides not to). The check and the insert share one lock, so a caller + /// racing another insert cannot queue behind a node it did not see. The + /// shape is the down-agent restart: a `Reconcile` on the agent's lease, + /// which reads the agent's power intent when it runs. + /// + /// # Errors + /// Propagates a graph-insert error. + pub fn insert_reconcile_if_idle(&self, agent: &str) -> anyhow::Result>> { + let mut inner = self.lock(); + let busy = inner + .graph() + .nodes() + .any(|n| !n.state.is_terminal() && n.payload.agent() == agent); + if busy { + return Ok(None); + } + let targets = [(agent.to_owned(), false)]; + let named = inner + .insert_job(None, |b| power::restart_nodes(b, &targets, false)) + .map_err(|e| anyhow::anyhow!("job_queue: graph insert failed: {e}"))?; + drop(inner); + self.notify.notify_one(); + Ok(Some(named)) + } + /// The scheduler itself, for `hive_jobq`'s run-loop seam /// (`Scheduler::claim_next`), which takes exactly this type. /// diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index e9c43a080..489476bdc 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -1556,6 +1556,62 @@ fn a_rebuild_is_live_for_its_own_agent_only() { assert!(!q.rebuild_live("bob")); } +/// The periodic reconcile tick queues one lone `Reconcile` on the agent's lease +/// and nothing for an agent any live node already targets — whichever kind +/// of work it is — while other agents and finished work do not block it. +#[test] +fn an_idle_agent_gets_one_reconcile_and_a_busy_one_none() { + let q = JobQueue::new(1); + let first = q + .insert_reconcile_if_idle("alice") + .expect("insert") + .expect("control: an idle agent gets a reconcile"); + assert_eq!( + declared_shape(&q) + .iter() + .filter(|d| d.kind == "reconcile") + .count(), + 1, + "exactly one Reconcile node" + ); + assert_eq!( + declared_resources_of_kind(&q, "reconcile"), + vec![vec![Resource::Agent("alice".to_owned())]], + "it holds the agent's lease" + ); + + assert!( + q.insert_reconcile_if_idle("alice") + .expect("insert") + .is_none(), + "the live reconcile blocks another" + ); + assert!( + q.insert_reconcile_if_idle("bob").expect("insert").is_some(), + "control: another agent is independent" + ); + + // Work that is not a reconcile blocks too. + insert(&q, |b| { + let _ = rebuild(b, "carol"); + }); + assert!( + q.insert_reconcile_if_idle("carol") + .expect("insert") + .is_none(), + "a rebuild in flight blocks it" + ); + + // A finished reconcile does not. + assert!(q.cancel(first[0].get()), "cancel the queued reconcile"); + assert!( + q.insert_reconcile_if_idle("alice") + .expect("insert") + .is_some(), + "a terminal node does not count as busy" + ); +} + /// Agents whose running closure differs from the declaration are rebuilt in one /// sweep. Each rebuild's window holds a build slot, so `buildSlots` caps how /// many swap at once, however many agents the sweep rebuilds. diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index a3d1e7a37..41cc4522a 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -342,6 +342,9 @@ async fn cmd_serve( // previously-running container goes away without an operator-initiated // transient state — a swarm-level grafana alert matches that line. crash_watch::spawn(coord.clone()); + // Periodic reconcile: queues a start for an agent that is declared up and + // found down. The crash alert above keeps firing; this only converges. + workers::reconcile_tick::spawn(coord.clone()); // Agent-sockets marker poll: re-fires `agent_sockets::write` // and `gateway_nginx::write` every 10s so the JSON and nginx // config pick up newly-bound `.bound` markers after a rebuild. diff --git a/hive-c0re/src/stats/otel_metrics.rs b/hive-c0re/src/stats/otel_metrics.rs index a08ed8a4b..e8d9b09f7 100644 --- a/hive-c0re/src/stats/otel_metrics.rs +++ b/hive-c0re/src/stats/otel_metrics.rs @@ -27,6 +27,7 @@ //! `agent` attribute**, and that absence is the point: it is what makes them //! selectable as `{hive!="", agent=""}`. +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex, OnceLock}; use std::time::{Duration, Instant}; @@ -58,6 +59,14 @@ static PROVIDER: OnceLock = OnceLock::new(); /// doc says exporter-start rather than claiming a precision it does not have. static STARTED: OnceLock = OnceLock::new(); +/// Restarts the periodic reconcile tick has queued since this process started. +static RECONCILE_RESTARTS: AtomicU64 = AtomicU64::new(0); + +/// Count one restart queued by the periodic reconcile tick. +pub fn count_reconcile_restart() { + RECONCILE_RESTARTS.fetch_add(1, Ordering::Relaxed); +} + /// Spawn the container-resource OTEL exporter if OTEL is configured /// (`OTEL_EXPORTER_OTLP_ENDPOINT` non-empty). No-op otherwise. Call once at /// startup. @@ -497,6 +506,17 @@ fn register_hive_instruments(meter: &opentelemetry::metrics::Meter) { }) .build(); + // Restarts the periodic reconcile tick queued for an agent that was wanted + // up and found down. Hive-scoped: the agent is in the log line, and a + // per-agent label would grow a series for every agent that ever failed. + meter + .u64_observable_counter("hyperhive.hive.reconcile_tick.restarts") + .with_unit("{restart}") + .with_callback(|obs| { + obs.observe(RECONCILE_RESTARTS.load(Ordering::Relaxed), HIVE_ATTRS); + }) + .build(); + // Host-wide stall totals from `/proc/pressure`, by `resource` (`cpu`, // `memory`, `io`) and `kind` (`some`, `full`). Counters in microseconds, // so a rate over them is the stalled fraction of wall time. No sample on a diff --git a/hive-c0re/src/workers/mod.rs b/hive-c0re/src/workers/mod.rs index cf839e7fc..003d4a387 100644 --- a/hive-c0re/src/workers/mod.rs +++ b/hive-c0re/src/workers/mod.rs @@ -1,5 +1,5 @@ //! Background tasks and periodic sweeps: crash/login watcher, boot-time -//! auto-update reconcile, the agent-sockets.json writer loop, the MCP socket listener +//! auto-update reconcile, the periodic declared ⇒ running reconcile tick, the agent-sockets.json writer loop, the MCP socket listener //! reconcile loop, knowledge-repo sync, the swarm wanted-state pull, and the //! swarm resource-cap and permissions reads, and the squid users file. //! Each submodule is re-exported @@ -11,6 +11,7 @@ pub mod crash_watch; pub mod knowledge; pub mod mcp_sockets; pub mod permissions_import; +pub mod reconcile_tick; pub mod squid_allow; pub mod squid_deny; pub mod squid_users; diff --git a/hive-c0re/src/workers/reconcile_tick.rs b/hive-c0re/src/workers/reconcile_tick.rs new file mode 100644 index 000000000..f0362688f --- /dev/null +++ b/hive-c0re/src/workers/reconcile_tick.rs @@ -0,0 +1,371 @@ +//! Periodic declared ⇒ running convergence: an agent whose power intent is +//! `Up` and whose container is down gets a `Reconcile` queued, once a +//! [`TICK_INTERVAL`], without anyone having to ask. +//! +//! The tick queues the same `NodeKind::Reconcile` the boot sweep, a watch +//! attach and a declaration write end in, so what it starts, and how, is +//! `run_reconcile`'s decision. It only ever converges *toward a start*: an +//! agent wanted offline is left alone, and a running one is never touched. +//! `crash_watch` keeps raising its alert; this worker does not replace it. +//! +//! # What the tick leaves alone +//! +//! - an agent with no known intent, or one wanted offline (the operator's stop +//! stands); +//! - a paused agent, and one whose container state could not be read in time; +//! - an agent any queued or running node already targets — an update, create, +//! destroy, power op or earlier reconcile (see +//! [`JobQueue::insert_reconcile_if_idle`]). +//! +//! # Backoff +//! +//! An agent that is still down on the tick after a restart was queued for it +//! has failed to come up. Restarts of one agent are spaced `1, 2, 4, …` ticks +//! apart, capped at [`MAX_BACKOFF`]. Seeing the agent running, or no longer +//! wanted up, forgets its history. + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; + +use crate::coordinator::Coordinator; +use crate::lifecycle::{self, AGENT_PREFIX}; +use crate::power::Wanted; + +/// How often declared intent is compared with the observed containers. The +/// first tick fires one interval after boot: the boot sweep owns startup. +const TICK_INTERVAL: Duration = Duration::from_mins(1); + +/// The longest wait between two restarts of one agent. +const MAX_BACKOFF: Duration = Duration::from_mins(15); + +/// What the tick can see of one agent. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct Observation { + /// The power intent this process holds; `None` before anything set or + /// seeded one. + wanted: Option, + /// Whether the container's unit is active; `None` when `systemctl` did not + /// answer in time. + running: Option, + paused: bool, + /// An update has the agent down and not yet back up. + updating: bool, +} + +/// Why the tick did nothing for an agent. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Leave { + /// Running, or wanted offline, or no intent: nothing to converge. + Converged, + /// Paused, updating, or the container state is unreadable. + Held, + /// A restart was queued recently and the agent has not come up; waiting + /// out the backoff. + BackingOff, +} + +/// What one tick does for one agent. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Step { + Leave(Leave), + /// Queue a `Reconcile` (unless the agent turns out to be busy). + Restart, +} + +/// Per-agent restart history, in ticks. +#[derive(Debug, Default)] +struct Backoffs { + agents: HashMap, +} + +#[derive(Debug, Default)] +struct Backoff { + /// Restarts queued since the agent was last seen converged. + attempts: u32, + /// Ticks still to skip before the next restart. + skip: u32, +} + +impl Backoffs { + /// Forget `agent`: it is running, or no longer wanted up. + fn settle(&mut self, agent: &str) { + self.agents.remove(agent); + } + + /// Whether a restart may be queued for `agent` this tick. A refusal uses + /// up one skipped tick. + fn ready(&mut self, agent: &str) -> bool { + match self.agents.get_mut(agent) { + Some(b) if b.skip > 0 => { + b.skip -= 1; + false + } + _ => true, + } + } + + /// Record that a restart was queued; returns how many this makes. + fn record_attempt(&mut self, agent: &str) -> u32 { + let b = self.agents.entry(agent.to_owned()).or_default(); + b.attempts = b.attempts.saturating_add(1); + b.skip = ticks_between(b.attempts).saturating_sub(1); + b.attempts + } +} + +/// Ticks between the `attempts`-th restart and the next: `2^(attempts-1)`, +/// capped at [`MAX_BACKOFF`]. +fn ticks_between(attempts: u32) -> u32 { + let cap = u32::try_from(MAX_BACKOFF.as_secs() / TICK_INTERVAL.as_secs()).unwrap_or(u32::MAX); + 1_u32 + .checked_shl(attempts.saturating_sub(1)) + .unwrap_or(u32::MAX) + .min(cap) +} + +/// The decision for one agent. Pure but for the backoff counters it advances. +fn step(backoffs: &mut Backoffs, agent: &str, seen: Observation) -> Step { + if seen.running == Some(true) || seen.wanted != Some(Wanted::Up) { + backoffs.settle(agent); + return Step::Leave(Leave::Converged); + } + if seen.running.is_none() || seen.paused || seen.updating { + return Step::Leave(Leave::Held); + } + if !backoffs.ready(agent) { + return Step::Leave(Leave::BackingOff); + } + Step::Restart +} + +/// Run the tick until shutdown. +pub fn spawn(coord: Arc) { + let mut shutdown = coord.shutdown_rx(); + tokio::spawn(async move { + let mut backoffs = Backoffs::default(); + loop { + tokio::select! { + () = tokio::time::sleep(TICK_INTERVAL) => {} + _ = shutdown.changed() => { + tracing::info!("reconcile tick: shutdown signal received"); + return; + } + } + tick(&coord, &mut backoffs).await; + } + }); +} + +/// One pass over every agent container. +async fn tick(coord: &Arc, backoffs: &mut Backoffs) { + let containers = match lifecycle::list().await { + Ok(c) => c, + Err(e) => { + tracing::warn!(error = ?e, "reconcile tick: container list unreadable — skipping this tick"); + return; + } + }; + for container in &containers { + let Some(name) = container.strip_prefix(AGENT_PREFIX) else { + continue; + }; + let seen = Observation { + wanted: coord.power.get(name), + running: lifecycle::running_state(name).await, + paused: hive_types::Ident::parse(name).is_ok_and(|id| Coordinator::is_paused(&id)), + updating: coord.is_updating(name), + }; + if step(backoffs, name, seen) != Step::Restart { + continue; + } + match coord.job_queue.insert_reconcile_if_idle(name) { + Ok(Some(_)) => { + let attempt = backoffs.record_attempt(name); + tracing::info!( + agent = %name, + observed = "stopped", + wanted = "up", + attempt, + "reconcile tick: restarting agent" + ); + crate::otel_metrics::count_reconcile_restart(); + coord.emit_rebuild_queue_snapshot(); + } + // Work targeting the agent is in flight; it converges the agent or + // decides not to. + Ok(None) => { + tracing::debug!(agent = %name, "reconcile tick: agent is busy; leaving it"); + } + Err(e) => { + tracing::warn!(agent = %name, error = ?e, "reconcile tick: queueing the reconcile failed"); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// A wanted-up agent whose container is down, nothing holding it back. + const DOWN: Observation = Observation { + wanted: Some(Wanted::Up), + running: Some(false), + paused: false, + updating: false, + }; + + #[test] + fn a_wanted_up_stopped_agent_is_restarted() { + assert_eq!(step(&mut Backoffs::default(), "a", DOWN), Step::Restart); + } + + /// Each deviation from [`DOWN`] that must leave the agent alone, against + /// the control above that the same agent is restarted without it. + #[test] + fn the_tick_leaves_alone_what_is_not_a_plain_down_agent() { + let cases = [ + ( + "wanted offline", + Observation { + wanted: Some(Wanted::Offline), + ..DOWN + }, + Leave::Converged, + ), + ( + "no intent yet", + Observation { + wanted: None, + ..DOWN + }, + Leave::Converged, + ), + ( + "already running", + Observation { + running: Some(true), + ..DOWN + }, + Leave::Converged, + ), + ( + "paused", + Observation { + paused: true, + ..DOWN + }, + Leave::Held, + ), + ( + "mid-update", + Observation { + updating: true, + ..DOWN + }, + Leave::Held, + ), + ( + "container state unreadable", + Observation { + running: None, + ..DOWN + }, + Leave::Held, + ), + ]; + for (what, seen, why) in cases { + assert_eq!( + step(&mut Backoffs::default(), "a", seen), + Step::Leave(why), + "{what}" + ); + } + } + + /// Ticks between restarts of an agent that never comes up: 1, 2, 4, 8, + /// then the 15-tick cap, and it stays there. + #[test] + fn restarts_of_a_failing_agent_back_off_to_the_cap() { + let mut backoffs = Backoffs::default(); + let mut gaps = Vec::new(); + let mut since = 0_u32; + for _ in 0..80 { + since += 1; + if step(&mut backoffs, "a", DOWN) == Step::Restart { + backoffs.record_attempt("a"); + gaps.push(since); + since = 0; + } + } + assert_eq!(gaps[..6], [1, 1, 2, 4, 8, 15]); + assert!(gaps[5..].iter().all(|g| *g == 15), "{gaps:?}"); + } + + #[test] + fn the_wait_is_a_backoff_not_a_refusal() { + let mut backoffs = Backoffs::default(); + backoffs.record_attempt("a"); + backoffs.record_attempt("a"); + assert_eq!( + step(&mut backoffs, "a", DOWN), + Step::Leave(Leave::BackingOff) + ); + assert_eq!(step(&mut backoffs, "a", DOWN), Step::Restart); + // Another agent's history is not this one's. + assert_eq!(step(&mut backoffs, "b", DOWN), Step::Restart); + } + + #[test] + fn seeing_the_agent_running_resets_its_backoff() { + let mut backoffs = Backoffs::default(); + for _ in 0..4 { + backoffs.record_attempt("a"); + } + assert_eq!( + step(&mut backoffs, "a", DOWN), + Step::Leave(Leave::BackingOff), + "control: the history holds the agent back" + ); + let up = Observation { + running: Some(true), + ..DOWN + }; + assert_eq!(step(&mut backoffs, "a", up), Step::Leave(Leave::Converged)); + assert_eq!(step(&mut backoffs, "a", DOWN), Step::Restart); + assert_eq!(backoffs.record_attempt("a"), 1, "the count starts over"); + } + + #[test] + fn an_operator_stop_resets_the_backoff() { + let mut backoffs = Backoffs::default(); + for _ in 0..4 { + backoffs.record_attempt("a"); + } + let stopped = Observation { + wanted: Some(Wanted::Offline), + ..DOWN + }; + step(&mut backoffs, "a", stopped); + assert_eq!(step(&mut backoffs, "a", DOWN), Step::Restart); + } + + /// A refused restart (the agent turned out busy) does not count as an + /// attempt, so it does not lengthen the next wait. + #[test] + fn only_a_recorded_attempt_lengthens_the_backoff() { + let mut backoffs = Backoffs::default(); + assert_eq!(step(&mut backoffs, "a", DOWN), Step::Restart); + assert_eq!(step(&mut backoffs, "a", DOWN), Step::Restart); + assert_eq!(backoffs.record_attempt("a"), 1); + } + + #[test] + fn the_cap_is_reached_without_overflow() { + assert_eq!(ticks_between(1), 1); + assert_eq!(ticks_between(4), 8); + assert_eq!(ticks_between(5), 15); + assert_eq!(ticks_between(u32::MAX), 15); + } +}