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

remove what no longer has a reader: build-log store, dashboard sent/delivered events, graph resource dep

Browse source
This commit is contained in:
atlas 2026-11-13 10:08:19 +01:00 • committed by mara
commit 49266e4410
3 changed files with 204 additions and 1054 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -12,7 +12,6 @@ use hive_sh4re::inbox::{InboxRow, Message};
use crate::db::Migration;
use rusqlite::{Connection, params};
use serde::Serialize;
use tokio::sync::broadcast;
const SCHEMA: &str = r"
@ -52,30 +51,11 @@ pub struct Delivery {
pub message: Message,
}
/// Intra-process broker event. `recv_blocking_batch` listens on the
/// same channel as the dashboard forwarder; the forwarder re-emits
/// each event as a `DashboardEvent`.
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "snake_case", tag = "kind")]
/// Intra-process broker event: a message was queued for `to`. Wakes
/// `recv_blocking_batch`, which then re-polls sqlite (the source of truth).
#[derive(Debug, Clone)]
pub enum MessageEvent {
Sent {
/// Broker row id — used by the dashboard to track thread parents.
id: i64,
from: String,
to: String,
body: String,
at: i64,
in_reply_to: Option<i64>,
},
Delivered {
/// Broker row id — used by the dashboard to track thread parents.
id: i64,
from: String,
to: String,
body: String,
at: i64,
in_reply_to: Option<i64>,
},
Sent { to: String },
}
/// Per-recipient in-memory bookkeeping for the deliver-then-ack
@ -182,15 +162,9 @@ impl Broker {
priority
],
)?;
let row_id = conn.last_insert_rowid();
drop(conn);
let _ = self.events.send(MessageEvent::Sent {
id: row_id,
from: message.from.to_string(),
to: message.to.clone(),
body: message.body.clone(),
at: now,
in_reply_to: message.in_reply_to,
});
Ok(())
}
@ -322,10 +296,7 @@ impl Broker {
/// pushed onto the per-recipient `unacked_ids` list (so the next
/// `ack_turn` closes them out), and tagged with
/// `redelivered = true` if it was resurfaced by the most recent
/// `requeue_inflight`. Emits one `MessageEvent::Delivered` per
/// popped row so the dashboard forwarder stream sees one event
/// per message regardless of batch size.
///
/// `requeue_inflight`. ///
/// `max == 0` short-circuits to an empty vec (no DB hit); any
/// positive value caps the batch at `max`. Ordering: higher
/// priority first; FIFO within the same priority level. Operator
@ -404,19 +375,6 @@ impl Broker {
});
}
drop(inflight);
// Mirror the per-row Delivered emit `recv` does so the
// dashboard forwarder sees one event per message regardless of
// which surface the harness used.
for d in &deliveries {
let _ = self.events.send(MessageEvent::Delivered {
id: d.id,
from: d.message.from.to_string(),
to: d.message.to.clone(),
body: d.message.body.clone(),
at: now,
in_reply_to: d.message.in_reply_to,
});
}
Ok(deliveries)
}

478
hive-c0re/src/stores/build_logs.rs
View file

@ -1,478 +0,0 @@
//! Sqlite-backed full build-log capture — stdout + stderr per
//! `nixos-container` / `nix build` invocation, accumulated live.
//! Schema, indices, retention, and the rationale for replacing
//! the old ring buffer: `docs/4-implementation/agent-state/persistence.md::/var/lib/hyperhive/db/build_logs.sqlite`.
use std::path::Path;
use std::sync::{Arc, Mutex, OnceLock};
use anyhow::{Context, Result};
use chrono::Utc;
use rusqlite::{Connection, params};
use serde::Serialize;
use crate::db::Migration;
/// Process-singleton handle, set once at coordinator startup. Lets
/// the `lifecycle` module's `run` access the
/// writer without threading an `Arc<BuildLogs>` through every
/// `pub async fn` entry point in the lifecycle surface — there are
/// 10+ callsites and the handle is the same `Arc` everywhere
/// anyway. Set by `Coordinator::open`.
static GLOBAL: OnceLock<Arc<BuildLogs>> = OnceLock::new();
/// Install the process-wide `BuildLogs` handle. Idempotent: a second
/// call (e.g. test harness setup) silently keeps the first handle.
pub fn install(handle: Arc<BuildLogs>) {
let _ = GLOBAL.set(handle);
}
/// Fetch the process-wide handle, or `None` if `install` hasn't run
/// yet (e.g. early in startup before `Coordinator::open`, or in unit
/// tests that don't bother with the global). Callers must gracefully
/// no-op when this returns `None`.
#[must_use]
pub fn global() -> Option<Arc<BuildLogs>> {
GLOBAL.get().cloned()
}
/// Retain failed-build rows for 30 days — they're what the operator
/// needs to investigate when diagnosing a regression.
const KEEP_FAIL_SECS: i64 = 30 * 24 * 3600;
/// Retain successful-build rows for 24 hours — useful for diffing
/// what changed across a recent rebuild, but past a day the log is
/// noise. In-progress rows are never reaped (they have
/// `finished_at IS NULL`).
const KEEP_OK_SECS: i64 = 24 * 3600;
/// Last line of a row `reap_orphans` closed.
const ORPHAN_NOTE: &str = "hive-c0re stopped before this build finished";
const SCHEMA: &str = "
CREATE TABLE IF NOT EXISTS build_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
agent TEXT NOT NULL,
kind TEXT NOT NULL,
cmdline TEXT NOT NULL,
started_at INTEGER NOT NULL,
finished_at INTEGER,
status TEXT,
stdout TEXT NOT NULL DEFAULT '',
stderr TEXT NOT NULL DEFAULT ''
);
CREATE INDEX IF NOT EXISTS idx_build_logs_agent_started
ON build_logs (agent, started_at DESC);
CREATE INDEX IF NOT EXISTS idx_build_logs_status_finished
ON build_logs (status, finished_at)
WHERE finished_at IS NOT NULL;
";
/// Ordered schema migrations tracked in `schema_versions` (key `"build_logs"`).
///
/// v1 makes the log row carry its node, replacing the host-side
/// `NodeId -> build_log_id` map the job queue used to hold. The link has a
/// single home again, and the direction is the one the type system allows:
/// a `hive_jobq` node payload is immutable after insert, but the log row is
/// written when the build starts and can name the node it belongs to.
///
/// Legacy rows keep `node_id IS NULL` — they predate the column and no node
/// still exists to link them to, so the dashboard's by-node lookup simply
/// misses them (the by-agent listing, which is how they're reached, is
/// unaffected).
const MIGRATIONS: &[Migration] = &[Migration {
sql: "BEGIN;
ALTER TABLE build_logs ADD COLUMN node_id INTEGER;
CREATE INDEX IF NOT EXISTS idx_build_logs_node
ON build_logs (node_id) WHERE node_id IS NOT NULL;
COMMIT;",
adds_column: Some(("build_logs", "node_id")),
}];
/// Status of a finished build attempt. Stored as the literal string in
/// the `status` column; `NULL` while the attempt is still in progress.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, strum::IntoStaticStr)]
#[serde(rename_all = "snake_case")]
#[strum(serialize_all = "snake_case")]
pub enum BuildStatus {
/// Child exited with success.
Ok,
/// Child exited non-zero (build / eval failure).
Fail,
}
/// A stored row with its captured text, as read back by tests.
#[cfg(test)]
#[derive(Debug, Clone)]
pub struct BuildLogFull {
pub agent: String,
pub kind: String,
pub status: Option<String>,
pub finished_at: Option<i64>,
pub stdout: String,
pub stderr: String,
}
/// Sqlite-backed build-log store. `Arc<BuildLogs>`-friendly: all
/// methods take `&self`, internal `Mutex<Connection>` serializes
/// access.
pub struct BuildLogs {
conn: Mutex<Connection>,
}
impl BuildLogs {
pub fn open(db_dir: &Path) -> Result<Self> {
let path = db_dir.join("build_logs.sqlite");
let conn = crate::db::open(&path, "build_logs")?;
conn.execute_batch(SCHEMA)
.context("apply build_logs schema")?;
crate::db::apply_versioned_migrations(&conn, "build_logs", MIGRATIONS)?;
Ok(Self {
conn: Mutex::new(conn),
})
}
/// Close every row still marked in-flight as failed. Returns the number
/// closed.
///
/// Only sound when no other writer holds this database: the caller must
/// be the one process that owns it, before it starts any build. A row
/// with `finished_at IS NULL` then belongs to a process that died, and
/// nothing will ever finish it. Kept out of `open` so a second opener
/// can't close another process's live rows.
pub fn reap_orphans(&self) -> Result<u64> {
let now = Utc::now().timestamp();
let conn = self.conn.lock().unwrap();
let closed = conn
.execute(
"UPDATE build_logs
SET finished_at = ?1,
status = 'fail',
stderr = stderr || ?2 || x'0a'
WHERE finished_at IS NULL",
params![now, ORPHAN_NOTE],
)
.context("close orphaned build_logs rows")?;
Ok(u64::try_from(closed).unwrap_or(0))
}
/// Open a row for a new build attempt. Returns the assigned id
/// — the caller threads it through `append_stdout` / `append_stderr`
/// while the child runs and into `finish` once it exits.
///
/// `node_id` is the queue node this build belongs to, when there is one.
/// It is `None` for builds that run outside the job queue; those are
/// reachable by agent + time, just not by node.
pub fn start(
&self,
agent: &str,
kind: &str,
cmdline: &str,
node_id: Option<u64>,
) -> Result<i64> {
let now = Utc::now().timestamp();
let node_id = node_id.and_then(|n| i64::try_from(n).ok());
let conn = self.conn.lock().unwrap();
conn.execute(
"INSERT INTO build_logs (agent, kind, cmdline, started_at, node_id) \
VALUES (?1, ?2, ?3, ?4, ?5)",
params![agent, kind, cmdline, now, node_id],
)
.context("insert build_logs row")?;
Ok(conn.last_insert_rowid())
}
/// The most recent build-log row for `node_id`, if any. Replaces the job
/// queue's in-memory `NodeId -> build_log_id` side map: the link lives in
/// the row itself now, so it survives a restart and needs no lock held
/// alongside the scheduler's.
///
/// `MAX(id)` rather than a uniqueness assumption — a node that is retried
/// opens a second row, and the newest is the one the panel should show.
#[must_use]
pub fn id_for_node(&self, node_id: u64) -> Option<i64> {
let node_id = i64::try_from(node_id).ok()?;
let conn = self.conn.lock().unwrap();
conn.query_row(
"SELECT MAX(id) FROM build_logs WHERE node_id = ?1",
params![node_id],
|row| row.get::<_, Option<i64>>(0),
)
.ok()
.flatten()
}
/// Append a single stdout line. Best-effort: errors are logged
/// but never returned to the caller, so a transient sqlite blip
/// never tears down a rebuild's stdout pump.
pub fn append_stdout(&self, id: i64, line: &str) {
self.append(id, "stdout", line);
}
/// Append a single stderr line. Same best-effort contract as
/// `append_stdout`.
pub fn append_stderr(&self, id: i64, line: &str) {
self.append(id, "stderr", line);
}
fn append(&self, id: i64, column: &'static str, line: &str) {
// `column` is hard-coded by the caller (`stdout` / `stderr`)
// — never user-supplied — so the string-format here is safe
// and lets us reuse one helper for both streams.
let sql = format!("UPDATE build_logs SET {column} = {column} || ?1 || x'0a' WHERE id = ?2");
let conn = self.conn.lock().unwrap();
if let Err(e) = conn.execute(&sql, params![line, id]) {
tracing::warn!(
build_log_id = id,
column = column,
error = ?e,
"build_logs: append failed (dropping line)"
);
}
}
/// Finalize a build attempt. Sets `finished_at` to now and
/// `status` to the terminal state. Best-effort.
pub fn finish(&self, id: i64, status: BuildStatus) {
let now = Utc::now().timestamp();
let conn = self.conn.lock().unwrap();
if let Err(e) = conn.execute(
"UPDATE build_logs SET finished_at = ?1, status = ?2 WHERE id = ?3",
params![now, <&str>::from(status), id],
) {
tracing::warn!(
build_log_id = id,
error = ?e,
"build_logs: finish failed"
);
}
}
/// Read one row back, text included. `None` when the id doesn't exist
/// (already vacuumed).
#[cfg(test)]
pub fn get_full(&self, id: i64) -> Result<Option<BuildLogFull>> {
use rusqlite::OptionalExtension;
let conn = self.conn.lock().unwrap();
let row = conn
.query_row(
"SELECT agent, kind, finished_at, status, stdout, stderr \
FROM build_logs WHERE id = ?1",
params![id],
|r| {
Ok(BuildLogFull {
agent: r.get(0)?,
kind: r.get(1)?,
finished_at: r.get(2)?,
status: r.get(3)?,
stdout: r.get(4)?,
stderr: r.get(5)?,
})
},
)
.optional()?;
Ok(row)
}
/// Drop rows past their retention window. Returns the number of
/// rows deleted. Called from this module's hourly `spawn` loop.
///
/// Rule:
/// - `status = 'fail'` rows kept for `KEEP_FAIL_SECS` past their
/// `finished_at` (failures are what operators dig into).
/// - `status = 'ok'` rows kept for `KEEP_OK_SECS` past their
/// `finished_at` (successes are mostly noise after a day).
/// - In-flight rows (`finished_at IS NULL`) are never touched —
/// a long-running build shouldn't disappear from its own log
/// viewer mid-stream.
pub fn vacuum(&self) -> Result<u64> {
let now = Utc::now().timestamp();
let conn = self.conn.lock().unwrap();
let fail_cutoff = now - KEEP_FAIL_SECS;
let ok_cutoff = now - KEEP_OK_SECS;
let removed = conn.execute(
"DELETE FROM build_logs
WHERE finished_at IS NOT NULL
AND (
(status = 'fail' AND finished_at < ?1)
OR (status = 'ok' AND finished_at < ?2)
)",
params![fail_cutoff, ok_cutoff],
)?;
Ok(u64::try_from(removed).unwrap_or(0))
}
}
/// Spawn the hourly retention sweep. A host-side sweep (`build_logs.sqlite`
/// is hive-c0re-owned, so no privsep ownership issue). Runs once at startup
/// before its first sleep so a long-uptime instance doesn't accumulate a
/// backlog the first hour after restart.
pub fn spawn_vacuum(coord: &Arc<crate::coordinator::Coordinator>) {
use std::time::Duration;
let logs = coord.build_logs.clone();
let mut shutdown = coord.shutdown_rx();
let interval = Duration::from_hours(1);
tokio::spawn(async move {
loop {
match logs.vacuum() {
Ok(0) => {}
Ok(n) => tracing::info!(removed = n, "build_logs vacuum"),
Err(e) => tracing::warn!(error = ?e, "build_logs vacuum failed"),
}
tokio::select! {
() = tokio::time::sleep(interval) => {}
_ = shutdown.changed() => {
tracing::info!("build_logs vacuum: shutdown signal received");
break;
}
}
}
});
}
#[cfg(test)]
mod tests {
use super::*;
fn tmpdb() -> (tempfile::TempDir, BuildLogs) {
let dir = tempfile::tempdir().expect("tempdir");
let db = BuildLogs::open(dir.path()).expect("open");
(dir, db)
}
#[test]
fn start_appends_finish_flow() {
let (_d, db) = tmpdb();
let id = db
.start("alice", "prebuild", "nix build foo", None)
.expect("start");
db.append_stdout(id, "building '/nix/store/abc.drv'");
db.append_stderr(id, "error: line 12");
db.append_stderr(id, " at /nix/store/.../module.nix:5");
db.finish(id, BuildStatus::Fail);
let full = db.get_full(id).expect("get").expect("Some");
assert_eq!(full.agent, "alice");
assert_eq!(full.kind, "prebuild");
assert_eq!(full.status.as_deref(), Some("fail"));
assert!(full.finished_at.is_some());
assert!(full.stdout.contains("/nix/store/abc.drv"));
assert!(full.stderr.contains("error: line 12"));
assert!(full.stderr.contains("module.nix:5"));
// Lines are terminator-delimited so each contributes a trailing
// newline — the viewer joins on the existing newlines rather
// than re-inserting them.
assert!(full.stderr.ends_with('\n'));
}
#[test]
fn reap_orphans_fails_rows_left_in_flight_by_a_previous_process() {
let dir = tempfile::tempdir().expect("tempdir");
let (orphan, done) = {
let db = BuildLogs::open(dir.path()).expect("open");
let orphan = db.start("alice", "run", "cmd one", None).expect("start");
db.append_stdout(orphan, "building");
let done = db.start("bob", "run", "cmd two", None).expect("start");
db.finish(done, BuildStatus::Ok);
(orphan, done)
};
let db = BuildLogs::open(dir.path()).expect("reopen");
assert_eq!(db.reap_orphans().expect("reap"), 1);
let row = db.get_full(orphan).expect("get").expect("Some");
assert_eq!(row.status.as_deref(), Some("fail"));
assert!(row.finished_at.is_some());
assert!(row.stdout.contains("building"));
assert!(row.stderr.ends_with(&format!("{ORPHAN_NOTE}\n")));
let ok = db.get_full(done).expect("get").expect("Some");
assert_eq!(ok.status.as_deref(), Some("ok"));
assert_eq!(db.reap_orphans().expect("reap again"), 0);
}
#[test]
fn get_full_returns_none_for_missing_id() {
let (_d, db) = tmpdb();
let missing = db.get_full(999_999).expect("get");
assert!(missing.is_none());
}
#[test]
fn vacuum_drops_old_finished_only_per_status() {
let (_d, db) = tmpdb();
let id_fresh_fail = db.start("alice", "run", "fresh fail", None).expect("start");
let id_old_fail = db.start("alice", "run", "old fail", None).expect("start");
let id_old_ok = db.start("alice", "run", "old ok", None).expect("start");
let id_running = db
.start("alice", "run", "still running", None)
.expect("start");
db.finish(id_fresh_fail, BuildStatus::Fail);
db.finish(id_old_fail, BuildStatus::Fail);
db.finish(id_old_ok, BuildStatus::Ok);
// Backdate two rows past their retention windows. fresh_fail
// stays within KEEP_FAIL_SECS so it survives; old_fail goes
// beyond; old_ok goes past KEEP_OK_SECS but inside
// KEEP_FAIL_SECS — proves the per-status rule.
let now = Utc::now().timestamp();
{
let conn = db.conn.lock().unwrap();
conn.execute(
"UPDATE build_logs SET finished_at = ?1 WHERE id = ?2",
params![now - KEEP_FAIL_SECS - 60, id_old_fail],
)
.unwrap();
conn.execute(
"UPDATE build_logs SET finished_at = ?1 WHERE id = ?2",
params![now - KEEP_OK_SECS - 60, id_old_ok],
)
.unwrap();
}
let removed = db.vacuum().expect("vacuum");
assert_eq!(removed, 2, "old_fail + old_ok should be vacuumed");
assert!(db.get_full(id_fresh_fail).unwrap().is_some());
assert!(db.get_full(id_old_fail).unwrap().is_none());
assert!(db.get_full(id_old_ok).unwrap().is_none());
// Running row must survive vacuum regardless of retention
// windows — finished_at IS NULL gates it out.
assert!(db.get_full(id_running).unwrap().is_some());
}
#[test]
fn node_link_survives_completion_and_newest_wins() {
// The queue used to hold this link in an in-memory side map, which
// meant it died with the process and needed the queue lock to read.
// On the row it outlives both the node's completion and a restart.
let (_d, db) = tmpdb();
let first = db.start("alice", "swap", "cmd", Some(7)).expect("start");
db.finish(first, BuildStatus::Fail);
assert_eq!(
db.id_for_node(7),
Some(first),
"link survives the build finishing"
);
// A retried node opens a second row; the panel wants the current
// attempt, not the first one.
let retry = db.start("alice", "swap", "cmd", Some(7)).expect("start");
assert_eq!(db.id_for_node(7), Some(retry), "newest attempt wins");
// Builds that run outside the queue carry no node and are found by
// agent + time instead — they must not collide with node lookups.
db.start("alice", "run", "no node", None).expect("start");
assert_eq!(db.id_for_node(999), None, "unknown node → no row");
}
#[test]
fn append_after_finish_still_appends() {
// Defensive: if a child's stdout pump fires one last line
// between the wait-syscall returning and `finish` running, the
// append should land on the row (status already set, but the
// log stays consistent with what happened).
let (_d, db) = tmpdb();
let id = db.start("alice", "run", "cmd", None).expect("start");
db.finish(id, BuildStatus::Ok);
db.append_stdout(id, "post-finish trailing line");
let full = db.get_full(id).expect("get").expect("Some");
assert!(full.stdout.contains("post-finish trailing line"));
}
}

7
hive-c0re/src/stores/mod.rs
View file

@ -1,8 +1,7 @@
//! Sqlite-backed host-side stores (broker, build logs, power intent)
//! plus the shared connection open/migration helper (`db`). Each submodule is re-exported at the crate root, so
//! `crate::broker::…` etc. keep working unchanged.
//! Sqlite-backed host-side stores (broker, power intent) plus the shared
//! connection open/migration helper (`db`). Each submodule is re-exported at the
//! crate root, so `crate::broker::…` etc. keep working unchanged.
pub mod broker;
pub mod build_logs;
pub mod db;
pub mod power;