From 77cc7bea6b608cc3365abb1d46149acf1bb30b6e Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 2 Aug 2026 18:26:32 +0200 Subject: [PATCH] refactor(#2949): the build-log row carries its node id MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `QueueInner` was `{ sched, node_rt }`, where `node_rt` held exactly one datum per node: the `build_logs` row id. It existed because a `hive_jobq` node payload is immutable after insert while the log row is created when the build starts — so the link could not ride the node. Invert it: the log row names its node (`build_logs.node_id`, one migration in the existing `schema_versions` framework). Same single-home property, in the direction the type system allows. `QueueInner` is now just the scheduler. That is the point: the queue holds no per-node side map, so nothing has to be locked alongside the graph. Deleted as a consequence, each surfaced by dead-code analysis after the edit above rather than predicted: - `NodeRuntime`, `node_rt`, `set_build_log_id`, and `build_log_id_of` (which linear-scanned the map to match a wire `u64` against opaque `NodeId`s). The lookup is an indexed query now. - `struct Ctx`, entirely. It carried `coord` + `dag_id` + `node_id` into the executors so the build-log callback could reach the queue; without the callback, `coord`/`dag_id` were never read and `node_id` was already on the `Claim` both executors receive. - `QueueInner::node_running`, which existed only for `set_build_log_id`'s "only while running" guard. - The `Fn(i64)` callbacks on `prebuild_toplevel` / `swap_update` / `priv_run_inner`, replaced by a `node_id: Option` passed down. The id travels one way now instead of being registered back. `meta.rs`'s `nix_logged` passes `None` deliberately: its callers reach it from outside the queue as well as inside, and nothing reads the link for them yet. `id_for_node` takes `MAX(id)` rather than assuming uniqueness — a retried node opens a second row and the panel wants the current attempt. The test moved to where the behaviour lives and covers that, plus survival across completion and non-collision with node-less rows. --- hive-c0re/src/dashboard/build_logs.rs | 4 +- hive-c0re/src/job_queue/exec.rs | 41 ++-------- hive-c0re/src/job_queue/mod.rs | 77 +++++------------- hive-c0re/src/job_queue/tests.rs | 27 ++----- hive-c0re/src/lifecycle/mod.rs | 49 ++++-------- hive-c0re/src/meta.rs | 7 +- hive-c0re/src/stores/build_logs.rs | 107 +++++++++++++++++++++++--- 7 files changed, 150 insertions(+), 162 deletions(-) diff --git a/hive-c0re/src/dashboard/build_logs.rs b/hive-c0re/src/dashboard/build_logs.rs index f4835878..763ee9fb 100644 --- a/hive-c0re/src/dashboard/build_logs.rs +++ b/hive-c0re/src/dashboard/build_logs.rs @@ -156,7 +156,7 @@ pub(super) async fn get_build_log_for_node( State(state): State, AxumPath(node_id): AxumPath, ) -> Response { - match state.coord.job_queue.build_log_id_of(node_id) { + match state.coord.build_logs.id_for_node(node_id) { Some(log_id) => get_build_log_full(State(state), AxumPath(log_id)).await, None => ( StatusCode::NOT_FOUND, @@ -183,7 +183,7 @@ pub(super) async fn get_build_log_raw_for_node( State(state): State, AxumPath(node_id): AxumPath, ) -> Response { - match state.coord.job_queue.build_log_id_of(node_id) { + match state.coord.build_logs.id_for_node(node_id) { Some(log_id) => get_build_log_raw(State(state), AxumPath(log_id)).await, None => ( StatusCode::NOT_FOUND, diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index 2d69a6b1..de9e1f7d 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -26,25 +26,6 @@ use crate::power::{ReconcileAction, reconcile_action}; /// N × this timeout. pub const GRACEFUL_STOP_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(3); -/// Build-log sink for one claimed node. -struct Ctx<'a> { - coord: &'a Arc, - dag_id: u64, - node_id: super::NodeId, -} - -impl Ctx<'_> { - fn build_log(&self, log_id: i64) { - if self - .coord - .job_queue - .set_build_log_id(self.dag_id, self.node_id, log_id) - { - self.coord.emit_rebuild_queue_snapshot(); - } - } -} - /// Run one claimed node to completion. Called from a task the /// scheduler spawns per claim; the `Result` (stringified) becomes the /// node's terminal state. @@ -67,19 +48,14 @@ pub(super) async fn run_node( job: super::Job, claim: &Claim, ) -> (super::Job, Result<()>) { - let ctx = Ctx { - coord, - dag_id: claim.dag_id, - node_id: claim.node_id, - }; // Every arm is `Result<()>`; the three that grow work declare into `job` // *synchronously*, after their own awaits have finished. Borrowing `&job` // inside an `.await` would make this future non-`Send` (see above), so the // growth executors return what to grow rather than taking the builder. let result = match &claim.kind { NodeKind::MetaSync { relock, .. } => run_meta_sync(coord, claim, *relock).await, - NodeKind::Prebuild { .. } => run_prebuild(claim, &ctx).await, - NodeKind::Swap { .. } => run_swap(coord, claim, &ctx).await, + NodeKind::Prebuild { .. } => run_prebuild(claim).await, + NodeKind::Swap { .. } => run_swap(coord, claim).await, NodeKind::PostSwap { .. } => run_post_swap(coord, claim).await, NodeKind::Provision { .. } => run_provision(coord, claim).await, NodeKind::Create { .. } => run_create(claim).await, @@ -241,7 +217,7 @@ async fn run_meta_sync(coord: &Arc, claim: &Claim, relock: bool) -> /// container is already down: its only purpose is to shrink the swap's /// downtime window, so a stopped agent (no uptime to preserve) doesn't /// pay the double eval — `Swap` builds inline instead. -async fn run_prebuild(claim: &Claim, ctx: &Ctx<'_>) -> Result<()> { +async fn run_prebuild(claim: &Claim) -> Result<()> { let name = &claim.agent; // Warm the toplevel build only when the container is up — the whole // point of prebuild is to shrink the swap's downtime window. A @@ -249,8 +225,7 @@ async fn run_prebuild(claim: &Claim, ctx: &Ctx<'_>) -> Result<()> { // eval and let the downstream `Swap` build inline. if crate::lifecycle::is_running(name).await { let flake_ref = format!("{}#{name}", crate::paths::meta_root().display()); - crate::lifecycle::prebuild_toplevel(name, &flake_ref, &|log_id| ctx.build_log(log_id)) - .await?; + crate::lifecycle::prebuild_toplevel(name, &flake_ref, Some(claim.node_id.get())).await?; } Ok(()) } @@ -260,17 +235,15 @@ async fn run_prebuild(claim: &Claim, ctx: &Ctx<'_>) -> Result<()> { /// (rev marker, `Rebuilt` event, forge/matrix sync, kick, rescan). /// 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(coord: &Arc, claim: &Claim, ctx: &Ctx<'_>) -> Result<()> { +async fn run_swap(coord: &Arc, claim: &Claim) -> Result<()> { let name = &claim.agent; // 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); let hive = coord.hive_env(); let paths = Coordinator::agent_paths(name, agent_dir); - let result = crate::lifecycle::swap_update(name, &hive, &paths, &|log_id| { - ctx.build_log(log_id); - }) - .await; + let result = + crate::lifecycle::swap_update(name, &hive, &paths, Some(claim.node_id.get())).await; // On success the Ok-only bookkeeping tail (rev marker, forge/matrix // sync, kick, rescan, snapshot) runs in the sibling `PostSwap` node, // which deps `AfterOk(Swap)`. On failure `PostSwap` is cancel-cascaded diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 3e351673..dfe916ec 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -36,7 +36,6 @@ pub mod templates; #[cfg(test)] mod tests; -use std::collections::HashMap; use std::sync::Mutex; use chrono::{DateTime, Utc}; @@ -101,15 +100,6 @@ pub struct Claim { pub agent: String, } -/// Per-node runtime metadata the crate graph doesn't carry. Lifecycle -/// (`started_at` / `finished_at` / `error`) lives on the `hive_jobq::Node` -/// itself now, so only the build-log row link remains host-side (the -/// client fetches the log by node id). -#[derive(Debug, Default, Clone)] -struct NodeRuntime { - build_log_id: Option, -} - /// An owned read-view of a DAG container's carried metadata ([`NodeKind::Dag`]). /// Derived on read from the container node — the data has a single home (the /// node payload); this is not a stored side-table. @@ -119,18 +109,21 @@ struct DagMeta { created_at: DateTime, } -/// The mutable queue state behind the mutex: the crate scheduler plus the -/// per-node runtime metadata the graph can't carry. A **DAG is a single -/// container node** ([`NodeKind::Dag`], `parent = None`) whose subtree is the -/// DAG's work — so the container's `NodeId` is the DAG id, its rolled-up state -/// is the DAG state, and there are no grouping side-tables: membership + meta -/// are graph queries ([`QueueInner::container`] / [`QueueInner::dag_meta`] + -/// the `hive_jobq::Graph` accessors). One shared crate [`Graph`] holds every DAG. +/// The mutable queue state behind the mutex: **just the crate scheduler**. +/// A **DAG is a single container node** ([`NodeKind::Dag`], `parent = None`) +/// whose subtree is the DAG's work — so the container's `NodeId` is the DAG id, +/// its rolled-up state is the DAG state, and there are no grouping side-tables: +/// membership + meta are graph queries ([`QueueInner::container`] / +/// [`QueueInner::dag_meta`] + the `hive_jobq::Graph` accessors). One shared +/// crate [`Graph`] holds every DAG. +/// +/// There is deliberately **no per-node side map** any more. The last one held +/// the `build_logs` row id; that link now lives on the log row itself +/// (`build_logs.node_id`), so it survives a restart and needs no lock held +/// alongside the scheduler's — which is what lets the scheduler's own lock be +/// the only one the run loop takes. struct QueueInner { sched: Scheduler, - /// Per-node runtime metadata (the build-log id) — mutable after - /// insert, so it can't ride the immutable node payload. - node_rt: HashMap, } /// The queue. Lives on `Coordinator` (one per hive-c0re process); a single @@ -208,7 +201,6 @@ impl JobQueue { Self { inner: Mutex::new(QueueInner { sched: Scheduler::new(Graph::new(), table), - node_rt: HashMap::new(), }), notify: Notify::new(), } @@ -244,7 +236,6 @@ impl JobQueue { None, ) .map_err(|e| anyhow::anyhow!("job_queue: container insert failed: {e}"))?; - inner.node_rt.insert(container, NodeRuntime::default()); insert_group(&mut inner, spec.declare, Some(container))?; // Settle the container's own (no-op) logic immediately so it parks in // `Finishing` and its children become runnable — it never needs claiming @@ -374,33 +365,6 @@ impl JobQueue { true } - /// Link a `build_logs` row to a specific `Running` node. - pub fn set_build_log_id(&self, dag_id: u64, node_id: NodeId, log_id: i64) -> bool { - let mut inner = self.lock(); - if inner.sched.graph().root_of(node_id).map(NodeId::get) != Some(dag_id) - || !inner.node_running(node_id) - { - return false; - } - inner.node_rt.entry(node_id).or_default().build_log_id = Some(log_id); - true - } - - /// The `build_logs` row id linked to the wire node id `node_id`, if any — - /// the lookup behind the `GET /api/build-log/` query endpoint (the - /// client fetches a node's captured build output on demand rather than - /// receiving it inline). Takes the raw wire `u64` (the endpoint's path - /// param); `node_rt` is keyed by the opaque `NodeId`, so this scans for the - /// matching id — the map is small (live + recently-terminal nodes). - #[must_use] - pub fn build_log_id_of(&self, node_id: u64) -> Option { - self.lock() - .node_rt - .iter() - .find(|(nid, _)| nid.get() == node_id) - .and_then(|(_, rt)| rt.build_log_id) - } - /// The first failed node's error in `dag_id`, if any has failed yet. /// /// Unlike the roll-up summary this is readable *mid-flight*, which is the @@ -498,14 +462,6 @@ impl JobQueue { } impl QueueInner { - /// Whether `id` is a `Running` node. - fn node_running(&self, id: NodeId) -> bool { - self.sched - .graph() - .node(id) - .is_some_and(|n| n.state == State::Running) - } - /// The container node of `dag_id` — the `NodeKind::Dag` root whose id equals /// `dag_id`. `NodeId` is un-fabricable from a raw `u64`, so this is a search. fn container(&self, dag_id: u64) -> Option { @@ -593,7 +549,12 @@ impl QueueInner { NodeKind::MetaLock { inputs, .. } => inputs.clone(), _ => Vec::new(), }; - let build_log_id = self.node_rt.get(&id).and_then(|r| r.build_log_id); + // Looked up from the log row itself (`build_logs.node_id`), not a + // host-side map. One indexed query per node in the snapshot; the + // node set is bounded by `MAX_HISTORY_DAGS` and the store is a + // local sqlite file, so this is cheaper than the lock contention + // a second shared map would reintroduce. + let build_log_id = crate::build_logs::global().and_then(|h| h.id_for_node(id.get())); // `node.parent` is the structural jobq parent. Top-level nodes // have `parent == Some(container)` (direct children of the Dag // container); those become `parent: None` on the wire since the diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index f460092a..82065941 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -1544,27 +1544,12 @@ fn deploy_dag_skips_apply_but_still_runs_tail_when_verify_fails() { assert_eq!(state_of(&q, id), State::Failed); } -// ---- build logs, history ---- - -#[test] -fn set_build_log_id_links_running_node() { - let q = JobQueue::new(1); - let id = submit(&q, rebuild("agent-a", "r")); - let c = claim_one(&q); - assert!(q.set_build_log_id(id, c.node_id, 42)); - q.complete_node(c.node_id, Ok(())); - assert!( - !q.set_build_log_id(id, c.node_id, 99), - "node no longer running → refused" - ); - // The log id is fetched by node id (the `GET /api/build-log/` lookup), - // not carried on the wire — it survives completion in the node runtime. - assert_eq!( - q.build_log_id_of(c.node_id.get()), - Some(42), - "log id survives completion" - ); -} +// ---- 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 6f816f02..a69b71a9 100644 --- a/hive-c0re/src/lifecycle/mod.rs +++ b/hive-c0re/src/lifecycle/mod.rs @@ -274,10 +274,10 @@ pub async fn swap_update( name: &str, hive: &HiveEnv, paths: &AgentPaths, - on_build_log_id: &(dyn Fn(i64) + Send + Sync), + node_id: Option, ) -> Result<()> { write_dropins(name, hive, paths).await?; - priv_run_inner("update", name, Some(on_build_log_id)).await + priv_run_inner("update", name, node_id).await } /// Build the `AgentSpec` list for the meta flake from `nixos-container @@ -582,14 +582,10 @@ pub async fn destroy(name: &str) -> Result<()> { /// the prebuild happens before stop, and `docs/coordinator.md::Prebuild /// attr path` for why the explicit nixosConfigurations attr is required. /// -/// `on_build_log_id` fires with the `build_logs` row id as soon as the -/// row opens, so queue-side callers can link their node to the live -/// stream. Pass `&|_| ()` when not needed. -pub async fn prebuild_toplevel( - name: &str, - flake_ref: &str, - on_build_log_id: &(dyn Fn(i64) + Send + Sync), -) -> Result<()> { +/// `node_id` is the queue node this build belongs to, when there is one — +/// it is stored on the `build_logs` row so the dashboard can find the log +/// from the node. Pass `None` for builds that run outside the queue. +pub async fn prebuild_toplevel(name: &str, flake_ref: &str, node_id: Option) -> Result<()> { use tokio::io::{AsyncBufReadExt, BufReader}; // Split `#` so we can re-emit with the explicit // `nixosConfigurations.` segment. The flake_ref shape is @@ -624,15 +620,12 @@ pub async fn prebuild_toplevel( // 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) + h.start(name, "prebuild", &cmdline, node_id) .map_err(|e| { tracing::warn!(error = ?e, "build_logs: start failed (prebuild log dropped)"); }) .ok() }); - if let Some(id) = log_id { - on_build_log_id(id); - } let mut child = Command::new("nix") .args(&args) @@ -784,37 +777,25 @@ async fn priv_run(kind: &str, name: &str) -> Result<()> { priv_run_inner(kind, name, None).await } -/// Like `priv_run` but calls `on_log_id(log_id)` immediately after the -/// build-log row is opened — before the actual container op starts. -/// This lets callers surface the row id for live streaming (e.g. the -/// rebuild-queue worker sets `build_log_id` on the queue entry so the -/// dashboard can link to `/api/build-logs/id/{id}/stream`). +/// Like `priv_run` but stamps `node_id` onto the build-log row it opens, so +/// the dashboard can find the log from the queue node (and link to +/// `/api/build-logs/id/{id}/stream`). /// -/// The callback fires only when a build-log row is successfully opened -/// (i.e. the global `BuildLogs` handle is installed AND `h.start()` -/// succeeds). No-op when `on_log_id` is `None` — that's the path for -/// all callers that don't need the id. -async fn priv_run_inner( - kind: &str, - name: &str, - on_log_id: Option<&(dyn Fn(i64) + Send + Sync)>, -) -> Result<()> { +/// 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, node_id: Option) -> 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) + h.start(name, kind, &cmdline, node_id) .map_err(|e| { tracing::warn!(error = ?e, "build_logs: start failed (priv_run log dropped)"); }) .ok() }); - // Notify the caller as soon as the log row exists so it can surface - // the id for live streaming before the container op even starts. - if let (Some(id), Some(cb)) = (log_id, on_log_id) { - cb(id); - } // For long-running ops use the streaming protocol so build_logs // receives lines in real time rather than as a batch at completion. diff --git a/hive-c0re/src/meta.rs b/hive-c0re/src/meta.rs index 8c4100ba..6da608f3 100644 --- a/hive-c0re/src/meta.rs +++ b/hive-c0re/src/meta.rs @@ -1590,7 +1590,12 @@ async fn nix_logged(dir: &Path, args: &[&str], agent: &str, kind: &str) -> Resul let cmdline = format!("nix {}", nix_argv(args).join(" ")); let logs = crate::build_logs::global(); let log_id = logs.as_ref().and_then(|h| { - h.start(agent, kind, &cmdline) + // No node id: `nix_logged`'s two callers are meta-flake operations + // reached from outside the queue as well as from inside it, and the + // agent+kind+time listing is how they're surfaced today. Linking them + // to a node would mean threading the id through `meta`'s public API + // for no current reader — worth doing when something wants it. + h.start(agent, kind, &cmdline, None) .map_err(|e| { tracing::warn!(error = ?e, %kind, "build_logs: start failed (meta log dropped)"); }) diff --git a/hive-c0re/src/stores/build_logs.rs b/hive-c0re/src/stores/build_logs.rs index 24b302bd..67b28c3e 100644 --- a/hive-c0re/src/stores/build_logs.rs +++ b/hive-c0re/src/stores/build_logs.rs @@ -13,6 +13,8 @@ use serde::Serialize; use tokio::sync::broadcast; use utoipa::ToSchema; +use crate::db::Migration; + /// Process-singleton handle, set once at coordinator startup. Lets /// the `lifecycle` module's `run` / `prebuild_toplevel` access the /// writer without threading an `Arc` through every @@ -65,6 +67,27 @@ CREATE INDEX IF NOT EXISTS idx_build_logs_status_finished 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)] @@ -152,6 +175,7 @@ impl BuildLogs { 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)?; let (notify_tx, _) = broadcast::channel(NOTIFY_CAP); Ok(Self { conn: Mutex::new(conn), @@ -169,17 +193,49 @@ impl BuildLogs { /// 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 { + /// + /// `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) VALUES (?1, ?2, ?3, ?4)", - params![agent, kind, cmdline, now], + "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. @@ -453,7 +509,7 @@ mod tests { fn start_appends_finish_flow() { let (_d, db) = tmpdb(); let id = db - .start("alice", "prebuild", "nix build foo") + .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"); @@ -482,9 +538,9 @@ mod tests { // 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"); + let id_a1 = db.start("alice", "run", "cmd one", None).expect("start"); + let _id_b = db.start("bob", "run", "cmd two", None).expect("start"); + let id_a2 = db.start("alice", "run", "cmd three", None).expect("start"); db.finish(id_a1, BuildStatus::Ok); let alice_rows = db.list_recent_for_agent("alice", 10).expect("list"); @@ -515,10 +571,12 @@ mod tests { #[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"); + 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); @@ -550,6 +608,31 @@ mod tests { 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 @@ -557,7 +640,7 @@ mod tests { // 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"); + 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");