diff --git a/docs/coordinator.md b/docs/coordinator.md index 09126c86..6911a49e 100644 --- a/docs/coordinator.md +++ b/docs/coordinator.md @@ -74,6 +74,28 @@ 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 can see which phase is taking time. +### Dependency tracking + +Each entry carries a `depends_on: Vec` 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. + --- ## Container view diff --git a/hive-c0re/src/actions.rs b/hive-c0re/src/actions.rs index 2f70c4c8..5e30bd0d 100644 --- a/hive-c0re/src/actions.rs +++ b/hive-c0re/src/actions.rs @@ -55,6 +55,7 @@ pub async fn approve(coord: Arc, id: i64) -> Result<()> { Vec::new(), Some(id), None, + Vec::new(), ); coord.emit_rebuild_queue_snapshot(); Ok(()) @@ -74,6 +75,7 @@ pub async fn approve(coord: Arc, id: i64) -> Result<()> { inputs.clone(), Some(id), None, + Vec::new(), ); // Pre-enqueue cascade rebuilds in topological order so // agents depending on updated inputs are rebuilt after the @@ -102,6 +104,7 @@ pub async fn approve(coord: Arc, id: i64) -> Result<()> { Vec::new(), Some(id), None, + Vec::new(), ); coord.emit_rebuild_queue_snapshot(); Ok(()) diff --git a/hive-c0re/src/rebuild_queue.rs b/hive-c0re/src/rebuild_queue.rs index a5238454..df225050 100644 --- a/hive-c0re/src/rebuild_queue.rs +++ b/hive-c0re/src/rebuild_queue.rs @@ -206,6 +206,14 @@ pub struct QueueEntry { /// the wire in those cases. #[serde(default, skip_serializing_if = "Option::is_none")] pub perm_payload: Option, + /// 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, } /// How many terminal-state entries (`Done` / `Failed` / `Cancelled`) @@ -286,6 +294,7 @@ impl RebuildQueue { Vec::new(), None, None, + Vec::new(), ) } @@ -303,7 +312,17 @@ impl RebuildQueue { parent_id: Option, inputs: Vec, ) -> 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 @@ -325,6 +344,7 @@ impl RebuildQueue { Vec::new(), None, Some(payload), + Vec::new(), ) } @@ -333,10 +353,10 @@ impl RebuildQueue { /// `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. - // 9 args: the queue entry has 6 independent submit-time fields plus - // three kind-specific payload fields (inputs, approval_id, perm_payload). - // A builder struct would obscure the call sites; the shorter wrappers - // already cover all common cases. + // 10 args: the queue entry has 6 independent submit-time fields plus + // four kind-specific payload fields (inputs, approval_id, perm_payload, + // depends_on). A builder struct would obscure the call sites; the + // shorter wrappers already cover all common cases. #[allow(clippy::too_many_arguments)] pub fn enqueue_full( &self, @@ -348,6 +368,7 @@ impl RebuildQueue { inputs: Vec, approval_id: Option, perm_payload: Option, + depends_on: Vec, ) -> u64 { let mut inner = self.inner.lock().expect("rebuild_queue mutex poisoned"); // Dedup against a pending entry with the same (kind, agent) — @@ -380,6 +401,7 @@ impl RebuildQueue { && 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 _; @@ -406,6 +428,7 @@ impl RebuildQueue { approval_id, step: None, perm_payload, + depends_on, }; inner.entries.push_back(entry); // Wake the worker. `notify_one` is a no-op when there's no @@ -414,16 +437,42 @@ impl RebuildQueue { id } - /// Pop the next `Queued` entry 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. + /// Pop the next `Queued` entry whose dependencies are resolved 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. + /// + /// 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 { 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 = inner .entries .iter() - .position(|e| e.state == QueueState::Queued)?; + .filter(|e| e.state.is_terminal()) + .map(|e| e.id) + .collect(); + // All queued ids — used to distinguish "not yet terminal" from + // "evicted (= resolved)". + let queued_ids: std::collections::HashSet = 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. + terminal_ids.contains(dep_id) || !queued_ids.contains(dep_id) + }) + })?; let entry = &mut inner.entries[pos]; entry.state = QueueState::Running; entry.started_at = Some(now_unix()); @@ -1131,6 +1180,8 @@ mod tests { None, Vec::new(), Some(42), + None, + Vec::new(), ); let snap = q.snapshot(); let entry = snap.iter().find(|e| e.id == id).expect("entry present"); @@ -1153,6 +1204,8 @@ mod tests { None, Vec::new(), Some(1), + None, + Vec::new(), ); let b = q.enqueue_full( QueueKind::Rebuild, @@ -1162,6 +1215,8 @@ mod tests { None, Vec::new(), Some(2), + None, + Vec::new(), ); assert_ne!(a, b); assert_eq!(q.snapshot().len(), 2); @@ -1175,6 +1230,8 @@ mod tests { None, Vec::new(), Some(1), + None, + Vec::new(), ); assert_eq!(a, c); assert_eq!(q.snapshot().len(), 2); @@ -1346,4 +1403,187 @@ mod tests { 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); + } }