From f7d84f4847304f278f9a18c323d79fc9f09a3c47 Mon Sep 17 00:00:00 2001 From: damocles Date: Tue, 23 Jun 2026 21:34:49 +0200 Subject: [PATCH] fix(#1957): persist hivectl-start running-agents snapshot in the broker kv table --- hive-c0re/src/broker.rs | 69 ++++++++++++++++++++++++++++++++++++ hive-c0re/src/coordinator.rs | 46 +++++++++++++++++++++--- 2 files changed, 110 insertions(+), 5 deletions(-) diff --git a/hive-c0re/src/broker.rs b/hive-c0re/src/broker.rs index faf98ca8..3f28787b 100644 --- a/hive-c0re/src/broker.rs +++ b/hive-c0re/src/broker.rs @@ -36,6 +36,11 @@ CREATE TABLE IF NOT EXISTS reminders ( ); CREATE INDEX IF NOT EXISTS idx_reminders_due ON reminders (agent, due_at) WHERE sent_at IS NULL; + +CREATE TABLE IF NOT EXISTS kv ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL +); "; /// Capacity of the live event channel. Slow subscribers (e.g. an idle browser) @@ -197,6 +202,52 @@ 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> { + let conn = self.conn.lock().unwrap(); + let value: Option = 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 = now_unix(); @@ -1216,6 +1267,24 @@ 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). diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index f8670bb8..6fd96eec 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -22,6 +22,12 @@ use crate::socket_server::{self, AgentSocket}; /// fine, the seq dedupe makes a reconnect resync safe. const DASHBOARD_CHANNEL: usize = 256; +/// Broker `kv` key under which the broad-stop running-agents snapshot is +/// persisted, so a `hivectl start` after a hive-c0re restart can still +/// restore only the previously-running agents. See +/// `Coordinator::set_last_stopped_running`. +const LAST_STOPPED_RUNNING_KEY: &str = "last_stopped_running"; + const AGENT_RUNTIME_ROOT: &str = "/run/hyperhive/agents"; /// Manager-editable per-agent config repos. Bind-mounted RW into the manager /// container as `/agents//`. Hive-c0re only writes to these on first @@ -137,8 +143,10 @@ pub struct Coordinator { /// every configured container, so agents the operator intentionally /// left stopped stay stopped. `None` when no broad stop has happened /// since the last start (or since daemon boot) — start then falls - /// back to "start all". In-daemon memory only (hive-c0re survives - /// `hivectl stop`); host-reboot persistence is a separate follow-up. + /// back to "start all". Mirrored to the broker `kv` table + /// (`last_stopped_running` key) so the snapshot also survives a + /// hive-c0re restart between the stop-all and the start: this + /// in-memory copy is the fast path, the persisted copy the backstop. last_stopped_running: Mutex>>, /// Unified wire-facing event channel feeding the dashboard SSE /// stream. Carries broker messages (mirrored from `broker.subscribe` @@ -1185,16 +1193,44 @@ impl Coordinator { /// Record the set of agents that were running at a broad-scope /// `hivectl stop`, so the next broad-scope `start` restores exactly - /// this set. See the `last_stopped_running` field doc. + /// this set. See the `last_stopped_running` field doc. Persists a + /// copy to the broker `kv` table (best-effort) so the snapshot + /// survives a hive-c0re restart; the in-memory copy is the fast path. pub fn set_last_stopped_running(&self, agents: Vec) { + match serde_json::to_string(&agents) { + Ok(json) => { + if let Err(e) = self.broker.kv_set(LAST_STOPPED_RUNNING_KEY, &json) { + tracing::warn!(error = ?e, "persist last_stopped_running failed (in-memory copy still set)"); + } + } + Err(e) => tracing::warn!(error = ?e, "serialise last_stopped_running failed"), + } *self.last_stopped_running.lock().unwrap() = Some(agents); } /// Take (and clear) the recorded broad-stop running set, if any. A /// broad-scope `start` uses this to restore only the previously - /// running agents; `None` means "no record — start all". + /// running agents; `None` means "no record — start all". Falls back + /// to the persisted `kv` copy when the in-memory snapshot is empty + /// (hive-c0re restarted between the stop-all and the start). Clears + /// the persisted copy either way — the snapshot is one-shot. pub fn take_last_stopped_running(&self) -> Option> { - self.last_stopped_running.lock().unwrap().take() + let result = self + .last_stopped_running + .lock() + .unwrap() + .take() + .or_else(|| { + self.broker + .kv_get(LAST_STOPPED_RUNNING_KEY) + .ok() + .flatten() + .and_then(|json| serde_json::from_str::>(&json).ok()) + }); + if let Err(e) = self.broker.kv_delete(LAST_STOPPED_RUNNING_KEY) { + tracing::warn!(error = ?e, "clear persisted last_stopped_running failed"); + } + result } /// Set of agents whose transient was cleared within the last