//! Global rebuild queue — serialises all long-running container/meta //! operations (rebuild, meta-update, first-spawn) through a single //! background worker. Design rationale, kind taxonomy, dedup rules, //! cascade parent tracking, and step labels: //! `docs/coordinator.md::Rebuild queue`. use std::collections::VecDeque; use std::sync::Mutex; use anyhow::Context as _; use serde::{Deserialize, Serialize}; use tokio::sync::Notify; /// What the queue can run. Each variant maps to a specific worker /// execution path; `agent` (in `QueueEntry`) names the target where /// relevant. #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize)] #[serde(rename_all = "snake_case")] pub enum QueueKind { /// Rebuild a single agent's container (`auto_update::rebuild_agent`). Rebuild, /// Run `nix flake update` on the meta flake. Triggers cascade /// `Rebuild` entries (with `parent_id`) once the lock bump lands. MetaUpdate, /// First-deploy spawn of a new agent (approval-driven). Spawn, /// Destroy with `--purge` (real fs work). Not yet routed here; the /// variant exists so the wire shape doesn't need to change later. #[allow(dead_code, reason = "wire shape — routed by a future PR")] Destroy, /// hive-c0re boot-time sweep: bumps the meta hyperhive lock then /// enqueues a `Rebuild` child for every managed container. Completes /// after the lock bump; children run as independent queue entries /// grouped under this parent's `id`. `agent` = `"hyperhive"`. StartupSweep, /// Stop + start a container without touching config. Fast op (~5-10s). /// Queued so it serialises against in-flight rebuilds for the same /// agent — prevents a restart racing a rebuild mid-flight. Restart, /// Write a tool-group or capability change to the shared JSON file, /// then rebuild the agent so the new env var takes effect. /// Serialised through the queue so concurrent dashboard batch-apply /// actions for different agents never race on the shared JSON file. PermChange, /// Gracefully stop a container: signal the harness to run one /// stop-checkpoint turn (flush durable `/state`), wait for it to drain, /// then `nixos-container stop`. Falls back to a hard stop on timeout. GracefulStop, /// Start a stopped container (`lifecycle::start`). Routed through the /// queue so the dashboard shows a visible queued→running transient — a /// direct sub-second start only flashes the badge — and bulk starts /// serialise legibly on the queue. Fast op. Start, /// Hard-stop a container (`lifecycle::kill`), no quiesce. Routed through /// the queue for the same visible-progress reason as `Start`; the /// quiescing variant is `GracefulStop`. Fast op. Stop, } impl QueueKind { pub fn as_str(self) -> &'static str { match self { QueueKind::Rebuild => "rebuild", QueueKind::MetaUpdate => "meta_update", QueueKind::Spawn => "spawn", QueueKind::Destroy => "destroy", QueueKind::StartupSweep => "startup_sweep", QueueKind::Restart => "restart", QueueKind::PermChange => "perm_change", QueueKind::GracefulStop => "graceful_stop", QueueKind::Start => "start", QueueKind::Stop => "stop", } } } /// Kind-specific payload for `QueueKind::PermChange` entries. /// Carries the desired new value so the worker can apply the file /// write (serialised, in FIFO order) without racing concurrent HTTP /// handlers writing to the same shared JSON file. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(tag = "type", rename_all = "snake_case")] pub enum PermPayload { /// Set the tool groups for one agent (`tool-groups.json`). ToolGroups { groups: Vec }, /// Set the capabilities for one agent (`capabilities.json`). Capabilities { caps: Vec }, /// Set both perm-types for one agent in a single entry — the batch /// `POST /api/permissions` path. Either field `None` leaves that /// file untouched (no write, no commit); the worker commits whichever /// are present in one git commit, then rebuilds once. Collapses the /// dedup key to `(kind, agent)` so caps + groups for one agent /// produce a single rebuild rather than two. Combined { groups: Option>, caps: Option>, }, } /// Where the enqueue request originated. Drives the "why" chip on the /// dashboard and lets the UI group cascade entries under their parent /// without parsing the reason text. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] #[serde(rename_all = "snake_case")] pub enum QueueSource { /// Operator clicked rebuild / update-all / meta-update on the /// dashboard, or any other direct human action (CLI, manager tool). Manual, /// Spawned as a cascade from a `MetaUpdate` entry's lock-bump /// fan-out. The `parent_id` on the `QueueEntry` points back at /// the originating meta-update. MetaUpdate, /// `auto_update::run` startup sweep — rebuild every container on /// hive-c0re boot. Legacy flat source (no parent); replaced by /// `StartupSweep` for the parent entry and child rebuilds once the /// queue introduced `parent_id` grouping. Kept for wire compatibility /// with entries logged before the migration. AutoUpdate, /// Direct child of a `StartupSweep` queue entry — one per agent in /// the boot-time rebuild sweep. Carries `parent_id` back-link so /// the dashboard renders the sweep's per-agent rebuilds nested under /// the parent header. The parent entry itself uses `QueueSource::AutoUpdate` /// (automated boot action, not operator-driven). StartupSweep, /// Crash recovery path (future use — currently no auto-rebuild on /// crash, but the variant exists for the imminent feature). #[allow(dead_code, reason = "wire shape — used by a future feature")] CrashRecover, /// Operator approved a pending `Approval` row on the dashboard. /// `QueueEntry.approval_id` points back at the source row so the /// worker can fetch the kind-specific payload (`commit_ref`, inputs, /// description) before dispatching. Approval, } impl QueueSource { pub fn as_str(self) -> &'static str { match self { QueueSource::Manual => "manual", QueueSource::MetaUpdate => "meta_update", QueueSource::AutoUpdate => "auto_update", QueueSource::StartupSweep => "startup_sweep", QueueSource::CrashRecover => "crash_recover", QueueSource::Approval => "approval", } } } /// Lifecycle state of an entry. `Done` / `Failed` / `Cancelled` are /// retained in the queue snapshot for a short tail (`MAX_HISTORY_PER_KIND`) /// so the dashboard can show "last few" runs alongside live state. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] #[serde(rename_all = "snake_case")] pub enum QueueState { Queued, Running, Done, Failed, Cancelled, } impl QueueState { pub fn is_terminal(self) -> bool { matches!( self, QueueState::Done | QueueState::Failed | QueueState::Cancelled ) } } /// A single queue entry — what's pending, running, or recently finished. /// Serialised verbatim onto the dashboard event channel and the /// `/api/state` snapshot. #[derive(Debug, Clone, Serialize)] pub struct QueueEntry { /// Monotonic per-process id. Stable for the lifetime of the entry /// so SSE upserts land in place rather than churning the list. pub id: u64, /// Target agent name, or the literal `"hyperhive"` for entries /// (`MetaUpdate`) that affect the meta flake rather than a single /// agent. pub agent: String, pub kind: QueueKind, pub state: QueueState, pub source: QueueSource, /// Groups cascade entries under their originating parent. For a /// `MetaUpdate` entry this is `None`; for the per-agent rebuilds /// the worker enqueues after the lock bump it's `Some(meta_id)`. #[serde(default, skip_serializing_if = "Option::is_none")] pub parent_id: Option, /// Human-readable "why" — populated by the enqueuer (`"manual via /// dashboard"`, `"meta-update cascade (hyperhive bumped)"`, /// `"startup sweep"`). Free-form; dedup appends `(also requested /// by …)` lines on repeated enqueues. pub reason: String, pub enqueued_at: i64, #[serde(default, skip_serializing_if = "Option::is_none")] pub started_at: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub finished_at: Option, /// Populated when `state == Failed`. Carries the worker's error /// string (already truncated to a reasonable length by the caller). #[serde(default, skip_serializing_if = "Option::is_none")] pub error: Option, /// `MetaUpdate`-only payload: the list of meta flake inputs to run /// through `nix flake update`. Empty / absent on `Rebuild` / /// `Spawn` / `Destroy` entries; absent on the wire (never /// serialised) when the entry kind doesn't have meaningful inputs. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub inputs: Vec, /// Source approval row id when this entry was created by an /// operator-approve POST (`source == Approval`). The worker uses /// it to re-fetch the kind-specific payload (`commit_ref` / inputs / /// description / `fetched_sha`) and to fire `ApprovalResolved` on /// completion. `None` for non-approval entries — preserved on /// the wire that way too. #[serde(default, skip_serializing_if = "Option::is_none")] pub approval_id: Option, /// Current sub-step inside the running entry. Worker mutates this /// as the kind-specific pipeline /// advances through phases (e.g. `"plant tags"` → /// `"nixos-container update"` → `"finalize deploy"`). `None` while /// `Queued` and after terminal — only meaningful with /// `state == Running`. Each transition fires a fresh /// `RebuildQueueChanged` snapshot so the dashboard can render /// the label as a sub-line on the queue card. Free-form per /// pipeline; the kind-specific worker is the source of truth. #[serde(default, skip_serializing_if = "Option::is_none")] pub step: Option, /// `PermChange`-only payload: the desired new permission value to /// apply. Absent (`None`) on all other entry kinds — omitted from /// 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`) /// to retain per kind in the snapshot. Older entries get evicted to /// keep `/api/state` tight; the live event channel is unaffected. const MAX_HISTORY_PER_KIND: usize = 5; /// Inner state guarded by a single mutex. Held briefly — every /// operation is constant-time relative to the queue's depth, and /// the depths in practice are tiny (single-digit). #[derive(Debug, Default)] struct Inner { entries: VecDeque, next_id: u64, } /// Global rebuild queue. Lives on `Coordinator` (one per hive-c0re /// process). The associated `Notify` wakes the worker when something /// new arrives. #[derive(Debug)] pub struct RebuildQueue { inner: Mutex, /// Worker wakes on this signal. The worker checks the queue and /// loops back to `notified().await` when there's nothing to run. pub(crate) notify: Notify, } impl Default for RebuildQueue { fn default() -> Self { Self { inner: Mutex::new(Inner::default()), notify: Notify::new(), } } } /// Full-shape submit spec for [`RebuildQueue::enqueue_full`] — every /// `QueueEntry` field settable at submit time. The thinner `enqueue` /// / `enqueue_with_inputs` / `enqueue_with_perm` wrappers build this /// for the common cases. pub struct FullEnqueue { pub kind: QueueKind, pub agent: String, pub source: QueueSource, pub reason: String, pub parent_id: Option, pub inputs: Vec, pub approval_id: Option, pub perm_payload: Option, pub depends_on: Vec, } impl RebuildQueue { pub fn new() -> Self { Self::default() } /// Add an entry to the queue. Returns the entry's id (newly-allocated /// or — on dedup — the existing entry's id with the new reason /// appended). /// /// Dedup rule: /// - `Rebuild` / `Spawn` / `Destroy`: a `Queued` entry with the same /// `(kind, agent, parent_id)` swallows the new request. `parent_id` /// is part of the key so that a `MetaUpdate` cascade rebuild (with a /// specific `parent_id`) never collapses into a standalone rebuild or /// a cascade from a different `MetaUpdate`. Without this guard a /// cascade rebuild pre-enqueued before the lock bump would be swallowed /// by an existing `Queued` startup-sweep rebuild, causing the agent to /// never rebuild against the post-bump meta. /// - `MetaUpdate`: dedup ALSO requires the `inputs` field to match — /// two meta-updates with different input lists are distinct work /// and must queue separately, otherwise the second meta-update /// would silently collapse into the first whenever it was still /// `Queued`, losing the second's input set. /// /// Running and terminal entries never dedup — operators are free /// to re-queue a rebuild that's currently running (something /// changed since it started) or re-run one that just finished. pub fn enqueue( &self, kind: QueueKind, agent: String, source: QueueSource, reason: String, parent_id: Option, ) -> u64 { self.enqueue_full(FullEnqueue { kind, agent, source, reason, parent_id, inputs: Vec::new(), approval_id: None, perm_payload: None, depends_on: Vec::new(), }) } /// Same as `enqueue` but carries an `inputs` payload — used by /// `MetaUpdate` enqueues to tell the worker which meta-flake /// inputs to bump. For `MetaUpdate` the `inputs` value is part of /// the dedup key (two meta-updates with different inputs are /// distinct operations). pub fn enqueue_with_inputs( &self, kind: QueueKind, agent: String, source: QueueSource, reason: String, parent_id: Option, inputs: Vec, ) -> u64 { self.enqueue_full(FullEnqueue { kind, agent, source, reason, parent_id, inputs, approval_id: None, perm_payload: None, depends_on: Vec::new(), }) } /// Enqueue a `PermChange` entry for `agent`. The worker applies the /// JSON file write (serialised through FIFO) then rebuilds the /// container so the updated env var takes effect. pub fn enqueue_with_perm( &self, agent: String, source: QueueSource, reason: String, payload: PermPayload, ) -> u64 { self.enqueue_full(FullEnqueue { kind: QueueKind::PermChange, agent, source, reason, parent_id: None, inputs: Vec::new(), approval_id: None, perm_payload: Some(payload), depends_on: Vec::new(), }) } /// Full-shape enqueue — every `QueueEntry` field that's settable /// at submit time. Existing `enqueue` / `enqueue_with_inputs` / /// `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. pub fn enqueue_full(&self, spec: FullEnqueue) -> u64 { let FullEnqueue { kind, agent, source, reason, parent_id, inputs, approval_id, perm_payload, depends_on, } = spec; let mut inner = self.inner.lock().expect("rebuild_queue mutex poisoned"); // Dedup against a pending entry with the same (kind, agent) — // and, for MetaUpdate, the same `inputs` list (see method // docstring for why). Approval-driven entries also require the // approval_id to match so two distinct approvals for the same // agent never collapse into one queue slot. Rebuild (and Spawn / // Destroy) entries also require parent_id to match so a // MetaUpdate cascade rebuild is never swallowed by an unrelated // queued rebuild (e.g. from the startup sweep). PermChange // entries additionally check the perm type discriminant — a // tool-groups change and a capabilities change for the same // agent are distinct operations and must not collapse into one. for entry in &mut inner.entries { let perm_type_matches = matches!( (&entry.perm_payload, &perm_payload), ( Some(PermPayload::ToolGroups { .. }), Some(PermPayload::ToolGroups { .. }) ) | ( Some(PermPayload::Capabilities { .. }), Some(PermPayload::Capabilities { .. }) ) | ( Some(PermPayload::Combined { .. }), Some(PermPayload::Combined { .. }) ) | (None, None) ); if entry.state == QueueState::Queued && entry.kind == kind && entry.agent == agent && (kind != QueueKind::MetaUpdate || entry.inputs == inputs) && 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 _; let _ = write!(entry.reason, "\nalso requested by: {reason}"); } return entry.id; } } inner.next_id += 1; let id = inner.next_id; let entry = QueueEntry { id, agent, kind, state: QueueState::Queued, source, parent_id, reason, enqueued_at: now_unix(), started_at: None, finished_at: None, error: None, inputs, 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 // waiter; the next `notified().await` returns immediately. self.notify.notify_one(); 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). 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 .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) }) })?; let entry = &mut inner.entries[pos]; entry.state = QueueState::Running; entry.started_at = Some(now_unix()); Some(entry.clone()) } /// Mark an entry terminal. `error` is populated for `Failed`; /// `Done` / `Cancelled` ignore it. Trims the history tail. /// Clears `step` — the field is only meaningful while `Running`, /// and leaving a stale "in flight" label after a terminal /// transition would mislead the dashboard render. pub fn finish(&self, id: u64, state: QueueState, error: Option) { debug_assert!( state.is_terminal(), "finish() called with non-terminal {state:?}" ); let mut inner = self.inner.lock().expect("rebuild_queue mutex poisoned"); if let Some(entry) = inner.entries.iter_mut().find(|e| e.id == id) { entry.state = state; entry.finished_at = Some(now_unix()); entry.error = error.filter(|_| state == QueueState::Failed); entry.step = None; } Self::trim_history(&mut inner); } /// Set the current sub-step label on a `Running` entry. /// Returns `true` when the row was found AND the label changed /// (caller should emit a `RebuildQueueChanged` snapshot only on /// `true` to avoid noisy duplicate frames). No-op for entries not /// in `Running` — the field is conceptually undefined outside /// that state. pub fn set_step(&self, id: u64, step: impl Into) -> bool { let new_step = step.into(); 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; } if entry.step.as_deref() == Some(new_step.as_str()) { return false; } entry.step = Some(new_step); 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 { let inner = self.inner.lock().expect("rebuild_queue mutex poisoned"); inner.entries.iter().cloned().collect() } /// Cancel every `Queued` entry whose `parent_id` matches `parent`. /// Used when a `MetaUpdate` parent fails its lock bump — the /// cascade rebuilds the enqueuer pre-queued no longer apply /// (nothing actually changed, so they'd be wasted work). Running /// children are left alone — they were started under the parent's /// assumption and can't be cleanly aborted from the queue side. /// Returns the count of cancelled entries. pub fn cancel_children(&self, parent: u64) -> usize { let mut inner = self.inner.lock().expect("rebuild_queue mutex poisoned"); let mut count = 0; for entry in &mut inner.entries { if entry.parent_id == Some(parent) && entry.state == QueueState::Queued { entry.state = QueueState::Cancelled; entry.finished_at = Some(now_unix()); count += 1; } } if count > 0 { Self::trim_history(&mut inner); } count } /// Cancel a `Queued` entry (no-op for `Running` / terminal — the /// in-flight rebuild owns the agent's nix store and can't be /// safely interrupted). Returns true when an entry was cancelled. pub fn cancel(&self, id: u64) -> bool { let mut inner = self.inner.lock().expect("rebuild_queue mutex poisoned"); if let Some(entry) = inner.entries.iter_mut().find(|e| e.id == id) && entry.state == QueueState::Queued { entry.state = QueueState::Cancelled; entry.finished_at = Some(now_unix()); Self::trim_history(&mut inner); return true; } false } /// Keep only the most recent `MAX_HISTORY_PER_KIND` terminal entries /// per kind. Pending + running entries are never evicted. fn trim_history(inner: &mut Inner) { let mut counts: std::collections::HashMap = std::collections::HashMap::new(); // Walk newest-first; keep the first MAX_HISTORY_PER_KIND // terminals per kind, evict the rest. let entries: Vec = inner .entries .iter() .rev() .filter(|e| { if !e.state.is_terminal() { return true; } let n = counts.entry(e.kind).or_insert(0); *n += 1; *n <= MAX_HISTORY_PER_KIND }) .cloned() .collect(); inner.entries = entries.into_iter().rev().collect(); } } /// Background worker that drains the queue. Spawned once at hive-c0re /// startup from `main.rs`. Loops forever: /// 1. Pop the next `Queued` entry (`take_next` marks it `Running` and /// fires a `RebuildQueueChanged` snapshot via the caller). /// 2. Dispatch by kind — single-agent rebuild, meta-update + cascade, /// or first-spawn. /// 3. Mark the entry terminal (`finish`) and emit another snapshot. /// 4. When the queue is empty, `await` on `notify` until something /// new lands. /// /// Shutdown semantics: subscribes to `coord.shutdown_rx()`. On a true /// signal the worker exits after its current entry finishes; pending /// `Queued` entries are dropped (they'll either be replayed by the /// startup sweep on next boot or left for an operator to re-queue). /// Max time the `GracefulStop` worker waits for the harness to run its /// stop-checkpoint turn + drain before falling back to a hard container stop. /// Generous — a checkpoint turn can take a while — but bounded so a wedged /// agent never blocks the stop indefinitely. const GRACEFUL_STOP_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(3); pub async fn run_worker(coord: std::sync::Arc) { let mut shutdown = coord.shutdown_rx(); loop { // Drain everything available now. while let Some(entry) = coord.rebuild_queue.take_next() { coord.emit_rebuild_queue_snapshot(); tracing::info!( id = entry.id, kind = entry.kind.as_str(), agent = %entry.agent, source = entry.source.as_str(), "rebuild_queue: running" ); let result = dispatch(&coord, &entry).await; match result { Ok(()) => { coord.rebuild_queue.finish(entry.id, QueueState::Done, None); tracing::info!(id = entry.id, "rebuild_queue: done"); } Err(e) => { let msg = format!("{e:#}"); let truncated = if msg.len() > 2_000 { format!("{}…", &msg[..2_000]) } else { msg.clone() }; coord .rebuild_queue .finish(entry.id, QueueState::Failed, Some(truncated)); tracing::warn!(id = entry.id, error = %msg, "rebuild_queue: failed"); } } coord.emit_rebuild_queue_snapshot(); } // Park until something new is enqueued OR shutdown fires. tokio::select! { biased; res = shutdown.changed() => { if res.is_err() || *shutdown.borrow() { tracing::info!("rebuild_queue: worker exiting on shutdown"); return; } } () = coord.rebuild_queue.notify.notified() => { // New entry — back to the drain loop. } } } } /// Run a single queue entry to completion. Kind-dispatched; failures /// bubble up to the worker which marks the entry `Failed`. /// /// Approval-driven entries (`approval_id.is_some()`) route through /// `actions::run_approval_*` which carry the kind-specific commit /// pipeline + the `ApprovalResolved` event fan-out. Non-approval /// entries hit the original auto/manual rebuild paths. async fn dispatch( coord: &std::sync::Arc, entry: &QueueEntry, ) -> anyhow::Result<()> { match (entry.kind, entry.approval_id) { (QueueKind::Rebuild, Some(approval_id)) => { crate::actions::run_approval_apply_commit(coord, Some(entry.id), approval_id).await } (QueueKind::Rebuild, None) => { let current_rev = crate::auto_update::current_flake_rev(&coord.hyperhive_flake).unwrap_or_default(); crate::auto_update::rebuild_agent(coord, &entry.agent, ¤t_rev, Some(entry.id)) .await } (QueueKind::MetaUpdate, Some(approval_id)) => { crate::actions::run_approval_update_meta_inputs(coord, Some(entry.id), approval_id) .await } (QueueKind::MetaUpdate, None) => run_meta_update(coord, entry).await, (QueueKind::Spawn, Some(approval_id)) => { crate::actions::run_approval_spawn(coord, Some(entry.id), approval_id).await } (QueueKind::Spawn, None) => { // Unreachable today: every Spawn entry is born from an // approval (HostRequest::RequestSpawn → submit_kind → // approve → enqueue with approval_id). The manager-side // `RequestSpawn` surface that used to bypass approvals // was removed; if a future direct-spawn admin path needs // to skip the approval ride it should wire its own action // call rather than route through here. anyhow::bail!( "rebuild_queue: Spawn entry id={} agent={} arrived without an approval_id — \ nothing should enqueue this shape today", entry.id, entry.agent, ) } (QueueKind::Destroy, _) => { // Reserved for future `destroy --purge` integration. anyhow::bail!("Destroy kind not yet implemented in rebuild_queue worker"); } (QueueKind::StartupSweep, _) => { // Bump meta's hyperhive input before per-agent child rebuilds // run so they build against the latest base. Non-fatal on // failure — child rebuilds proceed regardless. After the bump // (or failure) this entry transitions to Done and the worker // drains the pre-enqueued child Rebuild entries. coord.set_queue_step(Some(entry.id), "nix flake update hyperhive"); if let Err(e) = crate::meta::lock_update_hyperhive().await { tracing::warn!(error = ?e, "startup_sweep: meta lock_update_hyperhive failed"); } // `finish` clears the step label; no explicit clear needed here. Ok(()) } (QueueKind::Restart, _) => { let name = &entry.agent; let _guard = coord.transient_guard(name, crate::coordinator::TransientKind::Restarting); coord.set_queue_step(Some(entry.id), "nixos-container restart"); crate::lifecycle::restart(name).await?; coord.kick_agent(name, "container restarted"); coord.rescan_containers_and_emit().await; Ok(()) } (QueueKind::PermChange, _) => { let name = &entry.agent; // Write + commit the perm file under META_LOCK so the // working tree is never left dirty between the file write // and the subsequent prepare_deploy git operations. coord.set_queue_step(Some(entry.id), "writing + committing perm file"); match &entry.perm_payload { Some(PermPayload::ToolGroups { groups }) => { crate::meta::commit_tool_groups(name, groups) .await .with_context(|| format!("commit tool-groups for {name}"))?; // Emit after the commit so the P3RM1SS10NS tab // reflects the new assignment without the operator // needing to navigate away and back. coord.emit_tool_groups_snapshot(); } Some(PermPayload::Capabilities { caps }) => { crate::meta::commit_capabilities(name, caps) .await .with_context(|| format!("commit capabilities for {name}"))?; coord.emit_capabilities_snapshot(); } Some(PermPayload::Combined { groups, caps }) => { // Batch perm change: commit whichever file(s) are // present in a single git commit, then the rebuild // below runs once — no double-rebuild for an agent // whose caps AND groups both changed. crate::meta::commit_perms(name, groups.as_deref(), caps.as_deref()) .await .with_context(|| format!("commit perms for {name}"))?; if groups.is_some() { coord.emit_tool_groups_snapshot(); } if caps.is_some() { coord.emit_capabilities_snapshot(); } } None => { anyhow::bail!( "PermChange entry id={} agent={} is missing perm_payload", entry.id, entry.agent, ); } } // Now rebuild so the updated HIVE_TOOL_GROUPS / HIVE_CAPABILITIES // env var takes effect in the container. let current_rev = crate::auto_update::current_flake_rev(&coord.hyperhive_flake).unwrap_or_default(); crate::auto_update::rebuild_agent(coord, name, ¤t_rev, Some(entry.id)).await } (QueueKind::GracefulStop, _) => run_graceful_stop(coord, entry).await, (QueueKind::Start, _) => run_start(coord, entry).await, (QueueKind::Stop, _) => run_stop(coord, entry).await, } } /// Start a stopped container off the queue (`QueueKind::Start`), with a /// `Starting` transient so the dashboard shows a visible queued→running /// progression rather than the sub-second flash of a direct start. async fn run_start( coord: &std::sync::Arc, entry: &QueueEntry, ) -> anyhow::Result<()> { let name = &entry.agent; let _guard = coord.transient_guard(name, crate::coordinator::TransientKind::Starting); coord.set_queue_step(Some(entry.id), "nixos-container start"); crate::lifecycle::start(name).await?; coord.kick_agent(name, "container started"); coord.rescan_containers_and_emit().await; Ok(()) } /// Hard-stop a container off the queue (`QueueKind::Stop`) — same teardown as /// a direct kill (unregister + `Killed` event), but with a `Stopping` /// transient for visible queue progress. The quiescing variant is /// `run_graceful_stop`. async fn run_stop( coord: &std::sync::Arc, entry: &QueueEntry, ) -> anyhow::Result<()> { let name = &entry.agent; let _guard = coord.transient_guard(name, crate::coordinator::TransientKind::Stopping); coord.set_queue_step(Some(entry.id), "nixos-container stop"); crate::lifecycle::kill(name).await?; coord.unregister_agent(name); coord.notify_manager(&hive_sh4re::HelperEvent::Killed { agent: name.clone(), }); coord.rescan_containers_and_emit().await; Ok(()) } /// Run one `GracefulStop` entry: signal the harness to quiesce (it returns /// `GracefulStop` on its next `Recv`, runs one stop-checkpoint turn to flush /// durable `/state`, then exits), wait for it to drain — bounded by /// `GRACEFUL_STOP_TIMEOUT` so a wedged agent can't block forever — then stop /// the container with the same teardown as a plain kill. async fn run_graceful_stop( coord: &std::sync::Arc, entry: &QueueEntry, ) -> anyhow::Result<()> { let name = &entry.agent; let _guard = coord.transient_guard(name, crate::coordinator::TransientKind::Stopping); // Signal the harness; the kick breaks an idle long-poll so it's seen promptly. coord.set_queue_step(Some(entry.id), "graceful stop: signalling agent"); coord.mark_graceful_stop(name); coord.kick_agent(name, "graceful stop requested"); // Wait for the harness to drain (it clears the flag via `GracefulStopComplete`) // or fall back to a hard stop after the timeout. The single queue worker is // intentionally held for the duration — graceful stops are infrequent. coord.set_queue_step(Some(entry.id), "graceful stop: waiting for agent to drain"); let deadline = std::time::Instant::now() + GRACEFUL_STOP_TIMEOUT; while coord.is_graceful_stop_pending(name) { if std::time::Instant::now() >= deadline { tracing::warn!(agent = %name, "graceful stop: drain timed out — hard-stopping"); break; } tokio::time::sleep(std::time::Duration::from_millis(500)).await; } coord.clear_graceful_stop(name); // Stop the container — same teardown as a plain kill. coord.set_queue_step(Some(entry.id), "nixos-container stop"); crate::lifecycle::kill(name).await?; coord.unregister_agent(name); coord.notify_manager(&hive_sh4re::HelperEvent::Killed { agent: name.clone(), }); coord.rescan_containers_and_emit().await; Ok(()) } /// Run one `MetaUpdate` entry: bump the meta flake's locks for the /// requested inputs, then enqueue a cascade of `Rebuild` entries /// (with `parent_id` set to this entry's id) for every agent affected /// by the bump. Mirrors the previous `dashboard::run_meta_update` /// semantics; that path now enqueues into this queue rather than /// running the bump + rebuild loop inline. async fn run_meta_update( coord: &std::sync::Arc, entry: &QueueEntry, ) -> anyhow::Result<()> { let _progress = coord.meta_update_guard(); let inputs = entry.inputs.clone(); tracing::info!( ?inputs, parent = entry.id, "rebuild_queue: meta-update starting" ); coord.set_queue_step(Some(entry.id), "nix flake update"); let result = if inputs.is_empty() { crate::meta::lock_update(&[]).await } else { crate::meta::lock_update(&inputs).await }; if let Err(e) = result { // Lock bump failed — cancel any pending cascade rebuilds the // enqueuer pre-queued for this MetaUpdate. Their parent_id // matches this entry; the children no longer make sense (we // never bumped the lock that justified them). let cancelled = coord.rebuild_queue.cancel_children(entry.id); if cancelled > 0 { tracing::warn!( cancelled, parent = entry.id, "rebuild_queue: meta-update failed; cancelled cascade rebuilds" ); coord.emit_rebuild_queue_snapshot(); } return Err(e); } // Lock file changed — meta-inputs panel re-renders. The cascade // rebuilds were already enqueued at MetaUpdate submission time, // so no further enqueue is needed here. crate::dashboard::emit_meta_inputs_snapshot(coord.as_ref()); Ok(()) } /// Compute which agents a `nix flake update ` on the meta /// flake would affect. Used by callers that pre-enqueue cascade /// `Rebuild` entries at `MetaUpdate` submission time so the dashboard /// can render the dependent work alongside its parent before the lock /// bump actually runs. /// /// Mirrors `run_meta_update`'s post-bump fan-out logic. Empty `inputs` /// or any input under `hyperhive` → every container; otherwise just /// the agents named by `agent-` inputs. pub async fn meta_update_cascade_agents(inputs: &[String]) -> Vec { let touched_hyperhive = inputs .iter() .any(|i| i == "hyperhive" || i.starts_with("hyperhive/")); let touched_agents: Vec = inputs .iter() .filter_map(|i| i.strip_prefix("agent-")) .map(|rest| rest.split('/').next().unwrap_or(rest).to_owned()) .collect(); let mut names = if touched_hyperhive || inputs.is_empty() { crate::lifecycle::list() .await .unwrap_or_default() .into_iter() .filter_map(|c| { c.strip_prefix(crate::lifecycle::AGENT_PREFIX) .map(str::to_owned) }) .collect() } else { touched_agents }; // Sort parents before children so the sequential queue worker // always rebuilds a parent before any of its dependents. let topo = crate::topology::read(); crate::auto_update::topology_sort(&mut names, &topo); names } /// Current unix timestamp in seconds. `now()` calls are pulled into a /// helper so tests can swap them out later. fn now_unix() -> i64 { std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .ok() .and_then(|d| i64::try_from(d.as_secs()).ok()) .unwrap_or(0) } #[cfg(test)] mod tests { use super::*; #[test] fn enqueue_and_take_in_order() { let q = RebuildQueue::new(); let a = q.enqueue( QueueKind::Rebuild, "agent-a".to_owned(), QueueSource::Manual, "first".to_owned(), None, ); let b = q.enqueue( QueueKind::Rebuild, "agent-b".to_owned(), QueueSource::Manual, "second".to_owned(), None, ); assert_ne!(a, b); let next = q.take_next().expect("queued"); assert_eq!(next.id, a); assert_eq!(next.state, QueueState::Running); let next = q.take_next().expect("queued"); assert_eq!(next.id, b); assert!(q.take_next().is_none()); } #[test] fn dedup_pending_same_kind_and_agent() { let q = RebuildQueue::new(); let a = q.enqueue( QueueKind::Rebuild, "agent-a".to_owned(), QueueSource::Manual, "first".to_owned(), None, ); let b = q.enqueue( QueueKind::Rebuild, "agent-a".to_owned(), QueueSource::AutoUpdate, "auto sweep".to_owned(), None, ); assert_eq!(a, b, "dedup should return existing id"); let snap = q.snapshot(); assert_eq!(snap.len(), 1); assert!(snap[0].reason.contains("first")); assert!(snap[0].reason.contains("auto sweep")); } #[test] fn meta_update_dedup_matches_inputs() { // Two MetaUpdate enqueues with identical inputs → dedup. let q = RebuildQueue::new(); let a = q.enqueue_with_inputs( QueueKind::MetaUpdate, "hyperhive".to_owned(), QueueSource::Manual, "first".to_owned(), None, vec!["nixpkgs".to_owned()], ); let b = q.enqueue_with_inputs( QueueKind::MetaUpdate, "hyperhive".to_owned(), QueueSource::Manual, "duplicate click".to_owned(), None, vec!["nixpkgs".to_owned()], ); assert_eq!(a, b, "identical-inputs meta-updates should dedup"); assert_eq!(q.snapshot().len(), 1); } #[test] fn meta_update_dedup_separates_distinct_inputs() { // Two MetaUpdate enqueues with DIFFERENT inputs → distinct // entries, not deduped. let q = RebuildQueue::new(); let a = q.enqueue_with_inputs( QueueKind::MetaUpdate, "hyperhive".to_owned(), QueueSource::Manual, "bump nixpkgs".to_owned(), None, vec!["nixpkgs".to_owned()], ); let b = q.enqueue_with_inputs( QueueKind::MetaUpdate, "hyperhive".to_owned(), QueueSource::Manual, "bump bitburner-agent".to_owned(), None, vec!["agent-bitburner/bitburner-agent".to_owned()], ); assert_ne!(a, b, "different-inputs meta-updates must NOT dedup"); let snap = q.snapshot(); assert_eq!(snap.len(), 2); // Both inputs lists are preserved. let inputs: Vec<&[String]> = snap.iter().map(|e| e.inputs.as_slice()).collect(); assert!(inputs.iter().any(|i| *i == ["nixpkgs"])); assert!( inputs .iter() .any(|i| *i == ["agent-bitburner/bitburner-agent"]) ); } #[test] fn dedup_does_not_apply_across_kinds_or_agents() { let q = RebuildQueue::new(); let a = q.enqueue( QueueKind::Rebuild, "agent-a".to_owned(), QueueSource::Manual, "r".to_owned(), None, ); let b = q.enqueue( QueueKind::Rebuild, "agent-b".to_owned(), QueueSource::Manual, "r".to_owned(), None, ); let c = q.enqueue( QueueKind::Spawn, "agent-a".to_owned(), QueueSource::Manual, "s".to_owned(), None, ); assert_ne!(a, b); assert_ne!(a, c); assert_eq!(q.snapshot().len(), 3); } #[test] fn dedup_skips_running_entries() { let q = RebuildQueue::new(); q.enqueue( QueueKind::Rebuild, "agent-a".to_owned(), QueueSource::Manual, "first".to_owned(), None, ); let running = q.take_next().expect("queued"); assert_eq!(running.state, QueueState::Running); // While the original is running, re-enqueue is legitimate. let again = q.enqueue( QueueKind::Rebuild, "agent-a".to_owned(), QueueSource::Manual, "config bumped during build".to_owned(), None, ); assert_ne!(running.id, again); let snap = q.snapshot(); assert_eq!(snap.len(), 2); } #[test] fn finish_marks_state_and_keeps_history() { let q = RebuildQueue::new(); let id = q.enqueue( QueueKind::Rebuild, "agent-a".to_owned(), QueueSource::Manual, "r".to_owned(), None, ); q.take_next(); q.finish(id, QueueState::Done, None); let snap = q.snapshot(); assert_eq!(snap.len(), 1); assert_eq!(snap[0].state, QueueState::Done); assert!(snap[0].finished_at.is_some()); assert!(snap[0].error.is_none()); } #[test] fn finish_with_failure_records_error() { let q = RebuildQueue::new(); let id = q.enqueue( QueueKind::Rebuild, "agent-a".to_owned(), QueueSource::Manual, "r".to_owned(), None, ); q.take_next(); q.finish(id, QueueState::Failed, Some("nix build failed".to_owned())); let snap = q.snapshot(); assert_eq!(snap[0].state, QueueState::Failed); assert_eq!(snap[0].error.as_deref(), Some("nix build failed")); } #[test] fn history_evicts_old_terminals_per_kind() { let q = RebuildQueue::new(); for i in 0..(MAX_HISTORY_PER_KIND + 3) { let id = q.enqueue( QueueKind::Rebuild, format!("agent-{i}"), QueueSource::Manual, "r".to_owned(), None, ); q.take_next(); q.finish(id, QueueState::Done, None); } let snap = q.snapshot(); assert_eq!(snap.len(), MAX_HISTORY_PER_KIND); } #[test] fn cancel_clears_queued_entry() { let q = RebuildQueue::new(); let id = q.enqueue( QueueKind::Rebuild, "agent-a".to_owned(), QueueSource::Manual, "r".to_owned(), None, ); assert!(q.cancel(id)); let snap = q.snapshot(); assert_eq!(snap[0].state, QueueState::Cancelled); assert!(q.take_next().is_none()); } #[test] fn cancel_refuses_running_entry() { let q = RebuildQueue::new(); let id = q.enqueue( QueueKind::Rebuild, "agent-a".to_owned(), QueueSource::Manual, "r".to_owned(), None, ); q.take_next(); assert!(!q.cancel(id)); let snap = q.snapshot(); assert_eq!(snap[0].state, QueueState::Running); } #[test] fn parent_id_groups_cascade() { let q = RebuildQueue::new(); let meta = q.enqueue( QueueKind::MetaUpdate, "hyperhive".to_owned(), QueueSource::Manual, "lock bump".to_owned(), None, ); let child = q.enqueue( QueueKind::Rebuild, "agent-a".to_owned(), QueueSource::MetaUpdate, "cascade".to_owned(), Some(meta), ); let snap = q.snapshot(); let child_entry = snap.iter().find(|e| e.id == child).expect("child queued"); assert_eq!(child_entry.parent_id, Some(meta)); } #[test] fn cancel_children_marks_queued_descendants() { let q = RebuildQueue::new(); let meta = q.enqueue( QueueKind::MetaUpdate, "hyperhive".to_owned(), QueueSource::Manual, "lock bump".to_owned(), None, ); let a = q.enqueue( QueueKind::Rebuild, "agent-a".to_owned(), QueueSource::MetaUpdate, "cascade".to_owned(), Some(meta), ); let b = q.enqueue( QueueKind::Rebuild, "agent-b".to_owned(), QueueSource::MetaUpdate, "cascade".to_owned(), Some(meta), ); // An unrelated queued entry must not be cancelled. let c = q.enqueue( QueueKind::Rebuild, "agent-c".to_owned(), QueueSource::Manual, "operator queued".to_owned(), None, ); let cancelled = q.cancel_children(meta); assert_eq!(cancelled, 2); let snap = q.snapshot(); let find = |id: u64| snap.iter().find(|e| e.id == id).expect("present"); assert_eq!(find(a).state, QueueState::Cancelled); assert_eq!(find(b).state, QueueState::Cancelled); assert_eq!(find(c).state, QueueState::Queued); } #[test] fn approval_entries_keep_approval_id() { let q = RebuildQueue::new(); let id = q.enqueue_full(FullEnqueue { kind: QueueKind::Rebuild, agent: "agent-a".to_owned(), source: QueueSource::Approval, reason: "approval 42 apply commit".to_owned(), parent_id: None, inputs: Vec::new(), approval_id: Some(42), perm_payload: None, depends_on: Vec::new(), }); let snap = q.snapshot(); let entry = snap.iter().find(|e| e.id == id).expect("entry present"); assert_eq!(entry.approval_id, Some(42)); assert_eq!(entry.source, QueueSource::Approval); } #[test] fn approval_entries_dedup_only_on_matching_id() { // Two pending approval-driven entries for the same agent but // DIFFERENT approval ids must NOT collapse — each operator // approve click is a separate piece of work even when the // (kind, agent) pair matches. let q = RebuildQueue::new(); let a = q.enqueue_full(FullEnqueue { kind: QueueKind::Rebuild, agent: "agent-a".to_owned(), source: QueueSource::Approval, reason: "approval #1".to_owned(), parent_id: None, inputs: Vec::new(), approval_id: Some(1), perm_payload: None, depends_on: Vec::new(), }); let b = q.enqueue_full(FullEnqueue { kind: QueueKind::Rebuild, agent: "agent-a".to_owned(), source: QueueSource::Approval, reason: "approval #2".to_owned(), parent_id: None, inputs: Vec::new(), approval_id: Some(2), perm_payload: None, depends_on: Vec::new(), }); assert_ne!(a, b); assert_eq!(q.snapshot().len(), 2); // Same approval_id submitted twice DOES dedup (rapid double- // click on the dashboard's approve button is a single op). let c = q.enqueue_full(FullEnqueue { kind: QueueKind::Rebuild, agent: "agent-a".to_owned(), source: QueueSource::Approval, reason: "approval #1 (duplicate)".to_owned(), parent_id: None, inputs: Vec::new(), approval_id: Some(1), perm_payload: None, depends_on: Vec::new(), }); assert_eq!(a, c); assert_eq!(q.snapshot().len(), 2); } #[test] fn cancel_children_skips_running_and_terminal() { let q = RebuildQueue::new(); let meta = q.enqueue( QueueKind::MetaUpdate, "hyperhive".to_owned(), QueueSource::Manual, "lock bump".to_owned(), None, ); // Running child — must NOT be cancelled. let running = q.enqueue( QueueKind::Rebuild, "agent-a".to_owned(), QueueSource::MetaUpdate, "cascade".to_owned(), Some(meta), ); q.take_next(); // pops meta, marks it Running q.take_next(); // pops `running`, marks it Running // Terminal child — must NOT be re-cancelled (its state stays Done). let done = q.enqueue( QueueKind::Rebuild, "agent-b".to_owned(), QueueSource::MetaUpdate, "cascade".to_owned(), Some(meta), ); q.take_next(); q.finish(done, QueueState::Done, None); // Queued child that should be cancelled. let queued = q.enqueue( QueueKind::Rebuild, "agent-c".to_owned(), QueueSource::MetaUpdate, "cascade".to_owned(), Some(meta), ); let n = q.cancel_children(meta); assert_eq!(n, 1); let snap = q.snapshot(); let find = |id: u64| snap.iter().find(|e| e.id == id).expect("present"); assert_eq!(find(running).state, QueueState::Running); assert_eq!(find(done).state, QueueState::Done); assert_eq!(find(queued).state, QueueState::Cancelled); } #[test] fn set_step_updates_running_entry_and_signals_change() { let q = RebuildQueue::new(); let id = q.enqueue( QueueKind::Rebuild, "a".to_owned(), QueueSource::Manual, "test".to_owned(), None, ); // Queued — set_step should refuse (returns false). assert!(!q.set_step(id, "plant tags")); // Promote to Running. let entry = q.take_next().expect("queued entry"); assert_eq!(entry.id, id); // First label transition — true. assert!(q.set_step(id, "plant tags")); assert_eq!( q.snapshot() .iter() .find(|e| e.id == id) .and_then(|e| e.step.as_deref()), Some("plant tags") ); // Same label again — false (caller can skip the snapshot emit). assert!(!q.set_step(id, "plant tags")); // Different label — true. assert!(q.set_step(id, "nixos-container update")); assert_eq!( q.snapshot() .iter() .find(|e| e.id == id) .and_then(|e| e.step.as_deref()), Some("nixos-container update") ); } #[test] fn set_step_no_op_on_unknown_id() { let q = RebuildQueue::new(); assert!(!q.set_step(999, "anything")); } /// A `MetaUpdate` cascade `Rebuild` (with `parent_id` = `Some(meta_id)`) must /// NOT dedup into a pre-existing `Queued` `Rebuild` with a different `parent_id` /// (e.g. from a startup sweep). Without the `parent_id` dedup guard the /// cascade rebuild would be swallowed and the agent would never rebuild /// against the post-lock-bump meta. #[test] fn meta_update_cascade_does_not_dedup_into_startup_sweep_rebuild() { let q = RebuildQueue::new(); // Startup sweep enqueues a Rebuild for alice with its own parent_id. let sweep = q.enqueue( QueueKind::StartupSweep, "hyperhive".to_owned(), QueueSource::AutoUpdate, "boot sweep".to_owned(), None, ); let sweep_rebuild = q.enqueue( QueueKind::Rebuild, "alice".to_owned(), QueueSource::StartupSweep, "startup sweep".to_owned(), Some(sweep), ); // MetaUpdate cascade pre-enqueues another Rebuild for alice. let meta = q.enqueue( QueueKind::MetaUpdate, "hyperhive".to_owned(), QueueSource::Manual, "bump nixpkgs".to_owned(), None, ); let cascade_rebuild = q.enqueue( QueueKind::Rebuild, "alice".to_owned(), QueueSource::MetaUpdate, "meta-update cascade".to_owned(), Some(meta), ); // The two Rebuilds have different parent_ids — must NOT dedup. assert_ne!( sweep_rebuild, cascade_rebuild, "cascade rebuild must be distinct from startup-sweep rebuild" ); let snap = q.snapshot(); let rebuilds: Vec<_> = snap .iter() .filter(|e| e.kind == QueueKind::Rebuild && e.agent == "alice") .collect(); assert_eq!( rebuilds.len(), 2, "both rebuilds must be present in the queue" ); } #[test] fn finish_clears_step() { let q = RebuildQueue::new(); let id = q.enqueue( QueueKind::Rebuild, "a".to_owned(), QueueSource::Manual, "test".to_owned(), None, ); q.take_next(); assert!(q.set_step(id, "running phase")); q.finish(id, QueueState::Done, None); assert_eq!( q.snapshot() .iter() .find(|e| e.id == id) .and_then(|e| e.step.as_deref()), 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(FullEnqueue { kind: QueueKind::Rebuild, agent: "agent-b".to_owned(), source: QueueSource::Manual, reason: "second (blocked on a)".to_owned(), parent_id: None, inputs: Vec::new(), approval_id: None, perm_payload: None, depends_on: 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(); q.finish(dep, QueueState::Done, None); // Push `dep` out of the per-kind history window: `trim_history` // keeps the newest MAX_HISTORY_PER_KIND terminals per kind, so it // takes that many newer terminals to evict `dep`. for i in 0..MAX_HISTORY_PER_KIND { let extra = q.enqueue( QueueKind::Rebuild, format!("extra-{i}"), QueueSource::Manual, format!("extra-{i}"), 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(FullEnqueue { kind: QueueKind::Rebuild, agent: "downstream".to_owned(), source: QueueSource::Manual, reason: "downstream (dep evicted = resolved)".to_owned(), parent_id: None, inputs: Vec::new(), approval_id: None, perm_payload: None, depends_on: 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(FullEnqueue { kind: QueueKind::Rebuild, agent: "target".to_owned(), source: QueueSource::Manual, reason: "r".to_owned(), parent_id: None, inputs: Vec::new(), approval_id: None, perm_payload: None, depends_on: vec![dep1], }); // Same kind+agent but different depends_on — must NOT dedup. let b = q.enqueue_full(FullEnqueue { kind: QueueKind::Rebuild, agent: "target".to_owned(), source: QueueSource::Manual, reason: "r".to_owned(), parent_id: None, inputs: Vec::new(), approval_id: None, perm_payload: None, depends_on: 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(FullEnqueue { kind: QueueKind::Rebuild, agent: "target".to_owned(), source: QueueSource::Manual, reason: "r again".to_owned(), parent_id: None, inputs: Vec::new(), approval_id: None, perm_payload: None, depends_on: 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(FullEnqueue { kind: QueueKind::Rebuild, agent: "b".to_owned(), source: QueueSource::Manual, reason: "b (blocked on a)".to_owned(), parent_id: None, inputs: Vec::new(), approval_id: None, perm_payload: None, depends_on: 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); } }