nordisk/dwarves

crates/dwarves/src/brokkr/queue.rs

//! The build queue admission model — *what may start now, and if not, why?*
//!
//! The host backs this with the nornir-jobs ledger (each queued run is a
//! `CI_BUILD` record) and the process-global exclusive gate (`jobs::is_exclusive`
//! → at most one exclusive job at a time). This leaf owns the pure admission
//! decision: an item is admitted when it is not blocked by a running exclusive
//! job AND an idle capable agent exists; otherwise it stays queued *with a
//! reason* (never a silent stall). Exclusivity is carried as a per-item flag so
//! the dep-light leaf need not depend on nornir-jobs.

use std::collections::VecDeque;

use crate::brokkr::agent::AgentPool;

/// A queued build (one pipeline run, or one planned step-instance).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct QueueItem {
    /// Stable id (run id / planned-step label).
    pub id: String,
    /// Whether this item needs the exclusive gate (box-saturating work).
    pub exclusive: bool,
    /// Capability tags the running agent must advertise.
    pub requirements: Vec<String>,
}

/// The admission verdict for a queued item.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Admission {
    /// May start now, on this agent.
    Ready {
        /// The chosen agent's name.
        agent: String,
    },
    /// Stays queued, with a human reason (RAGNARÖK: never a silent stall).
    Blocked {
        /// Why the item cannot start yet.
        reason: String,
    },
}

impl Admission {
    /// Convenience: is this a ready verdict?
    pub fn is_ready(&self) -> bool {
        matches!(self, Admission::Ready { .. })
    }
}

/// Decide whether `item` may start, given whether an exclusive job is already
/// running and the current agent pool.
pub fn admit(item: &QueueItem, exclusive_running: bool, pool: &AgentPool) -> Admission {
    if item.exclusive && exclusive_running {
        return Admission::Blocked {
            reason: "an exclusive job is already running (waiting for the gate)".to_string(),
        };
    }
    match pool.pick(&item.requirements) {
        Some(a) => Admission::Ready {
            agent: a.name.clone(),
        },
        None => {
            let capable = pool.capable(&item.requirements);
            let reason = if capable.is_empty() {
                format!(
                    "no agent advertises the required capabilities {:?}",
                    item.requirements
                )
            } else {
                format!(
                    "all {} capable agent(s) are busy (requirements {:?})",
                    capable.len(),
                    item.requirements
                )
            };
            Admission::Blocked { reason }
        }
    }
}

/// A FIFO build queue. [`Self::admissible`] walks it in order and returns the
/// items that may start now (each with its assigned agent), leaving the blocked
/// ones queued.
#[derive(Debug, Clone, Default)]
pub struct BuildQueue {
    items: VecDeque<QueueItem>,
}

impl BuildQueue {
    /// Enqueue an item at the back.
    pub fn push(&mut self, item: QueueItem) {
        self.items.push_back(item);
    }

    /// The number of queued items.
    pub fn len(&self) -> usize {
        self.items.len()
    }

    /// Whether the queue is empty.
    pub fn is_empty(&self) -> bool {
        self.items.is_empty()
    }

    /// Walk the queue in FIFO order; return each item's admission verdict.
    /// Admitting one exclusive item flips the running flag so a second exclusive
    /// item behind it stays queued (the gate admits one at a time).
    pub fn verdicts(
        &self,
        mut exclusive_running: bool,
        pool: &AgentPool,
    ) -> Vec<(QueueItem, Admission)> {
        let mut out = Vec::with_capacity(self.items.len());
        for item in &self.items {
            let v = admit(item, exclusive_running, pool);
            if v.is_ready() && item.exclusive {
                exclusive_running = true;
            }
            out.push((item.clone(), v));
        }
        out
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::brokkr::agent::Agent;

    fn pool() -> AgentPool {
        let mut p = AgentPool::default();
        p.register(Agent::new("linux-1", ["linux", "cargo"]));
        p
    }

    #[test]
    fn admits_a_matched_item_and_reports_the_agent() {
        let p = pool();
        let item = QueueItem {
            id: "r1".into(),
            exclusive: false,
            requirements: vec!["linux".into()],
        };
        assert_eq!(
            admit(&item, false, &p),
            Admission::Ready {
                agent: "linux-1".into()
            }
        );
    }

    #[test]
    fn unmatched_item_is_queued_with_a_reason() {
        let p = pool();
        let item = QueueItem {
            id: "r1".into(),
            exclusive: false,
            requirements: vec!["windows".into()],
        };
        match admit(&item, false, &p) {
            Admission::Blocked { reason } => assert!(reason.contains("no agent"), "{reason}"),
            other => panic!("expected blocked, got {other:?}"),
        }
    }

    #[test]
    fn exclusive_gate_admits_one_at_a_time() {
        let p = pool();
        let mut q = BuildQueue::default();
        q.push(QueueItem {
            id: "bench-1".into(),
            exclusive: true,
            requirements: vec!["cargo".into()],
        });
        q.push(QueueItem {
            id: "bench-2".into(),
            exclusive: true,
            requirements: vec!["cargo".into()],
        });
        let v = q.verdicts(false, &p);
        assert!(v[0].1.is_ready(), "the first exclusive item starts");
        match &v[1].1 {
            Admission::Blocked { reason } => assert!(reason.contains("exclusive"), "{reason}"),
            other => panic!("second exclusive item must stay queued, got {other:?}"),
        }
    }
}