mara/hyperhive
mara/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

c0re: keep power intent in memory, drop agent_power/kv tables and the stop snapshot

Browse source
This commit is contained in:
damocles 2026-11-07 01:22:18 +01:00 • committed by mara
commit b77fa9edc3
2 changed files with 222 additions and 553 deletions
Download patch file Download diff file Expand all files Collapse all files

111
hive-c0re/src/stores/broker.rs
View file

@ -11,7 +11,7 @@ use chrono::Utc;
use hive_sh4re::inbox::{InboxRow, Message};
use crate::db::Migration;
use rusqlite::{Connection, OptionalExtension, params};
use rusqlite::{Connection, params};
use serde::Serialize;
use tokio::sync::broadcast;
@ -29,10 +29,11 @@ CREATE TABLE IF NOT EXISTS messages (
CREATE INDEX IF NOT EXISTS idx_messages_undelivered
ON messages (recipient, priority DESC, id) WHERE delivered_at IS NULL;
CREATE TABLE IF NOT EXISTS kv (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
-- Tables an earlier build kept here: the running-agents snapshot (`kv`) and
-- the durable power intent (`agent_power`). The swarm owns power intent now,
-- so a file from that build sheds them on its next open.
DROP TABLE IF EXISTS kv;
DROP TABLE IF EXISTS agent_power;
";
/// Capacity of the live event channel. Slow subscribers (e.g. an idle browser)
@ -163,52 +164,6 @@ impl Broker {
self.events.subscribe()
}
/// Set a small persistent key/value pair (upsert). The `kv` table is a
/// general single-value store for state that must survive a hive-c0re
/// restart but isn't worth a dedicated table — currently just the
/// `hivectl start` running-agents snapshot (see
/// `Coordinator::set_last_stopped_running`).
///
/// # Errors
///
/// Returns an error if the sqlite upsert fails.
pub fn kv_set(&self, key: &str, value: &str) -> Result<()> {
let conn = self.conn.lock().unwrap();
conn.execute(
"INSERT INTO kv (key, value) VALUES (?1, ?2)
ON CONFLICT(key) DO UPDATE SET value = excluded.value",
params![key, value],
)?;
Ok(())
}
/// Read a persistent key/value pair. `None` when the key is absent.
///
/// # Errors
///
/// Returns an error if the sqlite query fails.
pub fn kv_get(&self, key: &str) -> Result<Option<String>> {
let conn = self.conn.lock().unwrap();
let value: Option<String> = conn
.query_row("SELECT value FROM kv WHERE key = ?1", params![key], |row| {
row.get(0)
})
.optional()?;
Ok(value)
}
/// Delete a persistent key/value pair. Idempotent — deleting an absent
/// key is a no-op.
///
/// # Errors
///
/// Returns an error if the sqlite delete fails.
pub fn kv_delete(&self, key: &str) -> Result<()> {
let conn = self.conn.lock().unwrap();
conn.execute("DELETE FROM kv WHERE key = ?1", params![key])?;
Ok(())
}
pub fn send(&self, message: &Message) -> Result<()> {
let conn = self.conn.lock().unwrap();
let now = Utc::now().timestamp();
@ -664,6 +619,42 @@ mod tests {
TmpBroker { path, broker }
}
/// A file from the build that kept the running-agents snapshot (`kv`) and
/// the durable power intent (`agent_power`) in it sheds both tables on its
/// next open and keeps its messages.
#[test]
fn opening_an_older_file_drops_the_kv_and_power_tables() {
let first = open_broker();
first.broker.send(&msg("alice", "bob", "kept")).unwrap();
let path = first.path.clone();
{
let conn = Connection::open(&path).unwrap();
conn.execute_batch(
"CREATE TABLE kv (key TEXT PRIMARY KEY, value TEXT NOT NULL);
CREATE TABLE agent_power (
agent TEXT PRIMARY KEY, wanted TEXT NOT NULL, updated_at INTEGER NOT NULL);
INSERT INTO kv VALUES ('last_stopped_running', '[]');
INSERT INTO agent_power VALUES ('alice', 'up', 0);",
)
.unwrap();
}
let reopened = Broker::open(&path).unwrap();
let conn = Connection::open(&path).unwrap();
let left: Vec<String> = conn
.prepare("SELECT name FROM sqlite_master WHERE name IN ('kv', 'agent_power')")
.unwrap()
.query_map([], |r| r.get(0))
.unwrap()
.collect::<Result<_, _>>()
.unwrap();
assert!(left.is_empty(), "left behind: {left:?}");
assert_eq!(
pop_one(&reopened, "bob").map(|d| d.message.body),
Some("kept".to_owned())
);
}
fn msg(from: &str, to: &str, body: &str) -> Message {
Message {
from: hive_types::Ident::parse(from).expect("test sender must be a valid ident"),
@ -681,24 +672,6 @@ mod tests {
batch.pop()
}
#[test]
fn kv_set_get_delete_roundtrip() {
let tb = open_broker();
let b = &tb.broker;
// Absent key -> None.
assert_eq!(b.kv_get("k").unwrap(), None);
// Set then get.
b.kv_set("k", "v1").unwrap();
assert_eq!(b.kv_get("k").unwrap(), Some("v1".to_owned()));
// Upsert overwrites.
b.kv_set("k", "v2").unwrap();
assert_eq!(b.kv_get("k").unwrap(), Some("v2".to_owned()));
// Delete clears; deleting again is a no-op.
b.kv_delete("k").unwrap();
assert_eq!(b.kv_get("k").unwrap(), None);
b.kv_delete("k").unwrap();
}
/// Happy path: send → recv → `ack_turn` drains the in-memory list
/// and marks the row `acked_at IS NOT NULL`. A second recv finds
/// nothing pending (the row stays in the table for vacuum).

238
hive-c0re/src/stores/power.rs
View file

@ -1,33 +1,24 @@
//! 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.
//! Per-agent power *intent* (`wanted: Up | Offline`) this hive holds in
//! memory — what the operator last asked of an agent through `hivectl` or a
//! queued power op, until hive-c0re restarts. `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 `broker`, 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.
//! Nothing here is durable, on purpose: the swarm owns lasting power intent.
//! It declares each agent's state in the hive's wanted bucket (see
//! `crate::workers::wanted`), and that declaration outranks whatever this map
//! holds. After a restart the map is empty, an agent with no entry is seeded
//! from what its container is doing (so nothing starts or stops by itself),
//! and the declaration brings the declared agents to their state.
//!
//! Setting `wanted` is never a queued node's job to *read back*: operator
//! actions update the map 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.
use std::path::Path;
use std::collections::HashMap;
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")]
@ -37,8 +28,8 @@ pub enum Wanted {
}
impl Wanted {
/// Seed value from an observed running state (first boot after
/// this store lands, or an agent spawned outside the normal path).
/// Seed value from an observed running state (an agent no operator or
/// declaration has said anything about since this process started).
pub fn from_running(running: bool) -> Self {
if running { Wanted::Up } else { Wanted::Offline }
}
@ -62,96 +53,56 @@ pub fn reconcile_action(wanted: Wanted, running: bool) -> ReconcileAction {
}
}
/// Sqlite-backed store. `Arc`-friendly: all methods take `&self`, the
/// internal `Mutex<Connection>` serializes access.
/// The in-memory intent map. `Arc`-friendly: all methods take `&self`, the
/// internal `Mutex` serializes access.
#[derive(Default)]
pub struct PowerStore {
conn: Mutex<Connection>,
intents: Mutex<HashMap<String, Wanted>>,
}
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 store opens.
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:?}")
}),
}
/// Read an agent's intent. `None` when nothing has set or seeded one since
/// this process started (callers seed from observed state via
/// [`Self::get_or_seed`]).
pub fn get(&self, agent: &str) -> Option<Wanted> {
self.intents
.lock()
.expect("power intents mutex poisoned")
.get(agent)
.copied()
}
/// 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(())
pub fn set(&self, agent: &str, wanted: Wanted) {
self.intents
.lock()
.expect("power intents mutex poisoned")
.insert(agent.to_owned(), wanted);
}
/// 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)
/// Read an agent's intent, seeding it from the observed running state when
/// absent (running ⇒ `Up`, stopped ⇒ `Offline`), after which the map is
/// authoritative until the process restarts.
pub fn get_or_seed(&self, agent: &str, running: bool) -> Wanted {
*self
.intents
.lock()
.expect("power intents mutex poisoned")
.entry(agent.to_owned())
.or_insert_with(|| {
let seeded = Wanted::from_running(running);
tracing::info!(%agent, wanted = <&str>::from(seeded), "power intent: seeded from observed state");
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(())
/// Drop an agent's intent (container destroyed).
pub fn remove(&self, agent: &str) {
self.intents
.lock()
.expect("power intents mutex poisoned")
.remove(agent);
}
}
@ -175,75 +126,18 @@ mod tests {
);
}
/// 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);
let store = PowerStore::default();
assert_eq!(store.get("alice"), 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
assert_eq!(store.get_or_seed("alice", true), Wanted::Up);
// Thereafter the map 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);
assert_eq!(store.get_or_seed("alice", false), Wanted::Up);
store.set("alice", Wanted::Offline);
assert_eq!(store.get("alice"), Some(Wanted::Offline));
store.remove("alice");
assert_eq!(store.get("alice"), None);
}
}