2090 lines
79 KiB
Rust
2090 lines
79 KiB
Rust
//! 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",
|
|
}
|
|
}
|
|
|
|
/// Fast-lane kinds: hard `Start` / `Stop`. These run on a separate
|
|
/// serial fast worker concurrently with the build lane (so a stop/start
|
|
/// never waits behind another container's slow build). `GracefulStop`
|
|
/// and `Restart` are deliberately NOT fast — they go through the build
|
|
/// lane (`GracefulStop` holds the worker while the harness drains;
|
|
/// `Restart` is a stop+start).
|
|
pub fn is_fast(self) -> bool {
|
|
matches!(self, QueueKind::Start | QueueKind::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<String> },
|
|
/// Set the capabilities for one agent (`capabilities.json`).
|
|
Capabilities { caps: Vec<String> },
|
|
/// 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<Vec<String>>,
|
|
caps: Option<Vec<String>>,
|
|
},
|
|
}
|
|
|
|
/// 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<u64>,
|
|
/// 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<i64>,
|
|
#[serde(default, skip_serializing_if = "Option::is_none")]
|
|
pub finished_at: Option<i64>,
|
|
/// 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<String>,
|
|
/// `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<String>,
|
|
/// 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<i64>,
|
|
/// 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<String>,
|
|
/// `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<PermPayload>,
|
|
/// 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<u64>,
|
|
/// 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<i64>,
|
|
}
|
|
|
|
/// 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<QueueEntry>,
|
|
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<Inner>,
|
|
/// Build-lane worker wakes on this signal. The worker checks the queue
|
|
/// and loops back to `notified().await` when there's nothing to run.
|
|
/// Also nudged by the fast worker when a fast op finishes (a build may
|
|
/// have been deferred behind a `Running` fast op for the same agent).
|
|
pub(crate) notify: Notify,
|
|
/// Fast-lane worker wakes on this signal. Nudged on a fast `Start` /
|
|
/// `Stop` enqueue and by the build worker when a build finishes (a
|
|
/// deferred `Start` may now be runnable).
|
|
pub(crate) fast_notify: Notify,
|
|
}
|
|
|
|
impl Default for RebuildQueue {
|
|
fn default() -> Self {
|
|
Self {
|
|
inner: Mutex::new(Inner::default()),
|
|
notify: Notify::new(),
|
|
fast_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<u64>,
|
|
pub inputs: Vec<String>,
|
|
pub approval_id: Option<i64>,
|
|
pub perm_payload: Option<PermPayload>,
|
|
pub depends_on: Vec<u64>,
|
|
}
|
|
|
|
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>,
|
|
) -> 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<u64>,
|
|
inputs: Vec<String>,
|
|
) -> 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 for this entry's lane (fast = Start/Stop, build =
|
|
// everything else). `notify_one` is a no-op when there's no waiter;
|
|
// the next `notified().await` returns immediately.
|
|
if kind.is_fast() {
|
|
self.fast_notify.notify_one();
|
|
} else {
|
|
self.notify.notify_one();
|
|
}
|
|
id
|
|
}
|
|
|
|
/// Claim the next runnable `Queued` entry for the **build** lane (every
|
|
/// kind except the fast `Start` / `Stop`) and mark it `Running`. See
|
|
/// [`Self::claim`] for the dependency + per-agent rules.
|
|
pub fn take_next_build(&self) -> Option<QueueEntry> {
|
|
self.claim(false)
|
|
}
|
|
|
|
/// Claim the next runnable `Queued` entry for the **fast** lane (hard
|
|
/// `Start` / `Stop`) and mark it `Running`. Runs on its own serial
|
|
/// worker concurrently with the build lane. See [`Self::claim`].
|
|
pub fn take_next_fast(&self) -> Option<QueueEntry> {
|
|
self.claim(true)
|
|
}
|
|
|
|
/// Pop the next `Queued` entry for one lane whose dependencies are
|
|
/// resolved and which doesn't race the same agent's other-lane work,
|
|
/// 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 nothing in this lane is runnable.
|
|
///
|
|
/// `want_fast` selects the lane: `true` = fast (`Start` / `Stop`),
|
|
/// `false` = build (everything else). The two lanes run on separate
|
|
/// serial workers, so this is called from both — the lane filter keeps
|
|
/// each worker to its own kinds.
|
|
///
|
|
/// 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).
|
|
///
|
|
/// Per-agent cross-lane guard (so a fast op never races that agent's
|
|
/// own build, and vice versa):
|
|
/// - a fast op waits while its agent has a build entry `Running`;
|
|
/// a `Start` additionally waits while its agent has a build entry
|
|
/// `Queued` (a start of a soon-to-be-rebuilt container is pointless);
|
|
/// - a build op waits while its agent has a fast op `Running`.
|
|
fn claim(&self, want_fast: bool) -> Option<QueueEntry> {
|
|
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<u64> = 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<u64> = inner
|
|
.entries
|
|
.iter()
|
|
.filter(|e| !e.state.is_terminal())
|
|
.map(|e| e.id)
|
|
.collect();
|
|
let pos = {
|
|
let entries = &inner.entries;
|
|
entries.iter().position(|e| {
|
|
e.state == QueueState::Queued
|
|
&& e.kind.is_fast() == want_fast
|
|
&& 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. Callers must ensure acyclic dep graphs.
|
|
terminal_ids.contains(dep_id) || !active_ids.contains(dep_id)
|
|
})
|
|
&& lane_clear(entries, e)
|
|
})
|
|
}?;
|
|
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<String>) {
|
|
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<String>) -> 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<QueueEntry> {
|
|
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<QueueKind, usize> =
|
|
std::collections::HashMap::new();
|
|
// Walk newest-first; keep the first MAX_HISTORY_PER_KIND
|
|
// terminals per kind, evict the rest.
|
|
let entries: Vec<QueueEntry> = 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();
|
|
}
|
|
}
|
|
|
|
/// Per-agent cross-lane guard for [`RebuildQueue::claim`]: returns true when
|
|
/// entry `e` is safe to start given the same agent's other-lane work in
|
|
/// `entries`. A fast op waits for the agent's `Running` build (and a `Start`
|
|
/// also for a `Queued` build); a build op waits for the agent's `Running`
|
|
/// fast op. Keeps a stop/start from racing that container's own rebuild.
|
|
fn lane_clear(entries: &VecDeque<QueueEntry>, e: &QueueEntry) -> bool {
|
|
let agent = e.agent.as_str();
|
|
if e.kind.is_fast() {
|
|
let build_blocking = entries.iter().any(|b| {
|
|
!b.kind.is_fast()
|
|
&& b.agent == agent
|
|
&& (b.state == QueueState::Running
|
|
|| (e.kind == QueueKind::Start && b.state == QueueState::Queued))
|
|
});
|
|
!build_blocking
|
|
} else {
|
|
!entries
|
|
.iter()
|
|
.any(|f| f.kind.is_fast() && f.agent == agent && f.state == QueueState::Running)
|
|
}
|
|
}
|
|
|
|
/// 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);
|
|
|
|
/// Run one claimed queue entry to completion: snapshot, dispatch, mark
|
|
/// terminal, snapshot. Shared by both lane workers.
|
|
async fn run_one(coord: &std::sync::Arc<crate::coordinator::Coordinator>, entry: &QueueEntry) {
|
|
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"
|
|
);
|
|
match dispatch(coord, entry).await {
|
|
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();
|
|
}
|
|
|
|
/// Build-lane worker: drains every non-fast kind serially. Spawned once at
|
|
/// hive-c0re startup from `main.rs`, alongside [`run_fast_worker`] which
|
|
/// drains the fast `Start` / `Stop` lane concurrently.
|
|
///
|
|
/// Shutdown semantics: subscribes to `coord.shutdown_rx()`. On a true
|
|
/// signal the worker exits after its current entry finishes; pending
|
|
/// `Queued` entries are dropped (replayed by the startup sweep on next boot
|
|
/// or left for an operator to re-queue).
|
|
pub async fn run_worker(coord: std::sync::Arc<crate::coordinator::Coordinator>) {
|
|
let mut shutdown = coord.shutdown_rx();
|
|
loop {
|
|
while let Some(entry) = coord.rebuild_queue.take_next_build() {
|
|
run_one(&coord, &entry).await;
|
|
// A finished build may unblock a fast op that was deferred behind
|
|
// this agent's build — nudge the fast lane to re-check.
|
|
coord.rebuild_queue.fast_notify.notify_one();
|
|
}
|
|
tokio::select! {
|
|
biased;
|
|
res = shutdown.changed() => {
|
|
if res.is_err() || *shutdown.borrow() {
|
|
tracing::info!("rebuild_queue: build worker exiting on shutdown");
|
|
return;
|
|
}
|
|
}
|
|
() = coord.rebuild_queue.notify.notified() => {}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Fast-lane worker: drains hard `Start` / `Stop` serially, concurrently
|
|
/// with [`run_worker`], so a stop/start never waits behind another
|
|
/// container's slow build. Per-agent ordering vs that agent's own build is
|
|
/// enforced in [`RebuildQueue::claim`].
|
|
pub async fn run_fast_worker(coord: std::sync::Arc<crate::coordinator::Coordinator>) {
|
|
let mut shutdown = coord.shutdown_rx();
|
|
loop {
|
|
while let Some(entry) = coord.rebuild_queue.take_next_fast() {
|
|
run_one(&coord, &entry).await;
|
|
// A finished fast op may unblock a build deferred behind it.
|
|
coord.rebuild_queue.notify.notify_one();
|
|
}
|
|
tokio::select! {
|
|
biased;
|
|
res = shutdown.changed() => {
|
|
if res.is_err() || *shutdown.borrow() {
|
|
tracing::info!("rebuild_queue: fast worker exiting on shutdown");
|
|
return;
|
|
}
|
|
}
|
|
() = coord.rebuild_queue.fast_notify.notified() => {}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// 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.
|
|
/// Pick the right approval pipeline for a `Rebuild` queue entry. Both
|
|
/// `ApplyCommit` and `MergeConfigPr` approvals enqueue a `Rebuild` entry
|
|
/// (both end in a container rebuild); branch on the approval kind. Falls
|
|
/// back to the apply-commit path if the row can't be read — it re-fetches
|
|
/// + surfaces a clean error itself.
|
|
async fn dispatch_rebuild_approval(
|
|
coord: &std::sync::Arc<crate::coordinator::Coordinator>,
|
|
entry: &QueueEntry,
|
|
approval_id: i64,
|
|
) -> anyhow::Result<()> {
|
|
let kind = coord
|
|
.approvals
|
|
.get(approval_id)
|
|
.ok()
|
|
.flatten()
|
|
.map(|a| a.kind);
|
|
if kind == Some(hive_sh4re::ApprovalKind::MergeConfigPr) {
|
|
crate::actions::run_approval_merge_config_pr(coord, Some(entry.id), approval_id).await
|
|
} else {
|
|
crate::actions::run_approval_apply_commit(coord, Some(entry.id), approval_id).await
|
|
}
|
|
}
|
|
|
|
async fn dispatch(
|
|
coord: &std::sync::Arc<crate::coordinator::Coordinator>,
|
|
entry: &QueueEntry,
|
|
) -> anyhow::Result<()> {
|
|
match (entry.kind, entry.approval_id) {
|
|
(QueueKind::Rebuild, Some(approval_id)) => {
|
|
dispatch_rebuild_approval(coord, entry, approval_id).await
|
|
}
|
|
(QueueKind::Rebuild, None) => {
|
|
let current_rev =
|
|
crate::auto_update::current_flake_rev(&coord.hyperhive_flake).unwrap_or_default();
|
|
// A meta-update cascade has just set the meta lock; re-locking
|
|
// in the per-agent rebuild would revert it (the agent's own
|
|
// flake.lock wins). Every other source wants the relock so it
|
|
// advances to applied/<n>/main.
|
|
let relock = entry.source != QueueSource::MetaUpdate;
|
|
crate::auto_update::rebuild_agent(
|
|
coord,
|
|
&entry.agent,
|
|
¤t_rev,
|
|
Some(entry.id),
|
|
relock,
|
|
)
|
|
.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), true).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<crate::coordinator::Coordinator>,
|
|
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<crate::coordinator::Coordinator>,
|
|
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<crate::coordinator::Coordinator>,
|
|
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<crate::coordinator::Coordinator>,
|
|
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 <inputs>` 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-<name>` inputs.
|
|
pub async fn meta_update_cascade_agents(inputs: &[String]) -> Vec<String> {
|
|
let touched_hyperhive = inputs
|
|
.iter()
|
|
.any(|i| i == "hyperhive" || i.starts_with("hyperhive/"));
|
|
let touched_agents: Vec<String> = 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_build().expect("queued");
|
|
assert_eq!(next.id, a);
|
|
assert_eq!(next.state, QueueState::Running);
|
|
let next = q.take_next_build().expect("queued");
|
|
assert_eq!(next.id, b);
|
|
assert!(q.take_next_build().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_build().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_build();
|
|
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_build();
|
|
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_build();
|
|
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_build().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_build();
|
|
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_build(); // pops meta, marks it Running
|
|
q.take_next_build(); // 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_build();
|
|
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_build().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_build();
|
|
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_build().expect("a is ready");
|
|
assert_eq!(first.id, a);
|
|
// A is Running, not terminal — B must still be blocked.
|
|
assert!(
|
|
q.take_next_build().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_build().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_build();
|
|
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_build();
|
|
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_build();
|
|
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_build()
|
|
.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_build(); // pop a, mark Running
|
|
q.finish(a, QueueState::Failed, Some("nix build exploded".to_owned()));
|
|
let got = q.take_next_build().expect("b runnable after a failed");
|
|
assert_eq!(got.id, b);
|
|
}
|
|
|
|
// ---- fast lane (Start / Stop run on a separate concurrent worker) ----
|
|
|
|
#[test]
|
|
fn lanes_claim_only_their_own_kinds() {
|
|
let q = RebuildQueue::new();
|
|
let r = q.enqueue(
|
|
QueueKind::Rebuild,
|
|
"a".to_owned(),
|
|
QueueSource::Manual,
|
|
String::new(),
|
|
None,
|
|
);
|
|
let s = q.enqueue(
|
|
QueueKind::Stop,
|
|
"b".to_owned(),
|
|
QueueSource::Manual,
|
|
String::new(),
|
|
None,
|
|
);
|
|
let build = q.take_next_build().expect("build entry");
|
|
assert_eq!(build.id, r);
|
|
let fast = q.take_next_fast().expect("fast entry");
|
|
assert_eq!(fast.id, s);
|
|
assert!(q.take_next_build().is_none());
|
|
assert!(q.take_next_fast().is_none());
|
|
}
|
|
|
|
#[test]
|
|
fn start_defers_behind_same_agent_queued_build() {
|
|
let q = RebuildQueue::new();
|
|
let b = q.enqueue(
|
|
QueueKind::Rebuild,
|
|
"a".to_owned(),
|
|
QueueSource::Manual,
|
|
String::new(),
|
|
None,
|
|
);
|
|
q.enqueue(
|
|
QueueKind::Start,
|
|
"a".to_owned(),
|
|
QueueSource::Manual,
|
|
String::new(),
|
|
None,
|
|
);
|
|
assert!(
|
|
q.take_next_fast().is_none(),
|
|
"start blocked while same agent has a queued build"
|
|
);
|
|
q.take_next_build().expect("build runs");
|
|
q.finish(b, QueueState::Done, None);
|
|
let started = q
|
|
.take_next_fast()
|
|
.expect("start unblocked after build done");
|
|
assert_eq!(started.kind, QueueKind::Start);
|
|
}
|
|
|
|
#[test]
|
|
fn start_for_other_agent_runs_concurrently_with_a_build() {
|
|
let q = RebuildQueue::new();
|
|
q.enqueue(
|
|
QueueKind::Rebuild,
|
|
"a".to_owned(),
|
|
QueueSource::Manual,
|
|
String::new(),
|
|
None,
|
|
);
|
|
q.enqueue(
|
|
QueueKind::Start,
|
|
"a".to_owned(),
|
|
QueueSource::Manual,
|
|
String::new(),
|
|
None,
|
|
);
|
|
q.enqueue(
|
|
QueueKind::Start,
|
|
"b".to_owned(),
|
|
QueueSource::Manual,
|
|
String::new(),
|
|
None,
|
|
);
|
|
q.take_next_build().expect("a's build running");
|
|
let got = q
|
|
.take_next_fast()
|
|
.expect("start for b runs while a's build runs");
|
|
assert_eq!(got.agent, "b");
|
|
assert!(
|
|
q.take_next_fast().is_none(),
|
|
"start for a still blocked by a's running build"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn stop_jumps_queued_build_but_waits_running_build() {
|
|
// Stop jumps ahead of a *queued* build for the same agent.
|
|
let q = RebuildQueue::new();
|
|
q.enqueue(
|
|
QueueKind::Rebuild,
|
|
"a".to_owned(),
|
|
QueueSource::Manual,
|
|
String::new(),
|
|
None,
|
|
);
|
|
q.enqueue(
|
|
QueueKind::Stop,
|
|
"a".to_owned(),
|
|
QueueSource::Manual,
|
|
String::new(),
|
|
None,
|
|
);
|
|
let got = q
|
|
.take_next_fast()
|
|
.expect("stop jumps ahead of a's queued build");
|
|
assert_eq!(got.kind, QueueKind::Stop);
|
|
|
|
// But a stop waits for a *running* build of the same agent.
|
|
let q2 = RebuildQueue::new();
|
|
let b = q2.enqueue(
|
|
QueueKind::Rebuild,
|
|
"a".to_owned(),
|
|
QueueSource::Manual,
|
|
String::new(),
|
|
None,
|
|
);
|
|
q2.enqueue(
|
|
QueueKind::Stop,
|
|
"a".to_owned(),
|
|
QueueSource::Manual,
|
|
String::new(),
|
|
None,
|
|
);
|
|
q2.take_next_build().expect("a's build running");
|
|
assert!(
|
|
q2.take_next_fast().is_none(),
|
|
"stop waits for a's running build (no kill mid-rebuild)"
|
|
);
|
|
q2.finish(b, QueueState::Done, None);
|
|
assert!(
|
|
q2.take_next_fast().is_some(),
|
|
"stop runs once a's build is done"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn build_defers_behind_same_agent_running_fast_op() {
|
|
let q = RebuildQueue::new();
|
|
q.enqueue(
|
|
QueueKind::Stop,
|
|
"a".to_owned(),
|
|
QueueSource::Manual,
|
|
String::new(),
|
|
None,
|
|
);
|
|
q.enqueue(
|
|
QueueKind::Rebuild,
|
|
"a".to_owned(),
|
|
QueueSource::Manual,
|
|
String::new(),
|
|
None,
|
|
);
|
|
let s = q.take_next_fast().expect("stop running");
|
|
assert!(
|
|
q.take_next_build().is_none(),
|
|
"build waits while a's fast op is running"
|
|
);
|
|
q.finish(s.id, QueueState::Done, None);
|
|
assert!(
|
|
q.take_next_build().is_some(),
|
|
"build runs once the fast op is done"
|
|
);
|
|
}
|
|
}
|