diff --git a/docs/coordinator.md b/docs/coordinator.md index fadfd605..09126c86 100644 --- a/docs/coordinator.md +++ b/docs/coordinator.md @@ -74,33 +74,6 @@ as it progresses through lifecycle phases (`"nix build"`, `"nixos-container stop `/api/state` and renders the current step beneath the running entry so the operator can see which phase is taking time. -### Dependency tracking - -Each entry carries a `depends_on: Vec` field. The worker skips entries whose -dependencies are not yet resolved — a dependency is resolved when its id is either -in the queue as a terminal entry (`Done` / `Failed` / `Cancelled`) or no longer in -the queue at all (evicted by `trim_history`, which only evicts terminals). - -Use cases: -- Chain a `Rebuild` after an explicit prerequisite step without coupling them through - the `parent_id` cascade mechanism. -- Sequence a `PermChange` + `Rebuild` pair where the rebuild must not start until the - perm-file write commits (already handled by the single-worker FIFO today, but - `depends_on` allows explicit cross-kind sequencing when parallel workers are added). - -`depends_on` is part of the dedup key: two entries with the same `(kind, agent, -parent_id, inputs, approval_id)` but different dep sets are treated as distinct work. - -**Worker re-notification**: the worker drains `take_next()` in a tight loop after -each entry finishes. When a dep entry transitions to terminal, the loop re-evaluates -the queue immediately, so downstream entries are unblocked with no extra wakeup. No -additional `notify_one()` call is needed. - -**Circular-dep caveat**: if A depends on B and B depends on A, neither entry ever -becomes runnable — the worker skips both indefinitely with no error. Callers must -ensure acyclic dep graphs. Cycle detection is deferred to a future iteration (when -parallel workers make a stuck queue more visible). - --- ## Container view diff --git a/frontend/packages/dashboard/src/logs.js b/frontend/packages/dashboard/src/logs.js index 4e9b4b4a..ab27fec9 100644 --- a/frontend/packages/dashboard/src/logs.js +++ b/frontend/packages/dashboard/src/logs.js @@ -73,7 +73,7 @@ import { } const ul = el('ul', { class: 'build-logs-list' }); for (const h of rows) { - const li = el('li', { class: 'build-logs-item', 'data-log-id': String(h.id) }); + const li = el('li', { class: 'build-logs-item' }); // status is null while running, 'ok'/'fail' when finished. const live = !h.status; const ok = h.status === 'ok'; @@ -175,20 +175,6 @@ import { ul.append(li); } buildList.append(ul); - // Deep-link: if the URL contains ?id=N, auto-expand that log entry - // so a `logs →` link from the rebuild-queue panel drops the operator - // straight into the live output without extra clicks. - const deepId = new URLSearchParams(location.search).get('id'); - if (deepId) { - const target = ul.querySelector('[data-log-id="' + deepId + '"]'); - if (target) { - const btn = target.querySelector('.build-logs-row-btn'); - if (btn) { - btn.click(); - target.scrollIntoView({ behavior: 'smooth', block: 'start' }); - } - } - } } catch (err) { buildList.replaceChildren(); buildList.append(el('p', { class: 'meta' }, 'fetch failed: ' + err)); diff --git a/frontend/packages/dashboard/src/tabs.js b/frontend/packages/dashboard/src/tabs.js index 9895687a..d28e0568 100644 --- a/frontend/packages/dashboard/src/tabs.js +++ b/frontend/packages/dashboard/src/tabs.js @@ -2266,20 +2266,6 @@ window.marked = marked; if (entry.step) { li.append(el('div', { class: 'rqe-step' }, '↳ ' + entry.step)); } - // Live-log link: when a build_log_id is present the update/create op - // opened a build_logs row; the SSE stream endpoint lets the operator - // follow output in real time without polling. - if (entry.build_log_id != null) { - li.append( - ' ', - el('a', { - class: 'rqe-log-link', - href: '/logs.html?id=' + entry.build_log_id + '#build', - target: '_blank', - title: 'view build log #' + entry.build_log_id, - }, 'logs →'), - ); - } // Error block, when failed. if (entry.error) { li.append(el('pre', { class: 'rqe-error', title: entry.error }, truncate(entry.error, 200))); diff --git a/hive-c0re/src/actions.rs b/hive-c0re/src/actions.rs index 9b573ec6..2f70c4c8 100644 --- a/hive-c0re/src/actions.rs +++ b/hive-c0re/src/actions.rs @@ -55,7 +55,6 @@ pub async fn approve(coord: Arc, id: i64) -> Result<()> { Vec::new(), Some(id), None, - Vec::new(), ); coord.emit_rebuild_queue_snapshot(); Ok(()) @@ -75,7 +74,6 @@ pub async fn approve(coord: Arc, id: i64) -> Result<()> { inputs.clone(), Some(id), None, - Vec::new(), ); // Pre-enqueue cascade rebuilds in topological order so // agents depending on updated inputs are rebuilt after the @@ -104,7 +102,6 @@ pub async fn approve(coord: Arc, id: i64) -> Result<()> { Vec::new(), Some(id), None, - Vec::new(), ); coord.emit_rebuild_queue_snapshot(); Ok(()) @@ -550,19 +547,9 @@ async fn run_apply_commit( // "nixos-container update" label for the whole multi-minute window. let hive = coord.hive_env(); let paths = Coordinator::agent_paths(&approval.agent, agent_dir.to_path_buf()); - let build_result = lifecycle::rebuild_no_meta( - &approval.agent, - &hive, - &paths, - &|step| coord.set_queue_step(queue_entry_id, step), - &|log_id| { - if let Some(qid) = queue_entry_id { - if coord.rebuild_queue.set_build_log_id(qid, log_id) { - coord.emit_rebuild_queue_snapshot(); - } - } - }, - ) + let build_result = lifecycle::rebuild_no_meta(&approval.agent, &hive, &paths, &|step| { + coord.set_queue_step(queue_entry_id, step) + }) .await; match build_result { diff --git a/hive-c0re/src/auto_update.rs b/hive-c0re/src/auto_update.rs index 3f21a210..44886d86 100644 --- a/hive-c0re/src/auto_update.rs +++ b/hive-c0re/src/auto_update.rs @@ -84,19 +84,9 @@ pub async fn rebuild_agent( // lifecycle_action; this catches the auto-update scan + any // other direct caller. let guard = coord.transient_guard(name, crate::coordinator::TransientKind::Rebuilding); - let result = lifecycle::rebuild( - name, - &hive, - &paths, - &|step| coord.set_queue_step(queue_entry_id, step), - &|log_id| { - if let Some(qid) = queue_entry_id { - if coord.rebuild_queue.set_build_log_id(qid, log_id) { - coord.emit_rebuild_queue_snapshot(); - } - } - }, - ) + let result = lifecycle::rebuild(name, &hive, &paths, &|step| { + coord.set_queue_step(queue_entry_id, step) + }) .await; drop(guard); match &result { diff --git a/hive-c0re/src/lifecycle.rs b/hive-c0re/src/lifecycle.rs index c11c3bc0..0797409a 100644 --- a/hive-c0re/src/lifecycle.rs +++ b/hive-c0re/src/lifecycle.rs @@ -377,7 +377,6 @@ pub async fn rebuild( hive: &HiveEnv, paths: &AgentPaths, on_step: &(dyn Fn(&str) + Send + Sync), - on_build_log_id: &(dyn Fn(i64) + Send + Sync), ) -> Result<()> { // Sync the meta flake (idempotent — no-op when the rendered // flake matches disk) so a manual rebuild from the dashboard @@ -390,7 +389,7 @@ pub async fn rebuild( // `applied//main` currently points at (deployed/). // Commits the lock if it changed. crate::meta::lock_update_for_rebuild(name).await?; - rebuild_no_meta(name, hive, paths, on_step, on_build_log_id).await + rebuild_no_meta(name, hive, paths, on_step).await } /// Container-level rebuild without touching the meta repo. Callers @@ -403,17 +402,11 @@ pub async fn rebuild( /// label so callers can surface progress (e.g. update the rebuild-queue /// step shown in the dashboard). Pass `&|_| ()` when progress reporting /// is not needed. -/// -/// `on_build_log_id` is called with the build-log row id immediately after -/// the `nixos-container update` log row opens, before the actual update -/// command starts. Callers can use this to link the queue entry to the log -/// for live streaming. Pass `&|_| ()` when not needed. pub async fn rebuild_no_meta( name: &str, hive: &HiveEnv, paths: &AgentPaths, on_step: &(dyn Fn(&str) + Send + Sync), - on_build_log_id: &(dyn Fn(i64) + Send + Sync), ) -> Result<()> { validate(name)?; if let Some(other) = port_collision(name).await { @@ -447,7 +440,7 @@ pub async fn rebuild_no_meta( priv_run("stop", name).await?; } on_step("nixos-container update"); - let update_result = priv_run_inner("update", name, Some(on_build_log_id)).await; + let update_result = priv_run("update", name).await; if let Err(ref update_err) = update_result { // The update failed (e.g. nix build error). If the agent was // running before we stopped it, try to bring it back up on the @@ -1329,24 +1322,6 @@ fn make_log_callback( /// is appended to the build-log row as it arrives, so the dashboard /// shows live progress during long `nixos-container create` / `update` runs. 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`). -/// -/// 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<()> { let container = container_name(name); let cmdline = format!("nixos-container {kind} {container}"); @@ -1358,11 +1333,6 @@ async fn priv_run_inner( }) .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/rebuild_queue.rs b/hive-c0re/src/rebuild_queue.rs index e299b243..a5238454 100644 --- a/hive-c0re/src/rebuild_queue.rs +++ b/hive-c0re/src/rebuild_queue.rs @@ -206,23 +206,6 @@ pub struct QueueEntry { /// the wire in those cases. #[serde(default, skip_serializing_if = "Option::is_none")] pub perm_payload: Option, - /// Entries this entry must wait for before it can run. The worker - /// skips this entry until every id in the list has reached a - /// terminal state (`Done` / `Failed` / `Cancelled`) — or no longer - /// exists in the queue (evicted terminal entries are treated as - /// resolved, since `trim_history` only evicts terminals). Empty on - /// most entries; serialised only when non-empty. - #[serde(default, skip_serializing_if = "Vec::is_empty")] - pub depends_on: Vec, - /// Row id of the associated `build_logs` entry (opened by the - /// lifecycle worker when `nixos-container update` starts). Set - /// shortly after `state` transitions to `Running`; `None` while - /// `Queued` or for entries that don't open a build log (`Restart`, - /// `PermChange` file-write phase, etc.). Links the queue card to - /// the live-streaming `/api/build-logs/id/{id}/stream` endpoint so - /// the operator can follow the nix build output in real time. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub build_log_id: Option, } /// How many terminal-state entries (`Done` / `Failed` / `Cancelled`) @@ -303,7 +286,6 @@ impl RebuildQueue { Vec::new(), None, None, - Vec::new(), ) } @@ -321,17 +303,7 @@ impl RebuildQueue { parent_id: Option, inputs: Vec, ) -> u64 { - self.enqueue_full( - kind, - agent, - source, - reason, - parent_id, - inputs, - None, - None, - Vec::new(), - ) + self.enqueue_full(kind, agent, source, reason, parent_id, inputs, None, None) } /// Enqueue a `PermChange` entry for `agent`. The worker applies the @@ -353,7 +325,6 @@ impl RebuildQueue { Vec::new(), None, Some(payload), - Vec::new(), ) } @@ -362,10 +333,10 @@ impl RebuildQueue { /// `enqueue_with_perm` delegate to this; the approval-driven POST /// handlers call it directly with the source row's id so the /// worker can re-fetch the kind-specific payload. - // 10 args: the queue entry has 6 independent submit-time fields plus - // four kind-specific payload fields (inputs, approval_id, perm_payload, - // depends_on). A builder struct would obscure the call sites; the - // shorter wrappers already cover all common cases. + // 9 args: the queue entry has 6 independent submit-time fields plus + // three kind-specific payload fields (inputs, approval_id, perm_payload). + // A builder struct would obscure the call sites; the shorter wrappers + // already cover all common cases. #[allow(clippy::too_many_arguments)] pub fn enqueue_full( &self, @@ -377,7 +348,6 @@ impl RebuildQueue { inputs: Vec, approval_id: Option, perm_payload: Option, - depends_on: Vec, ) -> u64 { let mut inner = self.inner.lock().expect("rebuild_queue mutex poisoned"); // Dedup against a pending entry with the same (kind, agent) — @@ -410,7 +380,6 @@ impl RebuildQueue { && entry.approval_id == approval_id && entry.parent_id == parent_id && perm_type_matches - && entry.depends_on == depends_on { if !entry.reason.contains(&reason) { use std::fmt::Write as _; @@ -437,8 +406,6 @@ impl RebuildQueue { approval_id, step: None, perm_payload, - depends_on, - build_log_id: None, }; inner.entries.push_back(entry); // Wake the worker. `notify_one` is a no-op when there's no @@ -447,47 +414,16 @@ impl RebuildQueue { id } - /// Pop the next `Queued` entry whose dependencies are resolved and - /// mark it `Running`. Returns the entry (a clone — the original - /// stays in the queue so live state reflects "this is currently - /// running"). Returns `None` when there's nothing queued OR every - /// queued entry has unresolved dependencies. - /// - /// A dependency is "resolved" when the dep's id is either: - /// - still in the queue AND in a terminal state (`Done` / `Failed` - /// / `Cancelled`), OR - /// - no longer in the queue (evicted by `trim_history` — only - /// terminal entries are ever evicted, so missing == completed). + /// Pop the next `Queued` entry and mark it `Running`. Returns the + /// entry (a clone — the original stays in the queue so live state + /// reflects "this is currently running"). Returns `None` when there's + /// nothing queued. pub fn take_next(&self) -> Option { let mut inner = self.inner.lock().expect("rebuild_queue mutex poisoned"); - // Collect ids that are still in the queue and terminal. Entries - // absent from the queue are also considered resolved (see above). - let terminal_ids: std::collections::HashSet = inner + let pos = inner .entries .iter() - .filter(|e| e.state.is_terminal()) - .map(|e| e.id) - .collect(); - // Active (non-terminal) ids: Queued + Running. Named `active_ids` - // rather than `queued_ids` because Running entries are included; - // used to distinguish "still in flight" from "evicted (= resolved)". - let active_ids: std::collections::HashSet = inner - .entries - .iter() - .filter(|e| !e.state.is_terminal()) - .map(|e| e.id) - .collect(); - let pos = inner.entries.iter().position(|e| { - e.state == QueueState::Queued - && e.depends_on.iter().all(|dep_id| { - // Resolved if terminal in queue OR not in queue at all. - // Note: circular deps (A depends on B, B depends on A) - // silently deadlock — neither entry ever becomes runnable. - // Not a problem in v1 (no callers yet), but callers must - // ensure acyclic dep graphs. - terminal_ids.contains(dep_id) || !active_ids.contains(dep_id) - }) - })?; + .position(|e| e.state == QueueState::Queued)?; let entry = &mut inner.entries[pos]; entry.state = QueueState::Running; entry.started_at = Some(now_unix()); @@ -536,24 +472,6 @@ impl RebuildQueue { true } - /// Link a `build_logs` row to a `Running` entry. Called by the - /// lifecycle worker when `nixos-container update` opens a build log - /// row so the dashboard can surface a "view logs" link while the - /// build is in flight. Returns `true` when the row was found and - /// the id was stored; `false` when the entry is no longer in the - /// queue or is not `Running`. - pub fn set_build_log_id(&self, id: u64, log_id: i64) -> bool { - let mut inner = self.inner.lock().expect("rebuild_queue mutex poisoned"); - let Some(entry) = inner.entries.iter_mut().find(|e| e.id == id) else { - return false; - }; - if entry.state != QueueState::Running { - return false; - } - entry.build_log_id = Some(log_id); - true - } - /// Snapshot the queue for `/api/state` and `RebuildQueueChanged`. /// Cheap clone — entries are small (~hundreds of bytes each). pub fn snapshot(&self) -> Vec { @@ -1213,8 +1131,6 @@ mod tests { None, Vec::new(), Some(42), - None, - Vec::new(), ); let snap = q.snapshot(); let entry = snap.iter().find(|e| e.id == id).expect("entry present"); @@ -1237,8 +1153,6 @@ mod tests { None, Vec::new(), Some(1), - None, - Vec::new(), ); let b = q.enqueue_full( QueueKind::Rebuild, @@ -1248,8 +1162,6 @@ mod tests { None, Vec::new(), Some(2), - None, - Vec::new(), ); assert_ne!(a, b); assert_eq!(q.snapshot().len(), 2); @@ -1263,8 +1175,6 @@ mod tests { None, Vec::new(), Some(1), - None, - Vec::new(), ); assert_eq!(a, c); assert_eq!(q.snapshot().len(), 2); @@ -1436,187 +1346,4 @@ mod tests { None ); } - - // --- depends_on tests --- - - /// An entry whose dep is not yet terminal must be skipped by - /// `take_next`; it runs only after the dep finishes. - #[test] - fn depends_on_blocks_until_dep_is_terminal() { - let q = RebuildQueue::new(); - let a = q.enqueue( - QueueKind::Rebuild, - "agent-a".to_owned(), - QueueSource::Manual, - "first".to_owned(), - None, - ); - let b = q.enqueue_full( - QueueKind::Rebuild, - "agent-b".to_owned(), - QueueSource::Manual, - "second (blocked on a)".to_owned(), - None, - Vec::new(), - None, - None, - vec![a], - ); - // B depends on A — take_next should give A first. - let first = q.take_next().expect("a is ready"); - assert_eq!(first.id, a); - // A is Running, not terminal — B must still be blocked. - assert!(q.take_next().is_none(), "b must be blocked while a runs"); - // Finish A → B should now be available. - q.finish(a, QueueState::Done, None); - let second = q.take_next().expect("b unblocked after a done"); - assert_eq!(second.id, b); - } - - /// An entry whose dep finished and was evicted from history is - /// treated as resolved (eviction only happens to terminal entries). - #[test] - fn depends_on_evicted_dep_counts_as_resolved() { - let q = RebuildQueue::new(); - // Fill the history cap for Rebuild so old terminals get evicted. - for i in 0..MAX_HISTORY_PER_KIND { - let id = q.enqueue( - QueueKind::Rebuild, - format!("filler-{i}"), - QueueSource::Manual, - "filler".to_owned(), - None, - ); - q.take_next(); - q.finish(id, QueueState::Done, None); - } - // `dep` gets enqueued, run, finished, and evicted by the - // next history-trimming call. - let dep = q.enqueue( - QueueKind::Rebuild, - "dep-agent".to_owned(), - QueueSource::Manual, - "dep".to_owned(), - None, - ); - q.take_next(); - // One more terminal to push `dep` out of the history window. - q.finish(dep, QueueState::Done, None); - let extra = q.enqueue( - QueueKind::Rebuild, - "extra".to_owned(), - QueueSource::Manual, - "extra".to_owned(), - None, - ); - q.take_next(); - q.finish(extra, QueueState::Done, None); - // `dep` should now be evicted. - assert!( - q.snapshot().iter().all(|e| e.id != dep), - "dep must be evicted from history" - ); - // An entry that depends on the (evicted) dep must be immediately runnable. - let downstream = q.enqueue_full( - QueueKind::Rebuild, - "downstream".to_owned(), - QueueSource::Manual, - "downstream (dep evicted = resolved)".to_owned(), - None, - Vec::new(), - None, - None, - vec![dep], - ); - let got = q.take_next().expect("downstream runnable when dep evicted"); - assert_eq!(got.id, downstream); - } - - /// Dedup respects `depends_on`: two otherwise-identical entries with - /// different dep sets are distinct and must NOT collapse. - #[test] - fn depends_on_is_part_of_dedup_key() { - let q = RebuildQueue::new(); - let dep1 = q.enqueue( - QueueKind::Rebuild, - "dep1".to_owned(), - QueueSource::Manual, - "d1".to_owned(), - None, - ); - let dep2 = q.enqueue( - QueueKind::Rebuild, - "dep2".to_owned(), - QueueSource::Manual, - "d2".to_owned(), - None, - ); - let a = q.enqueue_full( - QueueKind::Rebuild, - "target".to_owned(), - QueueSource::Manual, - "r".to_owned(), - None, - Vec::new(), - None, - None, - vec![dep1], - ); - // Same kind+agent but different depends_on — must NOT dedup. - let b = q.enqueue_full( - QueueKind::Rebuild, - "target".to_owned(), - QueueSource::Manual, - "r".to_owned(), - None, - Vec::new(), - None, - None, - vec![dep2], - ); - assert_ne!(a, b, "different depends_on must produce distinct entries"); - // Same depends_on as a — must dedup. - let c = q.enqueue_full( - QueueKind::Rebuild, - "target".to_owned(), - QueueSource::Manual, - "r again".to_owned(), - None, - Vec::new(), - None, - None, - vec![dep1], - ); - assert_eq!(a, c, "identical depends_on must dedup"); - } - - /// An entry with Failed dep is still resolved — the dependent runs - /// regardless of whether its upstream succeeded or not. Callers that - /// need to abort on dep failure should cancel the downstream manually. - #[test] - fn depends_on_failed_dep_counts_as_resolved() { - let q = RebuildQueue::new(); - let a = q.enqueue( - QueueKind::Rebuild, - "a".to_owned(), - QueueSource::Manual, - "a".to_owned(), - None, - ); - let b = q.enqueue_full( - QueueKind::Rebuild, - "b".to_owned(), - QueueSource::Manual, - "b (blocked on a)".to_owned(), - None, - Vec::new(), - None, - None, - vec![a], - ); - q.take_next(); // pop a, mark Running - q.finish(a, QueueState::Failed, Some("nix build exploded".to_owned())); - let got = q.take_next().expect("b runnable after a failed"); - assert_eq!(got.id, b); - } } diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs index 0fc79567..12cacc92 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -161,7 +161,7 @@ async fn dispatch(req: &HostRequest, coord: Arc) -> HostResponse { let agent_dir = coord.ensure_runtime(name)?; let hive = coord.hive_env(); let paths = Coordinator::agent_paths(name, agent_dir); - let result = lifecycle::rebuild(name, &hive, &paths, &|_| (), &|_| ()).await; + let result = lifecycle::rebuild(name, &hive, &paths, &|_| ()).await; // Mirror auto_update::rebuild_agent — the manager wants // to know about every rebuild attempt regardless of // which surface triggered it, especially failures