hive-c0re: fast lane for dashboard start/stop queue ops

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

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<Inner>,
-    /// 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<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 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<QueueEntry> {
+    ///
+    /// 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).
@@ -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<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
@@ -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<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 {
-        // 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<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() => {}
         }
     }
 }
@@ -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"
+        );
+    }
 }