vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(#550): add depends_on to queue entries for explicit dep tracking

Browse source
This commit is contained in:
damocles 2026-06-04 17:22:44 +02:00 • committed by mara
commit d31a723daf
3 changed files with 276 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

22
docs/coordinator.md
View file

@ -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<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.
---
## Container view

3
hive-c0re/src/actions.rs
View file

@ -55,6 +55,7 @@ pub async fn approve(coord: Arc<Coordinator>, 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<Coordinator>, 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<Coordinator>, id: i64) -> Result<()> {
Vec::new(),
Some(id),
None,
Vec::new(),
);
coord.emit_rebuild_queue_snapshot();
Ok(())

262
hive-c0re/src/rebuild_queue.rs
View file

@ -206,6 +206,14 @@ pub struct QueueEntry {
/// the wire in those cases.
#[serde(default, skip_serializing_if = "Option::is_none")]
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>,
}
/// 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<u64>,
inputs: Vec<String>,
) -> 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<String>,
approval_id: Option<i64>,
perm_payload: Option<PermPayload>,
depends_on: Vec<u64>,
) -> 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<QueueEntry> {
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
.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<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.
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);
}
}