From 65895c6ebffe0db0fb8f1990e242cfebdf5f56a0 Mon Sep 17 00:00:00 2001 From: damocles Date: Sun, 21 Jun 2026 21:49:19 +0200 Subject: [PATCH] hive-c0re: fast lane for dashboard start/stop queue ops --- hive-c0re/src/main.rs | 9 + hive-c0re/src/rebuild_queue.rs | 443 +++++++++++++++++++++++++++------ 2 files changed, 374 insertions(+), 78 deletions(-) diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index 3cf6496c..cd9d3a20 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -435,6 +435,15 @@ async fn cmd_serve( tokio::spawn(async move { rebuild_queue::run_worker(q_coord).await; }); + // Fast lane: a second serial worker for hard Start/Stop, running + // concurrently with the build worker above so a stop/start never + // waits behind a slow build for another container. Per-agent + // ordering vs that agent's own build is enforced in the queue's + // claim logic (a fast op defers behind its agent's running build). + let fast_coord = coord.clone(); + tokio::spawn(async move { + rebuild_queue::run_fast_worker(fast_coord).await; + }); } // Forward every broker event onto the unified dashboard // channel with a freshly-stamped seq, so the dashboard SSE diff --git a/hive-c0re/src/rebuild_queue.rs b/hive-c0re/src/rebuild_queue.rs index 5e1c6930..6a816f84 100644 --- a/hive-c0re/src/rebuild_queue.rs +++ b/hive-c0re/src/rebuild_queue.rs @@ -72,6 +72,16 @@ impl QueueKind { 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. @@ -271,9 +281,15 @@ struct Inner { #[derive(Debug)] pub struct RebuildQueue { inner: Mutex, - /// Worker wakes on this signal. The worker checks the queue and - /// loops back to `notified().await` when there's nothing to run. + /// 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 { @@ -281,6 +297,7 @@ impl Default for RebuildQueue { Self { inner: Mutex::new(Inner::default()), notify: Notify::new(), + fast_notify: Notify::new(), } } } @@ -480,24 +497,55 @@ impl RebuildQueue { build_log_id: None, }; inner.entries.push_back(entry); - // Wake the worker. `notify_one` is a no-op when there's no - // waiter; the next `notified().await` returns immediately. - self.notify.notify_one(); + // 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 } - /// Pop the next `Queued` entry whose dependencies are resolved and - /// mark it `Running`. Returns the entry (a clone — the original + /// 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 { + 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 { + 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 there's nothing queued OR every - /// queued entry has unresolved dependencies. + /// 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). - pub fn take_next(&self) -> Option { + /// + /// 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 { 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). @@ -516,17 +564,21 @@ impl RebuildQueue { .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 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()); @@ -664,6 +716,28 @@ impl RebuildQueue { } } +/// 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, 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 @@ -684,52 +758,89 @@ impl RebuildQueue { /// 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, 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) { let mut shutdown = coord.shutdown_rx(); loop { - // Drain everything available now. - while let Some(entry) = coord.rebuild_queue.take_next() { - coord.emit_rebuild_queue_snapshot(); - tracing::info!( - id = entry.id, - kind = entry.kind.as_str(), - agent = %entry.agent, - source = entry.source.as_str(), - "rebuild_queue: running" - ); - let result = dispatch(&coord, &entry).await; - match result { - Ok(()) => { - coord.rebuild_queue.finish(entry.id, QueueState::Done, None); - tracing::info!(id = entry.id, "rebuild_queue: done"); - } - Err(e) => { - let msg = format!("{e:#}"); - let truncated = if msg.len() > 2_000 { - format!("{}…", &msg[..2_000]) - } else { - msg.clone() - }; - coord - .rebuild_queue - .finish(entry.id, QueueState::Failed, Some(truncated)); - tracing::warn!(id = entry.id, error = %msg, "rebuild_queue: failed"); - } - } - coord.emit_rebuild_queue_snapshot(); + 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(); } - // Park until something new is enqueued OR shutdown fires. tokio::select! { biased; res = shutdown.changed() => { if res.is_err() || *shutdown.borrow() { - tracing::info!("rebuild_queue: worker exiting on shutdown"); + tracing::info!("rebuild_queue: build worker exiting on shutdown"); return; } } - () = coord.rebuild_queue.notify.notified() => { - // New entry — back to the drain loop. + () = 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) { + 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() => {} } } } @@ -1052,12 +1163,12 @@ mod tests { None, ); assert_ne!(a, b); - let next = q.take_next().expect("queued"); + let next = q.take_next_build().expect("queued"); assert_eq!(next.id, a); assert_eq!(next.state, QueueState::Running); - let next = q.take_next().expect("queued"); + let next = q.take_next_build().expect("queued"); assert_eq!(next.id, b); - assert!(q.take_next().is_none()); + assert!(q.take_next_build().is_none()); } #[test] @@ -1181,7 +1292,7 @@ mod tests { "first".to_owned(), None, ); - let running = q.take_next().expect("queued"); + 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( @@ -1206,7 +1317,7 @@ mod tests { "r".to_owned(), None, ); - q.take_next(); + q.take_next_build(); q.finish(id, QueueState::Done, None); let snap = q.snapshot(); assert_eq!(snap.len(), 1); @@ -1225,7 +1336,7 @@ mod tests { "r".to_owned(), None, ); - q.take_next(); + 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); @@ -1243,7 +1354,7 @@ mod tests { "r".to_owned(), None, ); - q.take_next(); + q.take_next_build(); q.finish(id, QueueState::Done, None); } let snap = q.snapshot(); @@ -1263,7 +1374,7 @@ mod tests { assert!(q.cancel(id)); let snap = q.snapshot(); assert_eq!(snap[0].state, QueueState::Cancelled); - assert!(q.take_next().is_none()); + assert!(q.take_next_build().is_none()); } #[test] @@ -1276,7 +1387,7 @@ mod tests { "r".to_owned(), None, ); - q.take_next(); + q.take_next_build(); assert!(!q.cancel(id)); let snap = q.snapshot(); assert_eq!(snap[0].state, QueueState::Running); @@ -1431,8 +1542,8 @@ mod tests { "cascade".to_owned(), Some(meta), ); - q.take_next(); // pops meta, marks it Running - q.take_next(); // pops `running`, marks it Running + 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, @@ -1441,7 +1552,7 @@ mod tests { "cascade".to_owned(), Some(meta), ); - q.take_next(); + q.take_next_build(); q.finish(done, QueueState::Done, None); // Queued child that should be cancelled. let queued = q.enqueue( @@ -1473,7 +1584,7 @@ mod tests { // Queued — set_step should refuse (returns false). assert!(!q.set_step(id, "plant tags")); // Promote to Running. - let entry = q.take_next().expect("queued entry"); + 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")); @@ -1568,7 +1679,7 @@ mod tests { "test".to_owned(), None, ); - q.take_next(); + q.take_next_build(); assert!(q.set_step(id, "running phase")); q.finish(id, QueueState::Done, None); assert_eq!( @@ -1606,13 +1717,16 @@ mod tests { depends_on: vec![a], }); // B depends on A — take_next should give A first. - let first = q.take_next().expect("a is ready"); + 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().is_none(), "b must be blocked while a runs"); + 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().expect("b unblocked after a done"); + let second = q.take_next_build().expect("b unblocked after a done"); assert_eq!(second.id, b); } @@ -1630,7 +1744,7 @@ mod tests { "filler".to_owned(), None, ); - q.take_next(); + q.take_next_build(); q.finish(id, QueueState::Done, None); } // `dep` gets enqueued, run, finished, and evicted by the @@ -1642,7 +1756,7 @@ mod tests { "dep".to_owned(), None, ); - q.take_next(); + 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 @@ -1655,7 +1769,7 @@ mod tests { format!("extra-{i}"), None, ); - q.take_next(); + q.take_next_build(); q.finish(extra, QueueState::Done, None); } // `dep` should now be evicted. @@ -1675,7 +1789,9 @@ mod tests { perm_payload: None, depends_on: vec![dep], }); - let got = q.take_next().expect("downstream runnable when dep evicted"); + let got = q + .take_next_build() + .expect("downstream runnable when dep evicted"); assert_eq!(got.id, downstream); } @@ -1761,9 +1877,180 @@ mod tests { perm_payload: None, depends_on: vec![a], }); - q.take_next(); // pop a, mark Running + q.take_next_build(); // 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"); + 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" + ); + } }