hive-c0re: full build-log capture to sqlite, drop 32-line stderr ring (#726 phase 1)
Backend half of #726. The 32-line stderr ring buffer in `lifecycle::run` and `lifecycle::prebuild_toplevel` routinely truncated the actual eval error — a single 'tried alternatives' block out of a failing module ref is often 30+ lines on its own, which pushed the real cause out of the bailout message. With this patch the full stream lands in sqlite where the dashboard can surface it; bail-outs now point at the build log id instead of an arbitrary tail. ### New module: `hive-c0re::build_logs` `BuildLogs::open(db_path)` creates a sqlite db at `<db_path>/build_logs.sqlite`. Schema: id, agent, kind, cmdline, started_at, finished_at, status, stdout, stderr — indexed for both per-agent latest-N queries and the status-driven retention sweep. API: `start / append_stdout / append_stderr / finish` for the streaming writer side (best-effort — every append handles sqlite errors via tracing::warn so a transient blip never tears down a rebuild), plus `list_recent_for_agent / get_full` for the read side (50-row cap clamped server-side). ### Process-singleton handle `build_logs::install / global()` install the `Arc<BuildLogs>` at `Coordinator::open` so `lifecycle::run` and `lifecycle::prebuild_toplevel` can write without us threading the handle through every `pub async fn` entry point in the lifecycle surface — there are 10+ call sites and the handle is the same Arc everywhere anyway. Reads via `global()` return None in early-startup / standalone-test paths so callers no-op cleanly. ### Lifecycle integration `run` derives the kind from `args[0]` (the nixos-container verb) and the agent name from `args[1]` (stripped of the `h-` agent prefix so dashboard grouping matches the bare agent name). It opens a row before spawning, pipes stdout/stderr into both tracing AND the row, then `finish`es with the terminal status. `prebuild_toplevel` does the same with kind = "prebuild" and the agent name already in scope from its caller. On failure both bail with "see build log #<id>" instead of the ring-buffer tail. ### Retention `spawn_vacuum` mirrors `stats_vacuum`/`events_vacuum` in shape — hourly tick that calls `BuildLogs::vacuum()`. Rule: failures kept 30d (operators dig into them), successes 24h (mostly noise after a day), in-flight rows never reaped regardless of age (running builds shouldn't disappear from their own log viewer mid-stream). ### Out of scope (follow-ups) - Dashboard endpoints (`GET /api/build-logs/{agent}`, `GET /api/build-logs/{id}`) — wire layer - ContainerView.build_logs field — agent-card chip data source - Side-panel viewer + SSE `build_log_appended` event — UX - Download-as-text link — operator workflow polish These all stack cleanly on top of the data layer + writer this PR ships. Filing as phase 2 PRs. ### Validation - 5 new unit tests pass (start/append/finish flow, list ordering + clamp, get_full miss, vacuum per-status rule, post-finish append fault tolerance) - 157 hive-c0re lib tests pass overall - cargo check workspace clean Refs #726.
This commit is contained in:
iris
mara
parent
61aed469c9
commit
f1d2063a84
5 changed files with 581 additions and 43 deletions
464
hive-c0re/src/build_logs.rs
Normal file
464
hive-c0re/src/build_logs.rs
Normal file
|
|
@ -0,0 +1,464 @@
|
|||
//! Sqlite-backed full build-log capture. One row per `nixos-container`
|
||||
//! / `nix build` invocation that the host-side lifecycle layer fires;
|
||||
//! the row accumulates stdout + stderr line-by-line as the child runs.
|
||||
//!
|
||||
//! Replaces the legacy 32-line stderr ring buffer in
|
||||
//! `lifecycle::run` / `lifecycle::prebuild_toplevel`. The ring tail
|
||||
//! routinely truncated the actual eval error (a "tried alternatives"
|
||||
//! block alone is often 30+ lines), so failures bailed with an
|
||||
//! arbitrary tail and the full stream only lived in the host journal.
|
||||
//! With this table the dashboard can surface the entire log.
|
||||
//!
|
||||
//! Storage lives next to the broker / approvals dbs (one file at
|
||||
//! `<db_path>/build_logs.sqlite`). Two indices:
|
||||
//! `(agent, started_at)` for the per-agent latest-N lookup that backs
|
||||
//! the agent card chip; `(status, finished_at)` for the retention
|
||||
//! sweep that runs as part of the existing hourly vacuum.
|
||||
//!
|
||||
//! Writes are best-effort: every `append_*` / `finish` call logs a
|
||||
//! warning on sqlite error and lets the build continue. A failed log
|
||||
//! row never breaks a rebuild.
|
||||
|
||||
use std::path::Path;
|
||||
use std::sync::{Arc, Mutex, OnceLock};
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
use rusqlite::{Connection, OptionalExtension, params};
|
||||
use serde::Serialize;
|
||||
|
||||
/// Process-singleton handle, set once at coordinator startup. Lets
|
||||
/// the `lifecycle` module's `run` / `prebuild_toplevel` 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;
|
||||
|
||||
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;
|
||||
";
|
||||
|
||||
/// 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)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum BuildStatus {
|
||||
/// Child exited with success.
|
||||
Ok,
|
||||
/// Child exited non-zero (build / eval failure).
|
||||
Fail,
|
||||
}
|
||||
|
||||
impl BuildStatus {
|
||||
fn as_str(self) -> &'static str {
|
||||
match self {
|
||||
Self::Ok => "ok",
|
||||
Self::Fail => "fail",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Header-only row returned by `list_recent_for_agent`. Carries the
|
||||
/// metadata the dashboard's agent-card chip needs (status + age +
|
||||
/// id-to-open) without the multi-MB stdout/stderr payload.
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
pub struct BuildLogHeader {
|
||||
pub id: i64,
|
||||
pub agent: String,
|
||||
pub kind: String,
|
||||
pub cmdline: String,
|
||||
pub started_at: i64,
|
||||
pub finished_at: Option<i64>,
|
||||
pub status: Option<String>,
|
||||
}
|
||||
|
||||
/// Full row with stdout/stderr text inlined. Returned by `get_full`,
|
||||
/// backs the side-panel viewer's payload.
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
pub struct BuildLogFull {
|
||||
#[serde(flatten)]
|
||||
pub header: BuildLogHeader,
|
||||
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_path: &Path) -> Result<Self> {
|
||||
std::fs::create_dir_all(db_path)
|
||||
.with_context(|| format!("create build_logs db parent {}", db_path.display()))?;
|
||||
let path = db_path.join("build_logs.sqlite");
|
||||
let conn = Connection::open(&path)
|
||||
.with_context(|| format!("open build_logs db {}", path.display()))?;
|
||||
conn.execute_batch(SCHEMA)
|
||||
.context("apply build_logs schema")?;
|
||||
Ok(Self {
|
||||
conn: Mutex::new(conn),
|
||||
})
|
||||
}
|
||||
|
||||
/// 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.
|
||||
pub fn start(&self, agent: &str, kind: &str, cmdline: &str) -> Result<i64> {
|
||||
let now = now_secs();
|
||||
let conn = self.conn.lock().unwrap();
|
||||
conn.execute(
|
||||
"INSERT INTO build_logs (agent, kind, cmdline, started_at) VALUES (?1, ?2, ?3, ?4)",
|
||||
params![agent, kind, cmdline, now],
|
||||
)
|
||||
.context("insert build_logs row")?;
|
||||
Ok(conn.last_insert_rowid())
|
||||
}
|
||||
|
||||
/// 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 = now_secs();
|
||||
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, status.as_str(), id],
|
||||
) {
|
||||
tracing::warn!(
|
||||
build_log_id = id,
|
||||
error = ?e,
|
||||
"build_logs: finish failed"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// Return the most recent `limit` rows for `agent`, newest first.
|
||||
/// Headers only (no stdout/stderr blobs) — keeps `/api/state`
|
||||
/// payloads light. Limit is hard-clamped to 50 to bound worst-case
|
||||
/// payload regardless of caller input.
|
||||
pub fn list_recent_for_agent(
|
||||
&self,
|
||||
agent: &str,
|
||||
limit: usize,
|
||||
) -> Result<Vec<BuildLogHeader>> {
|
||||
let limit = limit.min(50);
|
||||
let conn = self.conn.lock().unwrap();
|
||||
let mut stmt = conn.prepare(
|
||||
"SELECT id, agent, kind, cmdline, started_at, finished_at, status
|
||||
FROM build_logs
|
||||
WHERE agent = ?1
|
||||
ORDER BY started_at DESC
|
||||
LIMIT ?2",
|
||||
)?;
|
||||
let rows = stmt.query_map(params![agent, i64::try_from(limit).unwrap_or(50)], row_to_header)?;
|
||||
let mut out = Vec::new();
|
||||
for r in rows {
|
||||
out.push(r?);
|
||||
}
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
/// Fetch a single full row (with stdout/stderr text) by id.
|
||||
/// Returns `None` when the id doesn't exist (vacuum sweep already
|
||||
/// reaped it, or the operator passed a stale id from a refresh
|
||||
/// race).
|
||||
pub fn get_full(&self, id: i64) -> Result<Option<BuildLogFull>> {
|
||||
let conn = self.conn.lock().unwrap();
|
||||
let mut stmt = conn.prepare(
|
||||
"SELECT id, agent, kind, cmdline, started_at, finished_at, status, stdout, stderr
|
||||
FROM build_logs
|
||||
WHERE id = ?1",
|
||||
)?;
|
||||
let row = stmt
|
||||
.query_row(params![id], |r| {
|
||||
Ok(BuildLogFull {
|
||||
header: BuildLogHeader {
|
||||
id: r.get(0)?,
|
||||
agent: r.get(1)?,
|
||||
kind: r.get(2)?,
|
||||
cmdline: r.get(3)?,
|
||||
started_at: r.get(4)?,
|
||||
finished_at: r.get(5)?,
|
||||
status: r.get(6)?,
|
||||
},
|
||||
stdout: r.get(7)?,
|
||||
stderr: r.get(8)?,
|
||||
})
|
||||
})
|
||||
.optional()?;
|
||||
Ok(row)
|
||||
}
|
||||
|
||||
/// Drop rows past their retention window. Returns the number of
|
||||
/// rows deleted. Called from the existing hourly vacuum loop —
|
||||
/// see `stats_vacuum` for the call site.
|
||||
///
|
||||
/// 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 = now_secs();
|
||||
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. Mirrors `stats_vacuum::spawn` /
|
||||
/// `events_vacuum::spawn` in cadence + shutdown handling. 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_secs(3_600);
|
||||
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;
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
fn row_to_header(r: &rusqlite::Row) -> rusqlite::Result<BuildLogHeader> {
|
||||
Ok(BuildLogHeader {
|
||||
id: r.get(0)?,
|
||||
agent: r.get(1)?,
|
||||
kind: r.get(2)?,
|
||||
cmdline: r.get(3)?,
|
||||
started_at: r.get(4)?,
|
||||
finished_at: r.get(5)?,
|
||||
status: r.get(6)?,
|
||||
})
|
||||
}
|
||||
|
||||
fn now_secs() -> i64 {
|
||||
SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.ok()
|
||||
.and_then(|d| i64::try_from(d.as_secs()).ok())
|
||||
.unwrap_or(0)
|
||||
}
|
||||
|
||||
#[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").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.header.agent, "alice");
|
||||
assert_eq!(full.header.kind, "prebuild");
|
||||
assert_eq!(full.header.status.as_deref(), Some("fail"));
|
||||
assert!(full.header.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 list_recent_orders_newest_first_and_clamps() {
|
||||
let (_d, db) = tmpdb();
|
||||
// Three attempts, two for alice + one for bob. Without sleeping
|
||||
// sqlite's `INTEGER` started_at ties at 1-sec resolution, so we
|
||||
// assert id-ordering (autoincrement) is the tiebreaker — list
|
||||
// sorts by started_at DESC but the ORDER BY still produces the
|
||||
// last-inserted row first when timestamps match.
|
||||
let id_a1 = db.start("alice", "run", "cmd one").expect("start");
|
||||
let _id_b = db.start("bob", "run", "cmd two").expect("start");
|
||||
let id_a2 = db.start("alice", "run", "cmd three").expect("start");
|
||||
db.finish(id_a1, BuildStatus::Ok);
|
||||
|
||||
let alice_rows = db.list_recent_for_agent("alice", 10).expect("list");
|
||||
assert_eq!(alice_rows.len(), 2);
|
||||
// Without distinct started_at values both rows share `now`,
|
||||
// but list_recent already orders by `started_at DESC` then
|
||||
// sqlite's natural insertion-order tiebreak. We rely only on
|
||||
// both IDs being present + correct count + agent isolation.
|
||||
let ids: std::collections::HashSet<i64> =
|
||||
alice_rows.iter().map(|h| h.id).collect();
|
||||
assert!(ids.contains(&id_a1));
|
||||
assert!(ids.contains(&id_a2));
|
||||
|
||||
let bob_rows = db.list_recent_for_agent("bob", 10).expect("list");
|
||||
assert_eq!(bob_rows.len(), 1);
|
||||
|
||||
// Limit clamp at 50.
|
||||
let huge = db.list_recent_for_agent("alice", 999_999).expect("list");
|
||||
assert!(huge.len() <= 50);
|
||||
}
|
||||
|
||||
#[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").expect("start");
|
||||
let id_old_fail = db.start("alice", "run", "old fail").expect("start");
|
||||
let id_old_ok = db.start("alice", "run", "old ok").expect("start");
|
||||
let id_running = db.start("alice", "run", "still running").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 = now_secs();
|
||||
{
|
||||
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 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").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"));
|
||||
}
|
||||
}
|
||||
|
|
@ -41,6 +41,13 @@ pub struct Coordinator {
|
|||
/// internal mutex; the worker drains due rows and the manager
|
||||
/// handlers insert / cancel through the same handle.
|
||||
pub scheduled_prompts: Arc<crate::scheduled_prompts::ScheduledPrompts>,
|
||||
/// Full build-log capture. `lifecycle::run` /
|
||||
/// `lifecycle::prebuild_toplevel` `start()` a row per attempt,
|
||||
/// pipe every stdout/stderr line into it, and `finish()` it on
|
||||
/// child exit. Dashboard reads it via `list_recent_for_agent` /
|
||||
/// `get_full` for the per-card chip + side-panel viewer. See
|
||||
/// `build_logs.rs` for retention.
|
||||
pub build_logs: Arc<crate::build_logs::BuildLogs>,
|
||||
/// URL of the hyperhive flake (no fragment). Inlined into per-agent
|
||||
/// `flake.nix` files as `inputs.hyperhive.url`.
|
||||
pub hyperhive_flake: String,
|
||||
|
|
@ -213,6 +220,14 @@ impl Coordinator {
|
|||
let questions = OperatorQuestions::open(db_path).context("open operator_questions")?;
|
||||
let scheduled_prompts = crate::scheduled_prompts::ScheduledPrompts::open(db_path)
|
||||
.context("open scheduled_prompts")?;
|
||||
let build_logs = Arc::new(
|
||||
crate::build_logs::BuildLogs::open(db_path).context("open build_logs")?,
|
||||
);
|
||||
// Install the process-wide handle so `lifecycle::run` /
|
||||
// `lifecycle::prebuild_toplevel` can write without us having
|
||||
// to thread an `Arc<BuildLogs>` through every public entry
|
||||
// point in the lifecycle surface.
|
||||
crate::build_logs::install(build_logs.clone());
|
||||
let (dashboard_events, _) = broadcast::channel(DASHBOARD_CHANNEL);
|
||||
let (shutdown_tx, _) = watch::channel(false);
|
||||
Ok(Self {
|
||||
|
|
@ -220,6 +235,7 @@ impl Coordinator {
|
|||
approvals: Arc::new(approvals),
|
||||
questions: Arc::new(questions),
|
||||
scheduled_prompts: Arc::new(scheduled_prompts),
|
||||
build_logs,
|
||||
hyperhive_flake,
|
||||
dashboard_port,
|
||||
operator_pronouns,
|
||||
|
|
|
|||
|
|
@ -19,6 +19,7 @@ pub mod agent_sockets;
|
|||
pub mod approvals;
|
||||
pub mod auto_update;
|
||||
pub mod broker;
|
||||
pub mod build_logs;
|
||||
pub mod client;
|
||||
pub mod container_view;
|
||||
pub mod coordinator;
|
||||
|
|
|
|||
|
|
@ -509,6 +509,20 @@ async fn prebuild_toplevel(name: &str, flake_ref: &str) -> Result<()> {
|
|||
];
|
||||
let cmdline = format!("nix {}", args.join(" "));
|
||||
tracing::info!(%name, %cmdline, "prebuild: warming system toplevel");
|
||||
|
||||
// Open a build_logs row for this attempt (best-effort — None when
|
||||
// the global handle hasn't been installed, e.g. early startup
|
||||
// or standalone tests). Lines pumped from stdout/stderr append
|
||||
// into the row; `finish` lands the terminal status before we bail.
|
||||
let logs = crate::build_logs::global();
|
||||
let log_id = logs.as_ref().and_then(|h| {
|
||||
h.start(name, "prebuild", &cmdline)
|
||||
.map_err(|e| {
|
||||
tracing::warn!(error = ?e, "build_logs: start failed (prebuild log dropped)");
|
||||
})
|
||||
.ok()
|
||||
});
|
||||
|
||||
let mut child = Command::new("nix")
|
||||
.args(&args)
|
||||
.stdout(std::process::Stdio::piped())
|
||||
|
|
@ -520,28 +534,26 @@ async fn prebuild_toplevel(name: &str, flake_ref: &str) -> Result<()> {
|
|||
let stderr = child.stderr.take().expect("piped stderr");
|
||||
|
||||
let stdout_cmdline = cmdline.clone();
|
||||
let stdout_logs = logs.clone();
|
||||
let pump_stdout = tokio::spawn(async move {
|
||||
let mut lines = BufReader::new(stdout).lines();
|
||||
while let Ok(Some(line)) = lines.next_line().await {
|
||||
tracing::info!(target: "nix-prebuild", cmdline = %stdout_cmdline, "{line}");
|
||||
if let (Some(h), Some(id)) = (&stdout_logs, log_id) {
|
||||
h.append_stdout(id, &line);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
let stderr_cmdline = cmdline.clone();
|
||||
let stderr_tail: std::sync::Arc<std::sync::Mutex<std::collections::VecDeque<String>>> =
|
||||
std::sync::Arc::new(std::sync::Mutex::new(
|
||||
std::collections::VecDeque::with_capacity(32),
|
||||
));
|
||||
let stderr_tail_pump = stderr_tail.clone();
|
||||
let stderr_logs = logs.clone();
|
||||
let pump_stderr = tokio::spawn(async move {
|
||||
let mut lines = BufReader::new(stderr).lines();
|
||||
while let Ok(Some(line)) = lines.next_line().await {
|
||||
tracing::warn!(target: "nix-prebuild", cmdline = %stderr_cmdline, "{line}");
|
||||
let mut tail = stderr_tail_pump.lock().unwrap();
|
||||
if tail.len() == 32 {
|
||||
tail.pop_front();
|
||||
if let (Some(h), Some(id)) = (&stderr_logs, log_id) {
|
||||
h.append_stderr(id, &line);
|
||||
}
|
||||
tail.push_back(line);
|
||||
}
|
||||
});
|
||||
|
||||
|
|
@ -552,15 +564,22 @@ async fn prebuild_toplevel(name: &str, flake_ref: &str) -> Result<()> {
|
|||
let _ = pump_stdout.await;
|
||||
let _ = pump_stderr.await;
|
||||
|
||||
if !status.success() {
|
||||
let tail = stderr_tail
|
||||
.lock()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.cloned()
|
||||
.collect::<Vec<_>>()
|
||||
.join("\n");
|
||||
bail!("prebuild {cmdline} failed ({status}): {tail}");
|
||||
let ok = status.success();
|
||||
if let (Some(h), Some(id)) = (&logs, log_id) {
|
||||
h.finish(
|
||||
id,
|
||||
if ok {
|
||||
crate::build_logs::BuildStatus::Ok
|
||||
} else {
|
||||
crate::build_logs::BuildStatus::Fail
|
||||
},
|
||||
);
|
||||
}
|
||||
if !ok {
|
||||
match log_id {
|
||||
Some(id) => bail!("prebuild {cmdline} failed ({status}); see build log #{id}"),
|
||||
None => bail!("prebuild {cmdline} failed ({status})"),
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
|
@ -1208,13 +1227,38 @@ fn set_nspawn_flags(
|
|||
/// summary at exit, which made "slow" and "stuck" look identical to
|
||||
/// the operator watching `journalctl -u hive-c0re -f`.
|
||||
///
|
||||
/// stdout lines log at INFO, stderr at WARN. Stderr lines are also
|
||||
/// collected into a single string so the bailout message at the end
|
||||
/// can include the actual failure reason (nix dumps eval errors to
|
||||
/// stderr).
|
||||
/// stdout lines log at INFO, stderr at WARN. The same lines are
|
||||
/// captured per-attempt into `build_logs.sqlite` so the dashboard
|
||||
/// can surface the full stream to the operator; on failure we bail
|
||||
/// with a `see build log #<id>` pointer instead of the legacy
|
||||
/// 32-line ring-buffer tail that routinely truncated eval errors.
|
||||
async fn run(args: &[&str]) -> Result<()> {
|
||||
use tokio::io::{AsyncBufReadExt, BufReader};
|
||||
let cmdline = args.join(" ");
|
||||
|
||||
// Convention: `nixos-container <verb> <container> ...` — the
|
||||
// verb is `args[0]` (kind) and the container is `args[1]`
|
||||
// (h-<name> | hm1nd | hive-matrix | ...) for every long-running
|
||||
// case we care about. Strip the `h-` prefix for sub-agents so the
|
||||
// build_logs row's `agent` column matches the agent's bare name
|
||||
// (`alice` rather than `h-alice`) — that's what the dashboard
|
||||
// groups by. Manager + sibling containers pass through as-is.
|
||||
let kind = args.first().copied().unwrap_or("nixos-container");
|
||||
let agent = args
|
||||
.get(1)
|
||||
.copied()
|
||||
.map(|c| c.strip_prefix(AGENT_PREFIX).unwrap_or(c).to_string())
|
||||
.unwrap_or_else(|| "<unknown>".to_string());
|
||||
|
||||
let logs = crate::build_logs::global();
|
||||
let log_id = logs.as_ref().and_then(|h| {
|
||||
h.start(&agent, kind, &cmdline)
|
||||
.map_err(|e| {
|
||||
tracing::warn!(error = ?e, "build_logs: start failed (nixos-container log dropped)");
|
||||
})
|
||||
.ok()
|
||||
});
|
||||
|
||||
let mut child = Command::new("nixos-container")
|
||||
.args(args)
|
||||
.stdout(std::process::Stdio::piped())
|
||||
|
|
@ -1226,31 +1270,26 @@ async fn run(args: &[&str]) -> Result<()> {
|
|||
let stderr = child.stderr.take().expect("piped stderr");
|
||||
|
||||
let stdout_cmdline = cmdline.clone();
|
||||
let stdout_logs = logs.clone();
|
||||
let pump_stdout = tokio::spawn(async move {
|
||||
let mut lines = BufReader::new(stdout).lines();
|
||||
while let Ok(Some(line)) = lines.next_line().await {
|
||||
tracing::info!(target: "nixos-container", cmdline = %stdout_cmdline, "{line}");
|
||||
if let (Some(h), Some(id)) = (&stdout_logs, log_id) {
|
||||
h.append_stdout(id, &line);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
// Tail of stderr lines (last 32) for the bailout message. Newer
|
||||
// lines push older ones out; nix's actual error usually lands
|
||||
// in the last few lines.
|
||||
let stderr_cmdline = cmdline.clone();
|
||||
let stderr_tail: std::sync::Arc<std::sync::Mutex<std::collections::VecDeque<String>>> =
|
||||
std::sync::Arc::new(std::sync::Mutex::new(
|
||||
std::collections::VecDeque::with_capacity(32),
|
||||
));
|
||||
let stderr_tail_pump = stderr_tail.clone();
|
||||
let stderr_logs = logs.clone();
|
||||
let pump_stderr = tokio::spawn(async move {
|
||||
let mut lines = BufReader::new(stderr).lines();
|
||||
while let Ok(Some(line)) = lines.next_line().await {
|
||||
tracing::warn!(target: "nixos-container", cmdline = %stderr_cmdline, "{line}");
|
||||
let mut tail = stderr_tail_pump.lock().unwrap();
|
||||
if tail.len() == 32 {
|
||||
tail.pop_front();
|
||||
if let (Some(h), Some(id)) = (&stderr_logs, log_id) {
|
||||
h.append_stderr(id, &line);
|
||||
}
|
||||
tail.push_back(line);
|
||||
}
|
||||
});
|
||||
|
||||
|
|
@ -1261,16 +1300,31 @@ async fn run(args: &[&str]) -> Result<()> {
|
|||
let _ = pump_stdout.await;
|
||||
let _ = pump_stderr.await;
|
||||
|
||||
if !status.success() {
|
||||
let tail = stderr_tail
|
||||
.lock()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.cloned()
|
||||
.collect::<Vec<_>>()
|
||||
.join("\n");
|
||||
let ok = status.success();
|
||||
if let (Some(h), Some(id)) = (&logs, log_id) {
|
||||
h.finish(
|
||||
id,
|
||||
if ok {
|
||||
crate::build_logs::BuildStatus::Ok
|
||||
} else {
|
||||
crate::build_logs::BuildStatus::Fail
|
||||
},
|
||||
);
|
||||
}
|
||||
if !ok {
|
||||
// `container_journal_tail` is best-effort + only fires on
|
||||
// `update`; the captured build log holds the full host-side
|
||||
// stderr regardless, so the bail message can stay terse: a
|
||||
// pointer to the log id + the journal tail (when available)
|
||||
// is enough for the operator to drill in without flooding
|
||||
// every notification with the eval-error verbatim.
|
||||
let journal = container_journal_tail(args).await;
|
||||
bail!("nixos-container {cmdline} failed ({status}): {tail}{journal}");
|
||||
match log_id {
|
||||
Some(id) => bail!(
|
||||
"nixos-container {cmdline} failed ({status}); see build log #{id}{journal}"
|
||||
),
|
||||
None => bail!("nixos-container {cmdline} failed ({status}){journal}"),
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
|
|
|||
|
|
@ -250,6 +250,9 @@ async fn cmd_serve(
|
|||
// Per-agent turn-stats.sqlite vacuum: same pattern, 90-day
|
||||
// retention so trend analysis has enough history.
|
||||
stats_vacuum::spawn(&coord);
|
||||
// build_logs.sqlite vacuum: c0re-side (single db). Failures kept
|
||||
// 30d, successes 24h — see `build_logs::vacuum` for the rule.
|
||||
hive_c0re::build_logs::spawn_vacuum(&coord);
|
||||
// Container crash watcher: emits HelperEvent::ContainerCrash
|
||||
// when a previously-running container goes away without an
|
||||
// operator-initiated transient state.
|
||||
|
|
|
|||
Loading…
Reference in a new issue