feat(queue): link rebuild queue entries to build log rows for live streaming

This commit is contained in:
damocles 2026-06-04 17:33:59 +02:00 committed by mara
commit 7118c5efdd
6 changed files with 101 additions and 9 deletions

View file

@ -2266,6 +2266,20 @@ window.marked = marked;
if (entry.step) {
li.append(el('div', { class: 'rqe-step' }, '↳ ' + entry.step));
}
// Live-log link: when a build_log_id is present the update/create op
// opened a build_logs row; the SSE stream endpoint lets the operator
// follow output in real time without polling.
if (entry.build_log_id != null) {
li.append(
' ',
el('a', {
class: 'rqe-log-link',
href: '/api/build-logs/id/' + entry.build_log_id,
target: '_blank',
title: 'view build log #' + entry.build_log_id,
}, 'logs →'),
);
}
// Error block, when failed.
if (entry.error) {
li.append(el('pre', { class: 'rqe-error', title: entry.error }, truncate(entry.error, 200)));

View file

@ -550,9 +550,19 @@ async fn run_apply_commit(
// "nixos-container update" label for the whole multi-minute window.
let hive = coord.hive_env();
let paths = Coordinator::agent_paths(&approval.agent, agent_dir.to_path_buf());
let build_result = lifecycle::rebuild_no_meta(&approval.agent, &hive, &paths, &|step| {
coord.set_queue_step(queue_entry_id, step)
})
let build_result = lifecycle::rebuild_no_meta(
&approval.agent,
&hive,
&paths,
&|step| coord.set_queue_step(queue_entry_id, step),
&|log_id| {
if let Some(qid) = queue_entry_id {
if coord.rebuild_queue.set_build_log_id(qid, log_id) {
coord.emit_rebuild_queue_snapshot();
}
}
},
)
.await;
match build_result {

View file

@ -84,9 +84,19 @@ pub async fn rebuild_agent(
// lifecycle_action; this catches the auto-update scan + any
// other direct caller.
let guard = coord.transient_guard(name, crate::coordinator::TransientKind::Rebuilding);
let result = lifecycle::rebuild(name, &hive, &paths, &|step| {
coord.set_queue_step(queue_entry_id, step)
})
let result = lifecycle::rebuild(
name,
&hive,
&paths,
&|step| coord.set_queue_step(queue_entry_id, step),
&|log_id| {
if let Some(qid) = queue_entry_id {
if coord.rebuild_queue.set_build_log_id(qid, log_id) {
coord.emit_rebuild_queue_snapshot();
}
}
},
)
.await;
drop(guard);
match &result {

View file

@ -377,6 +377,7 @@ pub async fn rebuild(
hive: &HiveEnv,
paths: &AgentPaths,
on_step: &(dyn Fn(&str) + Send + Sync),
on_build_log_id: &(dyn Fn(i64) + Send + Sync),
) -> Result<()> {
// Sync the meta flake (idempotent — no-op when the rendered
// flake matches disk) so a manual rebuild from the dashboard
@ -389,7 +390,7 @@ pub async fn rebuild(
// `applied/<n>/main` currently points at (deployed/<latest>).
// Commits the lock if it changed.
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
@ -402,11 +403,17 @@ pub async fn rebuild(
/// label so callers can surface progress (e.g. update the rebuild-queue
/// step shown in the dashboard). Pass `&|_| ()` when progress reporting
/// is not needed.
///
/// `on_build_log_id` is called with the build-log row id immediately after
/// the `nixos-container update` log row opens, before the actual update
/// command starts. Callers can use this to link the queue entry to the log
/// for live streaming. Pass `&|_| ()` when not needed.
pub async fn rebuild_no_meta(
name: &str,
hive: &HiveEnv,
paths: &AgentPaths,
on_step: &(dyn Fn(&str) + Send + Sync),
on_build_log_id: &(dyn Fn(i64) + Send + Sync),
) -> Result<()> {
validate(name)?;
if let Some(other) = port_collision(name).await {
@ -440,7 +447,7 @@ pub async fn rebuild_no_meta(
priv_run("stop", name).await?;
}
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 {
// 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
@ -1322,6 +1329,24 @@ fn make_log_callback(
/// is appended to the build-log row as it arrives, so the dashboard
/// shows live progress during long `nixos-container create` / `update` runs.
async fn priv_run(kind: &str, name: &str) -> Result<()> {
priv_run_inner(kind, name, None).await
}
/// Like `priv_run` but calls `on_log_id(log_id)` immediately after the
/// build-log row is opened — before the actual container op starts.
/// This lets callers surface the row id for live streaming (e.g. the
/// rebuild-queue worker sets `build_log_id` on the queue entry so the
/// dashboard can link to `/api/build-logs/id/{id}/stream`).
///
/// The callback fires only when a build-log row is successfully opened
/// (i.e. the global `BuildLogs` handle is installed AND `h.start()`
/// succeeds). No-op when `on_log_id` is `None` — that's the path for
/// all callers that don't need the id.
async fn priv_run_inner(
kind: &str,
name: &str,
on_log_id: Option<&(dyn Fn(i64) + Send + Sync)>,
) -> Result<()> {
let container = container_name(name);
let cmdline = format!("nixos-container {kind} {container}");
@ -1333,6 +1358,11 @@ async fn priv_run(kind: &str, name: &str) -> Result<()> {
})
.ok()
});
// Notify the caller as soon as the log row exists so it can surface
// the id for live streaming before the container op even starts.
if let (Some(id), Some(cb)) = (log_id, on_log_id) {
cb(id);
}
// For long-running ops use the streaming protocol so build_logs
// receives lines in real time rather than as a batch at completion.

View file

@ -214,6 +214,15 @@ pub struct QueueEntry {
/// 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`)
@ -429,6 +438,7 @@ impl RebuildQueue {
step: None,
perm_payload,
depends_on,
build_log_id: None,
};
inner.entries.push_back(entry);
// Wake the worker. `notify_one` is a no-op when there's no
@ -521,6 +531,24 @@ impl RebuildQueue {
true
}
/// Link a `build_logs` row to a `Running` entry. Called by the
/// lifecycle worker when `nixos-container update` opens a build log
/// row so the dashboard can surface a "view logs" link while the
/// build is in flight. Returns `true` when the row was found and
/// the id was stored; `false` when the entry is no longer in the
/// queue or is not `Running`.
pub fn set_build_log_id(&self, id: u64, log_id: i64) -> bool {
let mut inner = self.inner.lock().expect("rebuild_queue mutex poisoned");
let Some(entry) = inner.entries.iter_mut().find(|e| e.id == id) else {
return false;
};
if entry.state != QueueState::Running {
return false;
}
entry.build_log_id = Some(log_id);
true
}
/// Snapshot the queue for `/api/state` and `RebuildQueueChanged`.
/// Cheap clone — entries are small (~hundreds of bytes each).
pub fn snapshot(&self) -> Vec<QueueEntry> {

View file

@ -161,7 +161,7 @@ async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
let agent_dir = coord.ensure_runtime(name)?;
let hive = coord.hive_env();
let paths = Coordinator::agent_paths(name, agent_dir);
let result = lifecycle::rebuild(name, &hive, &paths, &|_| ()).await;
let result = lifecycle::rebuild(name, &hive, &paths, &|_| (), &|_| ()).await;
// Mirror auto_update::rebuild_agent — the manager wants
// to know about every rebuild attempt regardless of
// which surface triggered it, especially failures