From 49266e4410eb16a92a742bfa7198f06e0683e562 Mon Sep 17 00:00:00 2001 From: atlas Date: Fri, 13 Nov 2026 10:08:19 +0100 Subject: [PATCH] remove what no longer has a reader: build-log store, dashboard sent/delivered events, graph resource dep --- .../agent-state/persistence.md | 45 +- .../4-implementation/scheduler/coordinator.md | 9 +- .../trust-boundary/hive-c0re-privilege.md | 2 +- docs/4-implementation/web-ui/dashboard.md | 9 +- docs/4-implementation/web-ui/shape.md | 5 +- hive-c0re/src/coordinator.rs | 60 +-- hive-c0re/src/dashboard_events.rs | 45 +- hive-c0re/src/job_queue/exec.rs | 17 +- hive-c0re/src/job_queue/exec/tests.rs | 39 +- hive-c0re/src/job_queue/mod.rs | 6 +- hive-c0re/src/job_queue/model.rs | 10 +- hive-c0re/src/job_queue/resource.rs | 17 - hive-c0re/src/job_queue/tests.rs | 5 - hive-c0re/src/lifecycle/mod.rs | 130 ++--- hive-c0re/src/lifecycle/tests.rs | 85 ++-- hive-c0re/src/main.rs | 68 +-- hive-c0re/src/paths.rs | 11 +- hive-c0re/src/priv_client.rs | 2 +- hive-c0re/src/stores/broker.rs | 52 +- hive-c0re/src/stores/build_logs.rs | 478 ------------------ hive-c0re/src/stores/mod.rs | 7 +- hive-jobq-wire/README.md | 9 +- hive-jobq-wire/src/lib.rs | 124 ++--- hive-priv/src/container.rs | 4 +- hivectl/src/dag_progress.rs | 7 +- nix/module-eval/grafana.nix | 1 - swarm-controller/src/jobq.rs | 9 - 27 files changed, 203 insertions(+), 1053 deletions(-) delete mode 100644 hive-c0re/src/stores/build_logs.rs diff --git a/docs/4-implementation/agent-state/persistence.md b/docs/4-implementation/agent-state/persistence.md index ecabda264..addafed3b 100644 --- a/docs/4-implementation/agent-state/persistence.md +++ b/docs/4-implementation/agent-state/persistence.md @@ -142,45 +142,20 @@ hive-c0re reads this file on each `build_all` sweep (~10s) via sentinel files (`hyperhive-rate-limited`, `hyperhive-needs-login`) when the JSON is absent. -### `/var/lib/hyperhive/db/build_logs.sqlite` (host) +### Container-operation logs (host) -Full stdout + stderr capture for every `nixos-container` / `nix -build` invocation the lifecycle layer fires. One row per invocation; -the row accumulates lines as the child runs. - -Capturing the full stream (rather than a short tail buffer) matters -because real eval errors routinely run long — "tried alternatives" -blocks alone are often 30+ lines — so a truncated tail would cut off -the actual failure and leave only the host journal holding the -complete output. This table holds the entire log; no HTTP route serves -it. - -Three indices: - -- `(agent, started_at)` — per-agent lookup by recency. -- `(status, finished_at)` — backs the retention sweep that runs - as part of the existing hourly vacuum. -- `(node_id)` — added by a later migration so a build log row can be - looked up by the job-queue node it belongs to (a `hive_jobq` node is - immutable after insert, so hive-c0re records the link on the log row - instead); `node_id` is nullable. - -Writes are best-effort: `append_stdout` / `append_stderr` / `finish` -log a warning on sqlite error and let the build continue. A failed -log row never blocks a rebuild. - -Each operation also logs to the host journal, target `nixos-container`, and the -swarm log store collects those lines. Every line carries `build_id` (this -table's row id; empty without the store), `agent`, `kind` and `cmdline`. A +hive-c0re keeps no database of `nixos-container` / `nix build` output. Each +operation logs to the host journal, target `nixos-container`, and the swarm log +store collects those lines. Every line carries `agent`, `kind` and `cmdline`. A `start` line (`event=start`) opens the build, one line follows per line of output, and a `finish` line (`event=finish`) closes it with `status` (`ok` or -`fail`) and `runtime_secs`. hive-c0re logs a nix warning or error at warn, and a failed -finish too. +`fail`) and `runtime_secs`. hive-c0re logs a nix warning or error at warn, and a +failed finish too. A failed `start` also logs its error and the container +unit's journal tail. -A row is open (`finished_at IS NULL`) only while its child runs. At startup -`Coordinator::open` closes every row still open as `fail` and appends a -line saying hive-c0re stopped: the previous process died mid-build, and -the retention sweep never deletes open rows. +An earlier hive-c0re stored the same output in +`/var/lib/hyperhive/db/build_logs.sqlite`. Nothing reads or writes that file now, and +hive-c0re leaves it on disk untouched. ### `/harness/hyperhive-model` (per agent) diff --git a/docs/4-implementation/scheduler/coordinator.md b/docs/4-implementation/scheduler/coordinator.md index b8d62a81e..992c5ecec 100644 --- a/docs/4-implementation/scheduler/coordinator.md +++ b/docs/4-implementation/scheduler/coordinator.md @@ -25,8 +25,7 @@ from the same basic nodes. ### Two levels: DAG and node The **DAG** is the unit of cancel and the -dashboard group; the **node** is the unit of scheduling / execution / -build-log. Deps are intra-DAG edges only (`AfterOk` by default: +dashboard group; the **node** is the unit of scheduling / execution. Deps are intra-DAG edges only (`AfterOk` by default: the dep must succeed, a failed/cancelled dep cancels the dependent — cancel-downstream). Cross-DAG ordering comes from the per-agent lease, never from edges between DAGs. @@ -264,13 +263,11 @@ That endpoint serves the graph generically (`hive-jobq-wire`): every node carries `id`, `parent`, `deps`, `state`, `label` (the node kind's own wire string — the kind _is_ the phase label, there is no separate sub-step string) and free-form `data` for what only some kinds have (`agent`, -`inputs`, `build_log_id`). Group roots ride as ordinary +`inputs`). Group roots ride as ordinary nodes, so a group's state is just the root's own `state`. No group-level `agent` exists — agent is per-node, so one group can -span agents; consumers derive a group's agents from its nodes. Build logs -are likewise **per-node**: the dashboard renders the node tree and keys the -live-log panel off the running node. +span agents; consumers derive a group's agents from its nodes. The event carries no payload by design: shipping a typed projection of the whole queue in the event itself would be a second rendering of the diff --git a/docs/4-implementation/trust-boundary/hive-c0re-privilege.md b/docs/4-implementation/trust-boundary/hive-c0re-privilege.md index f0e5f5c31..2c2c4cd6d 100644 --- a/docs/4-implementation/trust-boundary/hive-c0re-privilege.md +++ b/docs/4-implementation/trust-boundary/hive-c0re-privilege.md @@ -32,7 +32,7 @@ known operations; there is no arbitrary command pass-through: | ------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `StartContainer` / `StopContainer` | `nixos-container start/stop ` | | `KillContainer` | `machinectl kill --signal=SIGKILL` (`nixos-container` has no kill verb) | -| `StartFailureLog` | `journalctl _SYSTEMD_UNIT=container@h-.service --since @ -n 30 -o cat --no-pager`, fixed argv, 5 s timeout, output capped at 8 KiB; hive-priv clamps `` to the last hour. hive-c0re appends the lines to the failed start's build log, which only the dashboard serves. | +| `StartFailureLog` | `journalctl _SYSTEMD_UNIT=container@h-.service --since @ -n 30 -o cat --no-pager`, fixed argv, 5 s timeout, output capped at 8 KiB; hive-priv clamps `` to the last hour. hive-c0re logs the lines under the failed start's `nixos-container` tag. | | `CreateContainer` / `UpdateContainer` | `nixos-container create --system-path `; for an update, `nix-env -p --set ` and a reload of the unit. `` must already be in the store with a signature by the swarm cache's key: hive-priv first runs `nix store copy-sigs --store daemon` for that path from the cache named in `/run/hyperhive-nix/cache.conf`, with the netrc beside it, which imports the cache's signature for a path that was already in the store (on a failure it logs a warning and the check goes on); it then runs `nix store verify --no-contents --sigs-needed 1` with `trusted-public-keys` set to the key in root-owned `/run/hyperhive-priv/attic/public-key`, and refuses a path that's absent or that the key didn't sign. hive-priv builds nothing. The check doesn't bind the path to the agent whose pull produced it: hive-c0re could already pull any swarm-signed path and apply it to any agent, so the check bounds the same risk (one agent gets another's closure), not an arbitrary-path one | | `RealiseClosure` | Kept for one release, for an older hive-c0re that still asks hive-priv to download. Downloads nothing: answers `Ok` when the path is in the store and signed by the swarm key (what `CreateContainer` needs), else `not available: `. The path must be `/nix/store/<32 chars>-` and the agent name valid | | `RefreshCacheFacts` | `systemctl start hive-priv-cache-facts.service`, waited for up to 60 s. The request names no unit and has no field. hive-c0re asks for it when the fragment is missing or lacks a setting | diff --git a/docs/4-implementation/web-ui/dashboard.md b/docs/4-implementation/web-ui/dashboard.md index 574a9e6f1..ad68093be 100644 --- a/docs/4-implementation/web-ui/dashboard.md +++ b/docs/4-implementation/web-ui/dashboard.md @@ -433,19 +433,14 @@ below — some endpoints aren't in it yet. `{ "marked": N }`. Agent name validated against `[a-z0-9_-]`, 1-63 chars; 400 on bad input. -- `GET /api/dashboard/stream` — unified live event channel: - broker `sent` / `delivered`, plus the mutation events listed - below. +- `GET /api/dashboard/stream` — unified live event channel: the + mutation events listed below. ## Dashboard event channel Wire vocabulary on `/api/dashboard/stream` (kind tag is in the JSON payload): -- `sent` / `delivered` — broker traffic, mirrored from the - intra-process channel by a forwarder task. Both carry `id: i64` - (the broker row id) and `in_reply_to: Option` for thread - rendering. - `transient_set` (name, transient_kind, since_unix) / `transient_cleared` (name) — lifecycle action spinners. The client ticks the elapsed-seconds badge off `since_unix` diff --git a/docs/4-implementation/web-ui/shape.md b/docs/4-implementation/web-ui/shape.md index 9fc5bd5aa..2c65876ea 100644 --- a/docs/4-implementation/web-ui/shape.md +++ b/docs/4-implementation/web-ui/shape.md @@ -52,9 +52,8 @@ domain would exhaust this budget on a live hive; dispatching by `kind` on the client is a one-liner. Per-domain splits are reserved for high-volume sub-streams most consumers skip (none - exist yet). The broker's intra-process channel stays separate - from the dashboard channel to avoid coupling `recv_blocking_batch` - (hot path inside the harness turn loop) to presentation concerns. + exist yet). Broker messages don't ride it: the broker's intra-process + channel only wakes `recv_blocking_batch`. **SSE multiplexing**: the dashboard uses a `SharedWorker` (`stream-worker.js`) to hold one upstream `EventSource` per URL. All same-origin tabs share this worker diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index 9a4dac522..10057f5f3 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -22,10 +22,6 @@ const DASHBOARD_CHANNEL: usize = 256; pub struct Coordinator { pub broker: Arc, - /// Full build-log capture. `lifecycle::run` `start()`s a row per attempt, - /// pipes every stdout/stderr line into it, and `finish()`es it on - /// child exit. See `build_logs.rs` for retention. - pub build_logs: Arc, /// Store path of the hyperhive flake source (no fragment). Used to /// identify the running revision (`current_flake_rev`). pub hyperhive_flake: String, @@ -110,9 +106,8 @@ pub struct Coordinator { /// publisher sends the change at once instead of at its next sweep. updating_changed: tokio::sync::Notify, /// Unified wire-facing event channel feeding the dashboard SSE - /// stream. Carries broker messages (mirrored from `broker.subscribe` - /// by the forwarder task in `main.rs`) and dashboard-only mutation - /// events (container state, rebuild queue, etc.). + /// stream. Carries the dashboard mutation events (container state, + /// rebuild queue, etc.). dashboard_events: broadcast::Sender, /// Last container snapshot seen by `rescan_containers_and_emit`, @@ -331,33 +326,11 @@ impl Coordinator { agent_io_weight, } = env; let broker = Broker::open(db_path).context("open broker")?; - // BuildLogs wants a directory (it picks its own `build_logs.sqlite` - // file under it); every other opener here takes the sibling - // sqlite-file path itself. Derive the dir from `db_path`'s - // parent so the two shapes line up. - let build_logs_dir = db_path.parent().unwrap_or_else(|| Path::new(".")); - let build_logs = Arc::new( - crate::build_logs::BuildLogs::open(build_logs_dir).context("open build_logs")?, - ); - // This process is the database's only writer and has started no - // build yet, so a row still in-flight belongs to a dead process. - let reaped = build_logs.reap_orphans()?; - if reaped > 0 { - tracing::warn!( - reaped, - "build_logs: closed rows left in-flight by a previous run" - ); - } - // Install the process-wide handle so `lifecycle::run` can write - // without us having to thread an `Arc` through every public entry - // point in the lifecycle surface. - crate::build_logs::install(build_logs.clone()); let power = Arc::new(crate::power::PowerStore::default()); let (dashboard_events, _) = broadcast::channel(DASHBOARD_CHANNEL); let (shutdown_tx, _) = watch::channel(false); Ok(Self { broker: Arc::new(broker), - build_logs, hyperhive_flake, claude_code_path, dashboard_port, @@ -447,8 +420,7 @@ impl Coordinator { } /// Subscribe to the unified dashboard event channel. Used by the - /// `/dashboard/stream` SSE handler and by the broker-to-dashboard - /// forwarder task. + /// `/dashboard/stream` SSE handler. pub fn dashboard_subscribe(&self) -> broadcast::Receiver { self.dashboard_events.subscribe() } @@ -1004,29 +976,3 @@ mod agent_paths_tests { assert_eq!(paths.agent, std::path::PathBuf::from("/run/x")); } } - -#[cfg(test)] -mod open_tests { - use super::{Coordinator, HiveEnv}; - - fn open(db: &std::path::Path) -> Coordinator { - Coordinator::open(db, HiveEnv::default(), 1).expect("open coordinator") - } - - /// A build log row the previous process left open is closed as failed - /// when the next one opens the coordinator. - #[tokio::test] - async fn open_closes_build_logs_left_in_flight() { - let dir = tempfile::tempdir().expect("tempdir"); - let db = dir.path().join("broker.sqlite"); - let id = open(&db) - .build_logs - .start("alice", "run", "cmd", None) - .expect("start"); - - let coord = open(&db); - let row = coord.build_logs.get_full(id).expect("get").expect("Some"); - assert_eq!(row.status.as_deref(), Some("fail")); - assert!(row.finished_at.is_some()); - } -} diff --git a/hive-c0re/src/dashboard_events.rs b/hive-c0re/src/dashboard_events.rs index af896f3f9..32947d17a 100644 --- a/hive-c0re/src/dashboard_events.rs +++ b/hive-c0re/src/dashboard_events.rs @@ -1,39 +1,16 @@ //! Unified dashboard event channel — all near-real-time browser events //! flow through `Coordinator.dashboard_events`. Design rationale -//! (single channel, broker forwarder isolation): +//! (single channel): //! `docs/4-implementation/web-ui/shape.md::One unified channel`. use serde::Serialize; use crate::container_view::ContainerView; use crate::dashboard::TombstoneView; -use chrono::{DateTime, Utc}; #[derive(Debug, Clone, Serialize)] #[serde(rename_all = "snake_case", tag = "kind")] pub enum DashboardEvent { - /// Broker `Sent` event mirrored onto the dashboard channel. - Sent { - /// Broker row id. Allows the dashboard to track reply threads. - id: i64, - from: String, - to: String, - body: String, - at: DateTime, - #[serde(default, skip_serializing_if = "Option::is_none")] - in_reply_to: Option, - }, - /// Broker `Delivered` event mirrored onto the dashboard channel. - Delivered { - /// Broker row id. Allows the dashboard to track reply threads. - id: i64, - from: String, - to: String, - body: String, - at: DateTime, - #[serde(default, skip_serializing_if = "Option::is_none")] - in_reply_to: Option, - }, /// A lifecycle action started for an agent (spawn / start / stop /// / restart / rebuild / destroy). Clients render a spinner next /// to the row; the client computes "seconds in this state" @@ -107,7 +84,7 @@ pub enum DashboardEvent { impl DashboardEvent { /// Snake-case identifier matching this variant's serde `tag` - /// (e.g. `Sent` → `"sent"`, `ContainerStateChanged` → + /// (e.g. `TransientSet` → `"transient_set"`, `ContainerStateChanged` → /// `"container_state_changed"`). Lets `/dashboard/stream`'s /// `?kinds=` filter decide whether to forward a frame without /// paying the JSON-serialise cost first. @@ -119,8 +96,6 @@ impl DashboardEvent { #[must_use] pub fn kind_tag(&self) -> &'static str { match self { - DashboardEvent::Sent { .. } => "sent", - DashboardEvent::Delivered { .. } => "delivered", DashboardEvent::TransientSet { .. } => "transient_set", DashboardEvent::TransientCleared { .. } => "transient_cleared", DashboardEvent::ContainerStateChanged { .. } => "container_state_changed", @@ -154,22 +129,6 @@ mod tests { )] fn kind_tag_matches_serde_kind_field() { let samples: Vec = vec![ - DashboardEvent::Sent { - id: 1, - from: "a".into(), - to: "b".into(), - body: String::new(), - at: hive_sh4re::wire_time::from_secs(0), - in_reply_to: None, - }, - DashboardEvent::Delivered { - id: 1, - from: "a".into(), - to: "b".into(), - body: String::new(), - at: hive_sh4re::wire_time::from_secs(0), - in_reply_to: None, - }, DashboardEvent::TransientSet { name: "x".into(), transient_kind: "rebuilding".into(), diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index 0eaae9801..28999f5e2 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -61,7 +61,6 @@ pub(super) trait Host: Sync { hive: &HiveEnv, paths: &AgentPaths, toplevel: &str, - node_id: u64, ) -> Result<()>; /// Write or remove the agent's pause marker. async fn set_paused(&self, agent: &hive_types::Ident, paused: bool) -> Result<()>; @@ -139,9 +138,8 @@ impl Host for RealHost { hive: &HiveEnv, paths: &AgentPaths, toplevel: &str, - node_id: u64, ) -> Result<()> { - crate::lifecycle::swap_update(name, hive, paths, toplevel, Some(node_id)).await + crate::lifecycle::swap_update(name, hive, paths, toplevel).await } async fn set_paused(&self, agent: &hive_types::Ident, paused: bool) -> Result<()> { @@ -270,7 +268,7 @@ async fn dispatch( let result = match kind { NodeKind::MetaSync { .. } => run_meta_sync(host, agent).await, NodeKind::Prebuild { .. } => run_prebuild(host, agent).await, - NodeKind::Swap { .. } => run_swap(host, coord, agent, id).await, + NodeKind::Swap { .. } => run_swap(host, coord, agent).await, NodeKind::RebuildBookkeeping { .. } => run_rebuild_bookkeeping(coord).await, NodeKind::Provision { .. } => run_provision(host, agent).await, NodeKind::Create { .. } => run_create(host, agent).await, @@ -736,12 +734,7 @@ fn without_closure(name: &str, step: &str, why: &NoClosure) -> anyhow::Error { /// kick, rescan) is the next node. /// The recovery-start on failure is NOT here — the DAG's tail /// `Reconcile` runs after this node terminal ok *or* fail. -async fn run_swap( - host: &impl Host, - coord: &Arc, - name: &str, - id: NodeId, -) -> Result<()> { +async fn run_swap(host: &impl Host, coord: &Arc, name: &str) -> Result<()> { // Swap runs on an already-existing (stopped) container — runtime dir // and listener were created earlier. Pure path accessor suffices. let agent_dir = crate::paths::agent_runtime_dir(name); @@ -749,9 +742,7 @@ async fn run_swap( let paths = Coordinator::agent_paths(name, agent_dir)?; let closure = realise_pinned(host, name, "swap").await?; tracing::info!(%name, "swap: applying {closure}"); - let result = host - .swap_update(name, &hive, &paths, &closure, id.get()) - .await; + let result = host.swap_update(name, &hive, &paths, &closure).await; // The marker names what was applied; a failed write is not a failed swap // (the container runs the closure), it only means the next comparison // sees the declaration as new. diff --git a/hive-c0re/src/job_queue/exec/tests.rs b/hive-c0re/src/job_queue/exec/tests.rs index 12c0ee69e..389e8a1e8 100644 --- a/hive-c0re/src/job_queue/exec/tests.rs +++ b/hive-c0re/src/job_queue/exec/tests.rs @@ -7,7 +7,6 @@ use std::path::PathBuf; use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; -use hive_jobq::NodeId; use hive_jobq::scheduler::Scheduler; use super::{ @@ -169,9 +168,8 @@ impl Host for FakeHost { _hive: &HiveEnv, _paths: &AgentPaths, toplevel: &str, - node_id: u64, ) -> anyhow::Result<()> { - self.record(format!("swap_update {name} {toplevel} node={node_id}")); + self.record(format!("swap_update {name} {toplevel}")); if self.fails("swap_update") { anyhow::bail!("nixos-container update failed"); } @@ -288,20 +286,6 @@ fn coordinator() -> (tempfile::TempDir, Arc) { (dir, Arc::new(coord)) } -/// A real node id, taken from a one-node job in a throwaway queue. -fn node_id() -> NodeId { - let q = crate::job_queue::JobQueue::new(1); - q.insert_job(|b| { - vec![ - b.node(NodeKind::AgentWindow { - agent: "alice".to_owned(), - }) - .guid(), - ] - }) - .expect("valid shape")[0] -} - // --- Prebuild --------------------------------------------------------------- /// What a block logged, level and text: the message and each field. @@ -456,13 +440,12 @@ async fn a_stopped_container_looks_up_nothing() { async fn swap_applies_the_swarms_closure() { let (_dir, coord) = coordinator(); let host = FakeHost::down(); - let id = node_id(); - run_swap(&host, &coord, "alice", id).await.expect("swap ok"); + run_swap(&host, &coord, "alice").await.expect("swap ok"); assert_eq!( host.calls(), [ format!("realise alice {CLOSURE}"), - format!("swap_update alice {CLOSURE} node={}", id.get()), + format!("swap_update alice {CLOSURE}"), format!("mark_deployed alice {CLOSURE}"), "unpin_closure alice".to_owned(), ] @@ -475,7 +458,7 @@ async fn swap_applies_the_swarms_closure() { async fn a_swap_without_a_closure_never_updates_and_logs_an_error() { for (case, host, why) in without_a_pinned_closure() { let (_dir, coord) = coordinator(); - let (result, log) = logging(run_swap(&host, &coord, "alice", node_id())).await; + let (result, log) = logging(run_swap(&host, &coord, "alice")).await; result.expect_err(case); let errors = log.at(tracing::Level::ERROR); assert_eq!(errors.len(), 1, "{case}: {errors:?}"); @@ -496,8 +479,7 @@ async fn a_swap_without_a_closure_never_updates_and_logs_an_error() { async fn swap_marks_the_applied_closure_deployed_only_after_the_apply() { let (_dir, coord) = coordinator(); let host = FakeHost::down(); - let id = node_id(); - run_swap(&host, &coord, "alice", id).await.expect("swap ok"); + run_swap(&host, &coord, "alice").await.expect("swap ok"); assert_eq!( host.calls()[2], format!("mark_deployed alice {CLOSURE}"), @@ -506,7 +488,7 @@ async fn swap_marks_the_applied_closure_deployed_only_after_the_apply() { ); let host = FakeHost::down().failing("swap_update"); - run_swap(&host, &coord, "alice", node_id()) + run_swap(&host, &coord, "alice") .await .expect_err("apply fails"); assert!( @@ -522,7 +504,7 @@ async fn swap_marks_the_applied_closure_deployed_only_after_the_apply() { async fn a_marker_that_cannot_be_written_does_not_fail_the_swap() { let (_dir, coord) = coordinator(); let host = FakeHost::down().failing("mark_deployed"); - let (result, log) = logging(run_swap(&host, &coord, "alice", node_id())).await; + let (result, log) = logging(run_swap(&host, &coord, "alice")).await; result.expect("swap ok"); assert_eq!(log.at(tracing::Level::WARN).len(), 1); } @@ -533,7 +515,7 @@ async fn a_marker_that_cannot_be_written_does_not_fail_the_swap() { async fn a_failed_swap_reports_the_error_and_touches_no_container() { let (_dir, coord) = coordinator(); let host = FakeHost::down().failing("swap_update"); - let err = run_swap(&host, &coord, "alice", node_id()) + let err = run_swap(&host, &coord, "alice") .await .expect_err("update failure propagates"); assert!(format!("{err:#}").contains("update failed"), "{err:#}"); @@ -882,8 +864,7 @@ async fn a_record_that_changes_mid_rebuild_does_not_change_what_swap_applies() { run_meta_sync(&host, "alice").await.expect("pinned"); run_prebuild(&host, "alice").await.expect("prebuilt"); host.builder_records(NEWER); - let id = node_id(); - run_swap(&host, &coord, "alice", id).await.expect("swapped"); + run_swap(&host, &coord, "alice").await.expect("swapped"); let calls = host.calls(); assert_eq!( calls @@ -894,7 +875,7 @@ async fn a_record_that_changes_mid_rebuild_does_not_change_what_swap_applies() { "the record is read once: {calls:?}" ); assert!( - calls.contains(&format!("swap_update alice {CLOSURE} node={}", id.get())), + calls.contains(&format!("swap_update alice {CLOSURE}")), "{calls:?}" ); assert!( diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 89f60bf77..92d76661b 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -93,10 +93,8 @@ pub struct RunningTransient { /// [`Graph`] holds every job's nodes. /// /// There is deliberately **no wrapper struct and no per-node side map**. The -/// last map held the `build_logs` row id; that link now lives on the log row -/// itself (`build_logs.node_id`). With nothing else to guard, the mutex holds -/// the scheduler *directly* — which is what lets `hive_jobq` drive the run loop -/// (it takes `&Arc>>`, a type a host-side wrapper could not +/// mutex holds the scheduler *directly* — which is what lets `hive_jobq` drive +/// the run loop (it takes `&Arc>>`, a type a host-side wrapper could not /// satisfy). type Sched = Scheduler; diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index 9f3850c7f..4e1b5c3cb 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -6,7 +6,7 @@ //! no second, typed view of it any more. //! //! **One level, not two.** The node is the unit of everything: scheduling, -//! execution, build-log, cancel, and the dashboard group (a group root's +//! execution, cancel, and the dashboard group (a group root's //! subtree *is* the group). A DAG used to be a second level above it, with //! its own store and its own id; there is no container node any more, so a //! job is exactly the nodes it declared. See `docs/4-implementation/scheduler/coordinator.md::Job queue` @@ -116,18 +116,12 @@ impl hive_jobq_wire::WireNode for NodeKind { <&str>::from(self).to_owned() } - fn data(&self, id: hive_jobq_wire::WireId) -> serde_json::Value { + fn data(&self, _id: hive_jobq_wire::WireId) -> serde_json::Value { let mut data = serde_json::Map::new(); let agent = self.agent(); if !agent.is_empty() { data.insert("agent".to_owned(), agent.into()); } - - // Not in the payload at all — the build log is keyed on node identity - // in a side table, which is why `data` is handed the id. - if let Some(log) = crate::build_logs::global().and_then(|h| h.id_for_node(id)) { - data.insert("build_log_id".to_owned(), log.into()); - } if data.is_empty() { serde_json::Value::Null } else { diff --git a/hive-c0re/src/job_queue/resource.rs b/hive-c0re/src/job_queue/resource.rs index 58c7047e1..dfa13a373 100644 --- a/hive-c0re/src/job_queue/resource.rs +++ b/hive-c0re/src/job_queue/resource.rs @@ -32,20 +32,3 @@ pub enum Resource { /// `.git/index.lock` and the remote-tracking ref lock. KnowledgeTree, } - -/// This resource's name on the generic graph wire. -/// -/// `hive_jobq` is generic over the resource type, so a viewer that can render -/// any graph gets a string here rather than this enum. The `agent:` prefix -/// keeps the per-agent leases from colliding with a hypothetical global -/// resource that happens to share an agent's name. -impl hive_jobq_wire::WireResource for Resource { - fn name(&self) -> String { - match self { - Resource::BuildSlot => "build-slot".to_owned(), - Resource::Agent(agent) => format!("agent:{agent}"), - - Resource::KnowledgeTree => "knowledge-tree".to_owned(), - } - } -} diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index 13ca9c530..e9c43a080 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -1165,11 +1165,6 @@ fn cancelled_power_op_runs_no_compensating_node() { } // ---- history ---- -// -// The node → build-log link is no longer queue state: the log row carries -// `node_id` and the lookup lives in `stores::build_logs` (see -// `node_link_survives_completion_and_newest_wins` there). Nothing in the queue -// needs testing for it any more, which is the point of that move. /// History retention is a **flat** newest-first cap over all terminal DAGs /// (`MAX_HISTORY_DAGS`), not a per-template bucket behind a grace window. diff --git a/hive-c0re/src/lifecycle/mod.rs b/hive-c0re/src/lifecycle/mod.rs index 64dcd013f..53dd27614 100644 --- a/hive-c0re/src/lifecycle/mod.rs +++ b/hive-c0re/src/lifecycle/mod.rs @@ -222,7 +222,7 @@ pub async fn provision_container(name: &str, paths: &AgentPaths) -> Result<()> { /// Assumes `provision_container` already registered the agent. /// `toplevel` is the store path of the system closure to create it from. pub async fn create_only(name: &str, toplevel: &str) -> Result<()> { - priv_run_inner("create", name, Some(toplevel), None).await + priv_run_inner("create", name, Some(toplevel)).await } /// Rebuild-path preamble, run by the job queue's `Prebuild` node: fail fast on @@ -252,10 +252,9 @@ pub async fn swap_update( hive: &HiveEnv, paths: &AgentPaths, toplevel: &str, - node_id: Option, ) -> Result<()> { write_dropins(name, hive, paths).await?; - priv_run_inner("update", name, Some(toplevel), node_id).await + priv_run_inner("update", name, Some(toplevel)).await } /// Build the `AgentSpec` list for [`crate::agent_set::refresh`] from @@ -728,22 +727,19 @@ fn is_actionable_nix_line(line: &str) -> bool { /// What every log line of one container operation carries, so the swarm log /// store can group a build's lines and the builds board can list the builds: -/// `build_id` (the build-log row's id, empty when the store is not installed), /// `agent` and `kind`. A `start` line opens the build and a `finish` line /// closes it with its `status` and `runtime_secs`; the output lines between /// them have no `event`. #[derive(Clone)] struct BuildTag { - build_id: String, agent: String, kind: String, cmdline: String, } impl BuildTag { - fn new(log_id: Option, agent: &str, kind: &str, cmdline: &str) -> Self { + fn new(agent: &str, kind: &str, cmdline: &str) -> Self { Self { - build_id: log_id.map(|id| id.to_string()).unwrap_or_default(), agent: agent.to_owned(), kind: kind.to_owned(), cmdline: cmdline.to_owned(), @@ -754,7 +750,6 @@ impl BuildTag { tracing::info!( target: "nixos-container", event = "start", - build_id = %self.build_id, agent = %self.agent, kind = %self.kind, cmdline = %self.cmdline, @@ -771,7 +766,6 @@ impl BuildTag { if actionable { tracing::warn!( target: "nixos-container", - build_id = %self.build_id, agent = %self.agent, kind = %self.kind, cmdline = %self.cmdline, @@ -780,7 +774,6 @@ impl BuildTag { } else { tracing::info!( target: "nixos-container", - build_id = %self.build_id, agent = %self.agent, kind = %self.kind, cmdline = %self.cmdline, @@ -795,7 +788,6 @@ impl BuildTag { tracing::info!( target: "nixos-container", event = "finish", - build_id = %self.build_id, agent = %self.agent, kind = %self.kind, status, @@ -808,7 +800,6 @@ impl BuildTag { tracing::warn!( target: "nixos-container", event = "finish", - build_id = %self.build_id, agent = %self.agent, kind = %self.kind, status, @@ -822,46 +813,22 @@ impl BuildTag { } /// Build the per-line callback for `create_container_streaming` / -/// `update_container_streaming`. Both ops share identical dispatch logic -/// (log the line through its `BuildTag`, then `append_stdout` or -/// `append_stderr`); this helper avoids duplicating that match body across the -/// two call sites. -fn make_log_callback( - logs: Option>, - log_id: Option, - tag: BuildTag, -) -> impl FnMut(hive_priv_sock::PrivStream, &str) { - use hive_priv_sock::PrivStream; - move |stream, line| { - tag.line(stream, line); - if let (Some(h), Some(id)) = (&logs, log_id) { - match stream { - PrivStream::Stdout => h.append_stdout(id, line), - PrivStream::Stderr => h.append_stderr(id, line), - } - } - } +/// `update_container_streaming`: log each line through its `BuildTag`. +fn make_log_callback(tag: BuildTag) -> impl FnMut(hive_priv_sock::PrivStream, &str) { + move |stream, line| tag.line(stream, line) } -/// Put a failed non-streaming op's error into its build-log row, which has no -/// other output: `start` runs through `call`, not the streaming protocol. The -/// error carries `nixos-container`'s stderr; `journal` is the container -/// unit's own log from hive-priv. An `Err` there (an older hive-priv does not -/// know the op) leaves the stderr alone. -fn record_failure( - logs: Option<&crate::build_logs::BuildLogs>, - log_id: Option, - err: &anyhow::Error, - journal: Result, -) { - if let (Some(h), Some(id)) = (logs, log_id) { - for line in format!("{err:#}").lines() { - h.append_stderr(id, line); - } - match journal { - Ok(text) => text.lines().for_each(|l| h.append_stderr(id, l)), - Err(e) => tracing::debug!(error = ?e, "start journal unavailable"), - } +/// Log a failed `start`'s error and the container unit's own journal (from +/// hive-priv) line by line. An `Err` journal (an older hive-priv does not know +/// the op) leaves just the error. +fn record_failure(tag: &BuildTag, err: &anyhow::Error, journal: Result) { + use hive_priv_sock::PrivStream; + for line in format!("{err:#}").lines() { + tag.line(PrivStream::Stderr, line); + } + match journal { + Ok(text) => text.lines().for_each(|l| tag.line(PrivStream::Stderr, l)), + Err(e) => tracing::debug!(error = ?e, "start journal unavailable"), } } @@ -872,42 +839,21 @@ fn missing_toplevel(kind: &str, name: &str) -> anyhow::Error { anyhow::anyhow!("{name}: {kind} needs a closure to apply") } -/// Execute a container operation via hive-priv and integrate with -/// `build_logs.sqlite`. hive-priv runs as root and forwards output lines -/// to hive-c0re in real time via the streaming priv protocol. Each line -/// is appended to the build-log row as it arrives, so the dashboard -/// shows live progress during long `nixos-container create` / `update` runs. +/// Execute a container operation via hive-priv. hive-priv runs as root and +/// forwards output lines to hive-c0re in real time via the streaming priv +/// protocol; each line is logged through the op's `BuildTag`. async fn priv_run(kind: &str, name: &str) -> Result<()> { - priv_run_inner(kind, name, None, None).await + priv_run_inner(kind, name, None).await } -/// Like `priv_run` but stamps `node_id` onto the build-log row it opens, so -/// the log can be found from the queue node. -/// -/// This used to be a `Fn(i64)` callback that handed the row id *back* to the -/// queue, which then held it in a side map. The row carries the link itself -/// now, so the id only ever travels one way. -async fn priv_run_inner( - kind: &str, - name: &str, - toplevel: Option<&str>, - node_id: Option, -) -> Result<()> { +/// [`priv_run`] with the closure a `create` or `update` applies. +async fn priv_run_inner(kind: &str, name: &str, toplevel: Option<&str>) -> Result<()> { let container = container_name(name); let cmdline = format!("nixos-container {kind} {container}"); - let logs = crate::build_logs::global(); - let log_id = logs.as_ref().and_then(|h| { - h.start(name, kind, &cmdline, node_id) - .map_err(|e| { - tracing::warn!(error = ?e, "build_logs: start failed (priv_run log dropped)"); - }) - .ok() - }); - - // For long-running ops use the streaming protocol so build_logs - // receives lines in real time rather than as a batch at completion. - let tag = BuildTag::new(log_id, name, kind, &cmdline); + // Long-running ops use the streaming protocol so lines are logged in real + // time rather than as a batch at completion. + let tag = BuildTag::new(name, kind, &cmdline); tag.started(); let began = std::time::Instant::now(); let result: Result<()> = match kind { @@ -916,7 +862,7 @@ async fn priv_run_inner( crate::priv_client::create_container_streaming( name, toplevel, - make_log_callback(logs.clone(), log_id, tag.clone()), + make_log_callback(tag.clone()), ) .await } @@ -927,7 +873,7 @@ async fn priv_run_inner( crate::priv_client::update_container_streaming( name, toplevel, - make_log_callback(logs.clone(), log_id, tag.clone()), + make_log_callback(tag.clone()), ) .await } @@ -940,7 +886,7 @@ async fn priv_run_inner( let started = crate::priv_client::start_container(name).await; if let Err(e) = &started { let journal = crate::priv_client::start_failure_log(name, attempt).await; - record_failure(logs.as_deref(), log_id, e, journal); + record_failure(&tag, e, journal); } started } @@ -952,22 +898,6 @@ async fn priv_run_inner( let succeeded = result.is_ok(); tag.finished(succeeded, began.elapsed()); - if let (Some(h), Some(id)) = (&logs, log_id) { - h.finish( - id, - if succeeded { - crate::build_logs::BuildStatus::Ok - } else { - crate::build_logs::BuildStatus::Fail - }, - ); - } - match result { - Ok(()) => Ok(()), - Err(e) => match log_id { - Some(id) => bail!("{e:#}; see build log #{id}"), - None => bail!("{e:#}"), - }, - } + result.map_err(|e| anyhow::anyhow!("{e:#}")) } diff --git a/hive-c0re/src/lifecycle/tests.rs b/hive-c0re/src/lifecycle/tests.rs index 2e6359256..9e5da826b 100644 --- a/hive-c0re/src/lifecycle/tests.rs +++ b/hive-c0re/src/lifecycle/tests.rs @@ -292,45 +292,57 @@ fn hive_facts_leave_unset_values_out() { assert_eq!(facts.name, None); } -/// A failed `start` leaves its error in the op's build-log row; without it the -/// row is empty. A missing row id records nothing and does not panic. +/// A failed `start` logs its error and the container unit's journal, line by +/// line, under the op's tag. #[test] -fn a_failed_start_puts_its_error_in_the_op_output() { - let dir = tempfile::tempdir().expect("tempdir"); - let logs = crate::build_logs::BuildLogs::open(dir.path()).expect("open"); - let id = logs - .start("updo", "start", "nixos-container start h-updo", None) - .expect("start"); +fn a_failed_start_logs_its_error_and_the_journal() { + let events = Events::default(); + let _guard = tracing::subscriber::set_default(events.clone()); + let tag = BuildTag::new("updo", "start", "nixos-container start h-updo"); let err = anyhow::anyhow!("nixos-container start h-updo failed (exit status: 1): Job failed"); let journal = Ok("Creating mount point failed: Read-only file system".to_owned()); - record_failure(Some(&logs), Some(id), &err, journal); - let full = logs.get_full(id).expect("get").expect("row"); - assert!(full.stderr.contains("Job failed"), "{}", full.stderr); + record_failure(&tag, &err, journal); + let logged = events.0.lock().unwrap(); + let messages: Vec<_> = logged + .iter() + .filter_map(|(_, f)| field(f, "message")) + .collect(); + assert_eq!(messages.len(), 2, "{messages:?}"); + assert!(messages[0].contains("Job failed"), "{messages:?}"); assert!( - full.stderr.contains("Read-only file system"), - "{}", - full.stderr + messages[1].contains("Read-only file system"), + "{messages:?}" ); - record_failure(Some(&logs), None, &err, Ok(String::new())); - record_failure(None, Some(id), &err, Ok(String::new())); + for (_, fields) in logged.iter() { + assert_eq!(field(fields, "agent"), Some("updo")); + } } -/// An older hive-priv rejects the op; the start's own error is still recorded. +/// An older hive-priv rejects the journal op; the start's own error is still +/// logged and the rejection is not. #[test] fn an_unknown_journal_op_keeps_the_start_error() { - let dir = tempfile::tempdir().expect("tempdir"); - let logs = crate::build_logs::BuildLogs::open(dir.path()).expect("open"); - let id = logs - .start("updo", "start", "nixos-container start h-updo", None) - .expect("start"); + let events = Events::default(); + let _guard = tracing::subscriber::set_default(events.clone()); + let tag = BuildTag::new("updo", "start", "nixos-container start h-updo"); let err = anyhow::anyhow!("Job failed"); let unknown = Err(anyhow::anyhow!( "parse request: unknown variant `start_failure_log`" )); - record_failure(Some(&logs), Some(id), &err, unknown); - let full = logs.get_full(id).expect("get").expect("row"); - assert!(full.stderr.contains("Job failed"), "{}", full.stderr); - assert!(!full.stderr.contains("unknown variant"), "{}", full.stderr); + record_failure(&tag, &err, unknown); + let logged = events.0.lock().unwrap(); + let messages: Vec<_> = logged + .iter() + .filter_map(|(_, f)| field(f, "message")) + .collect(); + assert!( + messages.iter().any(|m| m.contains("Job failed")), + "{messages:?}" + ); + assert!( + !messages.iter().any(|m| m.contains("unknown variant")), + "{messages:?}" + ); } /// One logged event: its level and each field as `name=value`. @@ -372,16 +384,16 @@ fn field<'a>(fields: &'a [(String, String)], name: &str) -> Option<&'a str> { .map(|(_, v)| v.as_str()) } -/// A build's log lines say which build they belong to, so the log store can -/// group them: a `start` line, the output lines, then a `finish` line with the +/// A build's log lines carry its agent and kind, so the log store can group +/// them: a `start` line, the output lines, then a `finish` line with the /// status and runtime. A nix error is a warning, chatter stays info, and a /// failed finish is a warning. #[test] -fn a_build_logs_its_start_its_lines_and_its_finish_under_one_build_id() { +fn a_build_logs_its_start_its_lines_and_its_finish() { use hive_priv_sock::PrivStream; let events = Events::default(); let _guard = tracing::subscriber::set_default(events.clone()); - let tag = BuildTag::new(Some(7), "alice", "update", "nixos-container update h-alice"); + let tag = BuildTag::new("alice", "update", "nixos-container update h-alice"); tag.started(); tag.line(PrivStream::Stdout, "copying 3 paths"); tag.line(PrivStream::Stderr, "error: builder failed"); @@ -390,7 +402,6 @@ fn a_build_logs_its_start_its_lines_and_its_finish_under_one_build_id() { let logged = events.0.lock().unwrap(); assert_eq!(logged.len(), 4); for (_, fields) in logged.iter() { - assert_eq!(field(fields, "build_id"), Some("7")); assert_eq!(field(fields, "agent"), Some("alice")); assert_eq!(field(fields, "kind"), Some("update")); } @@ -411,18 +422,6 @@ fn a_build_logs_its_start_its_lines_and_its_finish_under_one_build_id() { assert_eq!(field(&logged[3].1, "runtime_secs"), Some("12")); } -/// Without the build-log store there is no id; the lines still carry the -/// agent and kind, with an empty `build_id`. -#[test] -fn a_build_without_a_log_row_has_an_empty_build_id() { - let events = Events::default(); - let _guard = tracing::subscriber::set_default(events.clone()); - BuildTag::new(None, "alice", "create", "nixos-container create h-alice").started(); - let logged = events.0.lock().unwrap(); - assert_eq!(field(&logged[0].1, "build_id"), Some("")); - assert_eq!(field(&logged[0].1, "agent"), Some("alice")); -} - #[test] fn stop_kind_is_leaked_once_per_agent() { let first = stop_kind("stopkind-a"); diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index 59ff2eb76..a3d1e7a37 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -46,7 +46,7 @@ mod workers; pub(crate) use agent_config::{archetypes, limits, permissions, resource_limits, topology}; pub(crate) use stats::{container_stats, host_stats, otel_metrics, warnings}; -pub(crate) use stores::{broker, build_logs, db, power}; +pub(crate) use stores::{broker, db, power}; pub(crate) use workers::{agent_sockets, auto_update, crash_watch, knowledge, mcp_sockets}; use coordinator::{Coordinator, HiveEnv, ServeConfig}; @@ -218,7 +218,7 @@ async fn cmd_serve( ) -> Result<()> { // Move any host-side state still at the legacy flat layout into its // subdir (`db/`, `forge/`, `run/`) BEFORE opening the - // broker db — the broker + build-logs dbs are among the relocated + // broker db — the broker db is among the relocated // files. Idempotent; a no-op once migrated. crate::paths::relocate_legacy_state(); // `dashboard_port` is consumed into the Coordinator below; capture the @@ -338,9 +338,6 @@ async fn cmd_serve( // (`OTEL_EXPORTER_OTLP_ENDPOINT`), unauthenticated. The swarm-otel // collector alone forwards upstream. No-op when the endpoint is unset. crate::otel_metrics::spawn_exporter(&coord.hyperhive_flake); - // build_logs.sqlite vacuum: c0re-side (single db). Failures kept - // 30d, successes 24h — see `build_logs::vacuum` for the rule. - crate::build_logs::spawn_vacuum(&coord); // Container crash watcher: warn!()s "container crash detected" when a // previously-running container goes away without an operator-initiated // transient state — a swarm-level grafana alert matches that line. @@ -373,11 +370,6 @@ async fn cmd_serve( job_queue::scheduler::run_worker(q_coord).await; }); } - // Forward every broker event onto the unified dashboard - // channel, so the dashboard SSE sees broker messages and - // mutation events on one stream. The broker's intra-process - // channel (used by `recv_blocking_batch`) stays untouched. - spawn_broker_to_dashboard_forwarder(coord.clone()); let dash_coord = coord.clone(); tokio::spawn(async move { if let Err(e) = dashboard::serve(dashboard_port, dash_coord).await { @@ -406,59 +398,3 @@ async fn cmd_serve( } Ok(()) } - -/// Re-emit every broker `MessageEvent` onto the dashboard channel as -/// a `DashboardEvent::Sent` / `Delivered`. -/// Background task; runs for the life of the process. On a lagged -/// broker subscription we just keep going — the dashboard channel is -/// best-effort presentation plumbing, the broker keeps its own sqlite -/// log for replay. -fn spawn_broker_to_dashboard_forwarder(coord: Arc) { - use broker::MessageEvent; - use dashboard_events::DashboardEvent; - let mut rx = coord.broker.subscribe(); - tokio::spawn(async move { - loop { - match rx.recv().await { - Ok(MessageEvent::Sent { - id, - from, - to, - body, - at, - in_reply_to, - }) => { - coord.emit_dashboard_event(DashboardEvent::Sent { - id, - from, - to, - body, - at: hive_sh4re::wire_time::from_secs(at), - in_reply_to, - }); - } - Ok(MessageEvent::Delivered { - id, - from, - to, - body, - at, - in_reply_to, - }) => { - coord.emit_dashboard_event(DashboardEvent::Delivered { - id, - from, - to, - body, - at: hive_sh4re::wire_time::from_secs(at), - in_reply_to, - }); - } - Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => { - tracing::warn!(skipped = n, "broker-to-dashboard forwarder lagged"); - } - Err(tokio::sync::broadcast::error::RecvError::Closed) => break, - } - } - }); -} diff --git a/hive-c0re/src/paths.rs b/hive-c0re/src/paths.rs index ee9c45747..31d8f186d 100644 --- a/hive-c0re/src/paths.rs +++ b/hive-c0re/src/paths.rs @@ -48,8 +48,7 @@ pub const STATE_ROOT: &str = "/var/lib/hyperhive"; pub const RUNTIME_ROOT: &str = "/run/hyperhive"; /// Default broker db path (`db/broker.sqlite`). Exposed as a `&str` for -/// the `--broker-db` clap `default_value`; `build_logs.sqlite` is placed -/// alongside it (the build-logs store keys off the broker db's parent). +/// the `--broker-db` clap `default_value`. pub const BROKER_DB: &str = "/var/lib/hyperhive/db/broker.sqlite"; #[must_use] @@ -57,7 +56,7 @@ pub fn state_root() -> PathBuf { PathBuf::from(STATE_ROOT) } -/// `db/` — sqlite databases (broker, build logs). +/// `db/` — sqlite databases (broker). #[must_use] pub fn db_dir() -> PathBuf { state_root().join("db") @@ -210,15 +209,13 @@ pub fn agent_runtime_dir(name: &str) -> PathBuf { /// rename-based: a move only happens when the old path exists and the /// new one doesn't, so re-runs are no-ops. /// -/// Must run **before** the broker db is opened (the broker + build-logs -/// dbs are relocated here). Safe because those dbs use rollback-journal +/// Must run **before** the broker db is opened (the broker db is relocated here). Safe because it uses rollback-journal /// mode (no `-wal`/`-shm` sidecars after a clean shutdown), and a rename /// within the same filesystem is atomic. pub fn relocate_legacy_state() { let root = state_root(); - let moves: [(&str, PathBuf); 3] = [ + let moves: [(&str, PathBuf); 2] = [ ("broker.sqlite", db_dir().join("broker.sqlite")), - ("build_logs.sqlite", db_dir().join("build_logs.sqlite")), ("agent-sockets.json", agent_sockets_file()), ]; for (old_rel, new) in &moves { diff --git a/hive-c0re/src/priv_client.rs b/hive-c0re/src/priv_client.rs index 03dfa5f53..6cbcbe799 100644 --- a/hive-c0re/src/priv_client.rs +++ b/hive-c0re/src/priv_client.rs @@ -124,7 +124,7 @@ pub async fn kill_container(name: &str) -> Result<()> { /// Streaming variant: forward stdout/stderr lines to `on_line` as they /// arrive. Returns `Ok(())` on success; the callback is responsible for -/// appending lines to `build_logs` or otherwise capturing the output. +/// capturing the output. pub async fn update_container_streaming( name: &str, toplevel: &str, diff --git a/hive-c0re/src/stores/broker.rs b/hive-c0re/src/stores/broker.rs index 445ff91ea..2def199dd 100644 --- a/hive-c0re/src/stores/broker.rs +++ b/hive-c0re/src/stores/broker.rs @@ -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, - }, - 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, - }, + 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) } diff --git a/hive-c0re/src/stores/build_logs.rs b/hive-c0re/src/stores/build_logs.rs deleted file mode 100644 index bd02850c6..000000000 --- a/hive-c0re/src/stores/build_logs.rs +++ /dev/null @@ -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` 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> = 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) { - 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> { - 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, - pub finished_at: Option, - pub stdout: String, - pub stderr: String, -} - -/// Sqlite-backed build-log store. `Arc`-friendly: all -/// methods take `&self`, internal `Mutex` serializes -/// access. -pub struct BuildLogs { - conn: Mutex, -} - -impl BuildLogs { - pub fn open(db_dir: &Path) -> Result { - 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 { - 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, - ) -> Result { - 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 { - 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>(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> { - 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 { - 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) { - 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")); - } -} diff --git a/hive-c0re/src/stores/mod.rs b/hive-c0re/src/stores/mod.rs index 963fff80b..605e842fe 100644 --- a/hive-c0re/src/stores/mod.rs +++ b/hive-c0re/src/stores/mod.rs @@ -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; diff --git a/hive-jobq-wire/README.md b/hive-jobq-wire/README.md index 78031a4b8..d4c6398a4 100644 --- a/hive-jobq-wire/README.md +++ b/hive-jobq-wire/README.md @@ -9,8 +9,7 @@ different audience, and folding it in means every consumer of the scheduler also carries a JSON vocabulary it may never serve — the two get remixed and stay that way. A separate crate keeps that boundary where it can be seen. -A host implements `WireNode` for its node payload `N` and `WireResource` for -its resource name `R`. `GraphWire::wire_snapshot` is then blanket-implemented -for `hive_jobq::Graph` — so the projection exists exactly when both types -have said how they render, and a payload that hasn't cannot reach a viewer at -all. +A host implements `WireNode` for its node payload `N`. `GraphWire::wire_snapshot` +is then blanket-implemented for `hive_jobq::Graph` — so the projection +exists exactly when the payload has said how it renders, and a payload that +hasn't cannot reach a viewer at all. Resource edges stay off the wire. diff --git a/hive-jobq-wire/src/lib.rs b/hive-jobq-wire/src/lib.rs index 5b5878f74..cb8d10764 100644 --- a/hive-jobq-wire/src/lib.rs +++ b/hive-jobq-wire/src/lib.rs @@ -11,13 +11,15 @@ //! together they remix, and every consumer of the scheduler ends up carrying a //! JSON vocabulary it may never serve. //! -//! A host says how its `N` and `R` render by implementing [`WireNode`] and -//! [`WireResource`]. [`GraphWire`] is then blanket-implemented for any -//! `Graph` whose two type parameters do — so the projection exists -//! exactly when both have answered, and a payload that has never said how it -//! displays cannot reach a viewer at all. The alternative (a projection written -//! next to each host's endpoint) lets the second host ship a graph of -//! unlabelled integers. +//! A host says how its `N` renders by implementing [`WireNode`]. +//! [`GraphWire`] is then blanket-implemented for any `Graph` whose payload +//! does — so the projection exists exactly when `N` has answered, and a payload +//! that has never said how it displays cannot reach a viewer at all. The +//! alternative (a projection written next to each host's endpoint) lets the second +//! host ship a graph of unlabelled integers. +//! +//! Resource edges (`Dep::Resource`) are scheduling detail and do not ride the +//! wire: no viewer reads them. //! //! There is deliberately **no roll-up field**. A group root's own [`State`] //! *is* its subtree's answer: `Finishing` means "own logic done, children still @@ -97,15 +99,6 @@ pub trait WireNode { fn data(&self, id: WireId) -> serde_json::Value; } -/// How a resource name `R` rides the wire. -/// -/// Resource deps are named rather than typed on the wire: the resource -/// vocabulary belongs to the host, and a viewer only ever groups by the string. -pub trait WireResource { - /// Stable display name for this resource. - fn name(&self) -> String; -} - /// One node, serialised near-raw from [`hive_jobq::Node`]. #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] pub struct GraphNode { @@ -191,7 +184,7 @@ pub struct NodePayload { /// What must hold before a node runs. /// /// Externally tagged on `kind`, whose values are the **variant names verbatim** -/// (`"Node"`, `"Resource"`) — no `rename_all`. A rename is a second spelling of +/// (`"Node"`) — no `rename_all`. A rename is a second spelling of /// the same name that has to be kept in agreement with the Rust one by hand. #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] #[serde(tag = "kind")] @@ -202,14 +195,18 @@ pub enum GraphDep { /// (a filtered view); treat an absent target as satisfied. id: WireId, }, - /// Need `count` units of a named resource, per [`WireResource::name`]. - Resource { name: String, count: u32 }, + /// Any other `kind`. Only deserialising produces it: a sender from before + /// resource edges left the wire still emits `"Resource"`, and a viewer + /// must skip it rather than reject the whole graph. Nothing serialises + /// it; remove once no such sender runs. + #[serde(other)] + Unknown, } /// Serving a [`Graph`] to a viewer. /// -/// Blanket-implemented for every `Graph` whose payload and resource types -/// implement [`WireNode`] / [`WireResource`], and for no others — an extension +/// Blanket-implemented for every `Graph` whose payload implements +/// [`WireNode`], and for no others — an extension /// trait rather than an inherent method because the graph lives in the /// scheduler crate and this projection deliberately does not. pub trait GraphWire { @@ -229,7 +226,7 @@ pub trait GraphWire { fn wire_snapshot(&self, roots: impl IntoIterator) -> Vec; } -impl GraphWire for Graph { +impl GraphWire for Graph { fn wire_snapshot(&self, roots: impl IntoIterator) -> Vec { let mut roots: Vec = roots.into_iter().collect(); roots.sort_unstable_by_key(|id| id.get()); @@ -251,9 +248,9 @@ impl GraphWire for Graph { /// graph: a summary line, a badge, a health check. Every consumer re-deriving /// the same tally is the thing this replaces. /// -/// **Free of `WireNode`/`WireResource`, unlike [`GraphWire`]** — a node's state +/// **Free of `WireNode`, unlike [`GraphWire`]** — a node's state /// is a scheduler concept, so counting by state needs to know nothing about -/// what the payload or the resource *are*. Bounding it like the projection +/// what the payload *is*. Bounding it like the projection /// would make a host implement two display traits to be allowed to count, which /// is a requirement about rendering imposed on arithmetic. /// @@ -346,13 +343,13 @@ pub fn filter_nodes_by_state(nodes: Vec, states: Option<&[State]>) -> .collect() } -fn wire_node(node: &hive_jobq::Node) -> GraphNode { +fn wire_node(node: &hive_jobq::Node) -> GraphNode { let id = node.id.get(); GraphNode { id, parent: node.parent.map(NodeId::get), state: node.state, - deps: node.deps.iter().map(wire_dep).collect(), + deps: node.deps.iter().filter_map(wire_dep).collect(), error: node.error.clone(), payload: NodePayload { label: node.payload.label(), @@ -361,13 +358,10 @@ fn wire_node(node: &hive_jobq::Node) -> Grap } } -fn wire_dep(dep: &Dep) -> GraphDep { +fn wire_dep(dep: &Dep) -> Option { match dep { - Dep::Node { id, .. } => GraphDep::Node { id: id.get() }, - Dep::Resource { name, count } => GraphDep::Resource { - name: name.name(), - count: *count, - }, + Dep::Node { id, .. } => Some(GraphDep::Node { id: id.get() }), + Dep::Resource { .. } => None, } } @@ -426,7 +420,7 @@ mod tests { } // The test payloads. `&str` labels itself and has nothing extra to say; - // `String` is its own resource name. + impl WireNode for &str { fn label(&self) -> String { (*self).to_owned() @@ -438,12 +432,6 @@ mod tests { } } - impl super::WireResource for String { - fn name(&self) -> String { - self.clone() - } - } - /// The bound is the *caller's*: `wire_snapshot` serialises the roots it is /// handed and their subtrees, and nothing else. Nothing in a `Graph` is /// ever removed, so a host that couldn't restrict this would have no way to @@ -479,10 +467,11 @@ mod tests { assert_eq!(wire[1].parent, Some(shown[0].get())); } - /// `label` / `data` / `name` all come from the host's impls — the graph - /// layer never invents a rendering for a payload it cannot read. + /// `label` / `data` come from the host's impl — the graph layer never + /// invents a rendering for a payload it cannot read. A resource edge is in + /// the scheduler's graph but not on the wire. #[test] - fn payload_and_resource_rendering_come_from_the_host_impls() { + fn payload_rendering_comes_from_the_host_impl_and_resource_edges_stay_off_the_wire() { let mut sched = sched(); let ids = sched .insert_job(None, |job| { @@ -498,15 +487,18 @@ mod tests { let wire = sched.graph().wire_snapshot([first, second]); + assert_eq!( + sched.graph().node(second).expect("node").deps.len(), + 2, + "control: the scheduler holds the node dep and the resource dep" + ); assert_eq!(wire[0].payload.label, "reconcile"); assert_eq!(wire[0].payload.data["seen_id"], first.get()); match &wire[1].deps[..] { - [GraphDep::Node { id }, GraphDep::Resource { name, count }] => { + [GraphDep::Node { id }] => { assert_eq!(*id, first.get()); - assert_eq!(name, "build-slot", "the host's WireResource::name"); - assert_eq!(*count, 2); } - other => panic!("expected a node dep then a resource dep, got {other:?}"), + other => panic!("expected only the node dep, got {other:?}"), } } @@ -526,23 +518,39 @@ mod tests { #[test] fn a_node_round_trips_through_json() { - let before = node( - 7, - Some(3), - State::Running, - vec![ - GraphDep::Node { id: 3 }, - GraphDep::Resource { - name: "build-slot".to_owned(), - count: 1, - }, - ], - ); + let before = node(7, Some(3), State::Running, vec![GraphDep::Node { id: 3 }]); let json = serde_json::to_string(&before).expect("serialises"); let after: GraphNode = serde_json::from_str(&json).expect("round trips"); assert_eq!(format!("{after:?}"), format!("{before:?}")); } + /// A sender from before resource edges left the wire still emits them; the + /// receiver keeps the node and reads the edge as `Unknown` instead of + /// rejecting the graph. + #[test] + fn an_old_senders_resource_edge_is_tolerated() { + let mut json = serde_json::to_value(node( + 7, + None, + State::Pending, + vec![GraphDep::Node { id: 3 }], + )) + .expect("serialises"); + json["deps"] + .as_array_mut() + .expect("deps") + .push(serde_json::json!({ "kind": "Resource", "name": "build-slot", "count": 1 })); + + let after: GraphNode = serde_json::from_value(json).expect("tolerated"); + + assert_eq!(after.deps.len(), 2); + assert!( + matches!(after.deps[0], GraphDep::Node { id: 3 }), + "control: a node edge still parses" + ); + assert!(matches!(after.deps[1], GraphDep::Unknown)); + } + /// The serialised node carries exactly these keys (`parent`, `deps` and /// `error` are omitted when empty, `payload.data` when null). #[test] diff --git a/hive-priv/src/container.rs b/hive-priv/src/container.rs index 89542630c..942704f3e 100644 --- a/hive-priv/src/container.rs +++ b/hive-priv/src/container.rs @@ -692,8 +692,8 @@ async fn container_run_streaming( if !status.success() { // Only the last stderr line is embedded — the full stderr was // already forwarded line-by-line as PrivEvent::Line messages and - // is captured in build_logs.sqlite by the caller. Keeping the - // error message short avoids bloating the anyhow chain. + // is logged to the host journal by the caller, through its BuildTag. + // Keeping the error message short avoids bloating the anyhow chain. bail!( "nixos-container {} failed ({}): {}", args.join(" "), diff --git a/hivectl/src/dag_progress.rs b/hivectl/src/dag_progress.rs index 2c2050672..fe7b84ef7 100644 --- a/hivectl/src/dag_progress.rs +++ b/hivectl/src/dag_progress.rs @@ -266,9 +266,8 @@ fn fmt_dur(secs: i64) -> String { /// One animated node line: kind, an `(after …)` marker for a fan-in node /// (>1 node-dependency), and a truncated error tail. -/// Resource deps (`GraphDep::Resource`) don't name another node, so they're -/// filtered out of the fan-in count the same way `hive-c0re`'s own -/// `dag_view` projection already does. +/// Only node deps count toward the fan-in: an edge of unknown kind (see +/// `GraphDep::Unknown`) names no node. fn node_line(nodes: &[GraphNode], n: &GraphNode) -> String { use std::fmt::Write as _; let mut s = n.payload.label.clone(); @@ -277,7 +276,7 @@ fn node_line(nodes: &[GraphNode], n: &GraphNode) -> String { .iter() .filter_map(|d| match d { GraphDep::Node { id, .. } => Some(*id), - GraphDep::Resource { .. } => None, + GraphDep::Unknown => None, }) .collect(); if dep_ids.len() > 1 { diff --git a/nix/module-eval/grafana.nix b/nix/module-eval/grafana.nix index 97b7a99eb..025c5b1e6 100644 --- a/nix/module-eval/grafana.nix +++ b/nix/module-eval/grafana.nix @@ -643,7 +643,6 @@ let && lib.hasInfix "event = \"start\"" logged # each field as the logging macro spells it (`status` is a bare local) && lib.all (f: lib.hasInfix f logged) [ - "build_id = " "agent = " "kind = " "status," diff --git a/swarm-controller/src/jobq.rs b/swarm-controller/src/jobq.rs index 02a170dff..83453d31d 100644 --- a/swarm-controller/src/jobq.rs +++ b/swarm-controller/src/jobq.rs @@ -263,15 +263,6 @@ pub(crate) enum SwarmResourceKind { Meta, } -impl hive_jobq_wire::WireResource for SwarmResourceKind { - fn name(&self) -> String { - match self { - SwarmResourceKind::Builder => "builder".to_owned(), - SwarmResourceKind::Meta => "meta".to_owned(), - } - } -} - /// A dependency this daemon may only gain after it starts serving: empty /// until then, filled once. Readers take what is there at the moment they ask. pub(crate) type Late = Arc>;