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