250 lines
9.3 KiB
Rust
250 lines
9.3 KiB
Rust
//! Durable per-agent power *intent* (`wanted: Up | Offline`) — the
|
||
//! spec half of spec-vs-status desired-state reconciliation.
|
||
//! `container_view` remains the observed *status*; the job queue's
|
||
//! `Reconcile` nodes are the mechanism that converges the two.
|
||
//!
|
||
//! Stored as the `agent_power` table in the coordinator DB
|
||
//! (`/var/lib/hyperhive/db/broker.sqlite`, one tiny row per agent) —
|
||
//! same one-file-many-modules pattern as `approvals` /
|
||
//! `scheduled_prompts`, each with its own connection. Intent persists
|
||
//! across hive-c0re restarts; in-flight
|
||
//! queue work deliberately does not. Setting `wanted` is never a
|
||
//! queued node: operator/intent actions update the row synchronously
|
||
//! at request time, then submit the DAG whose terminal `Reconcile`
|
||
//! reads the fresh value — rapid toggles are last-writer-wins and the
|
||
//! reconciles converge. Power toggles never commit to the meta repo.
|
||
|
||
use std::path::Path;
|
||
use std::sync::Mutex;
|
||
|
||
use anyhow::{Context, Result};
|
||
use chrono::Utc;
|
||
use rusqlite::{Connection, OptionalExtension, params};
|
||
|
||
const SCHEMA: &str = "
|
||
CREATE TABLE IF NOT EXISTS agent_power (
|
||
agent TEXT PRIMARY KEY,
|
||
wanted TEXT NOT NULL,
|
||
updated_at INTEGER NOT NULL
|
||
);
|
||
";
|
||
|
||
/// Per-agent power intent.
|
||
#[derive(Debug, Clone, Copy, PartialEq, Eq, strum::IntoStaticStr, strum::EnumString)]
|
||
#[strum(serialize_all = "snake_case")]
|
||
pub enum Wanted {
|
||
Up,
|
||
Offline,
|
||
}
|
||
|
||
impl Wanted {
|
||
/// Seed value from an observed running state (first boot after
|
||
/// this store lands, or an agent spawned outside the normal path).
|
||
pub fn from_running(running: bool) -> Self {
|
||
if running { Wanted::Up } else { Wanted::Offline }
|
||
}
|
||
}
|
||
|
||
/// What a `Reconcile` should do given intent + observation. Pure so
|
||
/// the `{Up,Offline} × {up,down}` matrix is unit-testable.
|
||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||
pub enum ReconcileAction {
|
||
Start,
|
||
Stop,
|
||
Noop,
|
||
}
|
||
|
||
#[must_use]
|
||
pub fn reconcile_action(wanted: Wanted, running: bool) -> ReconcileAction {
|
||
match (wanted, running) {
|
||
(Wanted::Up, false) => ReconcileAction::Start,
|
||
(Wanted::Offline, true) => ReconcileAction::Stop,
|
||
(Wanted::Up, true) | (Wanted::Offline, false) => ReconcileAction::Noop,
|
||
}
|
||
}
|
||
|
||
/// Sqlite-backed store. `Arc`-friendly: all methods take `&self`, the
|
||
/// internal `Mutex<Connection>` serializes access.
|
||
pub struct PowerStore {
|
||
conn: Mutex<Connection>,
|
||
}
|
||
|
||
impl PowerStore {
|
||
/// Open (a connection to) the shared coordinator DB and ensure the
|
||
/// `agent_power` table exists. `db_path` is the same sqlite file
|
||
/// the broker / approvals stores open.
|
||
pub fn open(db_path: &Path) -> Result<Self> {
|
||
let conn = crate::db::open(db_path, "agent_power")?;
|
||
conn.execute_batch(SCHEMA)
|
||
.context("apply agent_power schema")?;
|
||
Ok(Self {
|
||
conn: Mutex::new(conn),
|
||
})
|
||
}
|
||
|
||
/// In-memory store for tests.
|
||
#[cfg(test)]
|
||
pub fn open_in_memory() -> Result<Self> {
|
||
let conn = Connection::open_in_memory().context("open in-memory agent_power db")?;
|
||
conn.execute_batch(SCHEMA)
|
||
.context("apply agent_power schema")?;
|
||
Ok(Self {
|
||
conn: Mutex::new(conn),
|
||
})
|
||
}
|
||
|
||
/// Read an agent's intent. `None` when the agent has no row yet
|
||
/// (callers seed from observed state via [`Self::get_or_seed`]).
|
||
///
|
||
/// A row that exists but holds an unparseable value is an **error**, not
|
||
/// `None`. The two used to collapse, and the damage was not that a caller
|
||
/// read the wrong value: [`Self::get_or_seed`] takes `None` to mean "never
|
||
/// seeded" and writes what the container is *currently doing*, so a
|
||
/// corrupted row silently overwrote itself with the observed state this
|
||
/// store exists to overrule.
|
||
pub fn get(&self, agent: &str) -> Result<Option<Wanted>> {
|
||
let conn = self.conn.lock().expect("agent_power mutex poisoned");
|
||
let row: Option<String> = conn
|
||
.query_row(
|
||
"SELECT wanted FROM agent_power WHERE agent = ?1",
|
||
params![agent],
|
||
|r| r.get(0),
|
||
)
|
||
.optional()
|
||
.context("select agent_power")?;
|
||
match row {
|
||
None => Ok(None),
|
||
Some(stored) => stored.parse::<Wanted>().ok().map(Some).with_context(|| {
|
||
format!("agent_power row for {agent} holds an unknown intent {stored:?}")
|
||
}),
|
||
}
|
||
}
|
||
|
||
/// Write an agent's intent (last-writer-wins, synchronous at
|
||
/// request time).
|
||
pub fn set(&self, agent: &str, wanted: Wanted) -> Result<()> {
|
||
let conn = self.conn.lock().expect("agent_power mutex poisoned");
|
||
conn.execute(
|
||
"INSERT INTO agent_power (agent, wanted, updated_at) VALUES (?1, ?2, ?3)
|
||
ON CONFLICT(agent) DO UPDATE SET wanted = ?2, updated_at = ?3",
|
||
params![agent, <&str>::from(wanted), Utc::now().timestamp()],
|
||
)
|
||
.context("upsert agent_power")?;
|
||
Ok(())
|
||
}
|
||
|
||
/// Read an agent's intent, seeding the row from the observed
|
||
/// running state when absent — the migration rule for agents that
|
||
/// predate this store (running ⇒ `Up`, stopped ⇒ `Offline`), after
|
||
/// which the DB is authoritative.
|
||
pub fn get_or_seed(&self, agent: &str, running: bool) -> Result<Wanted> {
|
||
if let Some(w) = self.get(agent)? {
|
||
return Ok(w);
|
||
}
|
||
let seeded = Wanted::from_running(running);
|
||
self.set(agent, seeded)?;
|
||
tracing::info!(%agent, wanted = <&str>::from(seeded), "agent_power: seeded from observed state");
|
||
Ok(seeded)
|
||
}
|
||
|
||
/// Drop an agent's row (container destroyed).
|
||
pub fn remove(&self, agent: &str) -> Result<()> {
|
||
let conn = self.conn.lock().expect("agent_power mutex poisoned");
|
||
conn.execute("DELETE FROM agent_power WHERE agent = ?1", params![agent])
|
||
.context("delete agent_power")?;
|
||
Ok(())
|
||
}
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
|
||
/// The full `{Up,Offline} × {up,down}` reconcile matrix:
|
||
/// start / stop / noop / noop.
|
||
#[test]
|
||
fn reconcile_matrix() {
|
||
assert_eq!(reconcile_action(Wanted::Up, false), ReconcileAction::Start);
|
||
assert_eq!(
|
||
reconcile_action(Wanted::Offline, true),
|
||
ReconcileAction::Stop
|
||
);
|
||
assert_eq!(reconcile_action(Wanted::Up, true), ReconcileAction::Noop);
|
||
assert_eq!(
|
||
reconcile_action(Wanted::Offline, false),
|
||
ReconcileAction::Noop
|
||
);
|
||
}
|
||
|
||
/// Write a value `set` cannot produce, the way corruption would.
|
||
fn write_raw(store: &PowerStore, agent: &str, wanted: &str) {
|
||
let conn = store.conn.lock().expect("agent_power mutex poisoned");
|
||
conn.execute(
|
||
"INSERT INTO agent_power (agent, wanted, updated_at) VALUES (?1, ?2, 0)",
|
||
params![agent, wanted],
|
||
)
|
||
.expect("insert raw row");
|
||
}
|
||
|
||
/// A row that exists but does not parse is an error, not an absent one.
|
||
/// The two rows beside it are the control: this must not pass by breaking
|
||
/// `get` for everything.
|
||
#[test]
|
||
fn an_unparseable_row_is_an_error_not_an_absent_one() {
|
||
let store = PowerStore::open_in_memory().expect("open");
|
||
store.set("valid", Wanted::Up).expect("set");
|
||
write_raw(&store, "corrupt", "sideways");
|
||
|
||
assert_eq!(store.get("valid").expect("valid reads"), Some(Wanted::Up));
|
||
assert_eq!(store.get("missing").expect("absent reads"), None);
|
||
|
||
let err = store
|
||
.get("corrupt")
|
||
.expect_err("corrupt must not read as absent");
|
||
assert!(
|
||
format!("{err:#}").contains("sideways"),
|
||
"the error should name the offending value: {err:#}"
|
||
);
|
||
}
|
||
|
||
/// The damage the error arm exists to prevent: `get_or_seed` treats
|
||
/// `None` as "never seeded" and writes observed state, so a corrupt row
|
||
/// used to overwrite itself with whatever the container was doing.
|
||
#[test]
|
||
fn a_corrupt_row_is_not_reseeded_from_observed_state() {
|
||
let store = PowerStore::open_in_memory().expect("open");
|
||
write_raw(&store, "corrupt", "sideways");
|
||
|
||
assert!(store.get_or_seed("corrupt", true).is_err());
|
||
|
||
let stored: String = store
|
||
.conn
|
||
.lock()
|
||
.expect("agent_power mutex poisoned")
|
||
.query_row(
|
||
"SELECT wanted FROM agent_power WHERE agent = ?1",
|
||
params!["corrupt"],
|
||
|r| r.get(0),
|
||
)
|
||
.expect("row still there");
|
||
assert_eq!(stored, "sideways", "the corrupt row must be left alone");
|
||
}
|
||
|
||
#[test]
|
||
fn get_set_roundtrip_and_seed() {
|
||
let store = PowerStore::open_in_memory().expect("open");
|
||
assert_eq!(store.get("alice").expect("get"), None);
|
||
// Seed from observed running state, once.
|
||
assert_eq!(store.get_or_seed("alice", true).expect("seed"), Wanted::Up);
|
||
// Thereafter the DB is authoritative — observed state no longer
|
||
// overrides.
|
||
assert_eq!(
|
||
store.get_or_seed("alice", false).expect("seeded"),
|
||
Wanted::Up
|
||
);
|
||
store.set("alice", Wanted::Offline).expect("set");
|
||
assert_eq!(store.get("alice").expect("get"), Some(Wanted::Offline));
|
||
store.remove("alice").expect("remove");
|
||
assert_eq!(store.get("alice").expect("get"), None);
|
||
}
|
||
}
|