vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

swarm-builder: scope the forge token to the forge, one build at a time, opt-in

Browse source 
argus's review on #5177: the git credential helper answered every `get`,
so a config PR's own flake inputs could have git send the builder's forge
token to any host. The helper is now configured under
`credential.<forge-url>.helper` and itself answers only an https request
whose host is the forge's; a test runs it as git does and checks that
other hosts, http and non-get operations get nothing.

Builds are bounded: every `build_agent` node needs the one `builder` jobq
resource, so one runs at a time and the rest wait pending, and a newer
head cancels its PR's pending build (a running one finishes and writes on
its own commit). The builder also holds a one-permit semaphore around each
build.

`builder` and `review-classifier`, the forge logins swarm-controller owns,
join nix/reserved-names.nix, with a test tying the list to both constants.

The builder is now opt-in like the CI runner: `allSwarmServices` no longer
turns it on, since it runs unreviewed config PRs through the host's nix
daemon. mara had not answered atlas's question on the PR.
This commit is contained in:
atlas 2026-10-12 22:47:30 +02:00
commit cafa120d77
13 changed files with 360 additions and 70 deletions
Download patch file Download diff file Expand all files Collapse all files

57
swarm-controller/src/builder.rs
View file

@ -4,6 +4,10 @@
//! The status is the check's only record besides the node itself: nothing is
//! pushed, and the closure the builder made stays in its host's store. A red
//! status blocks nothing; the merge deploys as before.
//!
//! Builds share one `SwarmResourceKind::Builder` slot, so one runs at a time,
//! and a newer push replaces its PR's build while that build is still
//! pending (`queue_build`).
use std::future::Future;
@ -227,6 +231,59 @@ mod tests {
);
}
/// A superseded build that finishes after the newer head's writes only on
/// its own commit, so the newer head keeps its status.
#[tokio::test]
async fn a_late_superseded_build_writes_only_on_its_own_commit() {
struct ByRevQueue;
impl BuildSink for ByRevQueue {
async fn build(&self, request: Vec<u8>) -> Result<Reply, String> {
let sent: BuildAgent = serde_json::from_slice(&request).expect("a build request");
Ok(if sent.rev == "old" {
Reply::Failed {
error_tail: "error: boom".to_owned(),
}
} else {
Reply::Built {
out_path: "/nix/store/new".to_owned(),
}
})
}
}
#[derive(Default)]
struct ByRev(Mutex<Vec<(String, CommitStatusState)>>);
impl StatusSink for ByRev {
async fn set(
&self,
_: &str,
rev: &str,
state: CommitStatusState,
_: &str,
) -> anyhow::Result<()> {
self.0.lock().unwrap().push((rev.to_owned(), state));
Ok(())
}
}
let statuses = ByRev::default();
let at = |rev: &str| BuildAgent {
rev: rev.to_owned(),
..job()
};
run(Some(&ByRevQueue), Some(&statuses), at("new")).await;
run(Some(&ByRevQueue), Some(&statuses), at("old")).await;
let written = statuses.0.into_inner().unwrap();
let last_on = |rev: &str| {
written
.iter()
.rev()
.find(|(r, _)| r == rev)
.map(|(_, s)| *s)
};
assert_eq!(last_on("new"), Some(CommitStatusState::Success));
assert_eq!(last_on("old"), Some(CommitStatusState::Failure));
}
#[tokio::test]
async fn no_builder_is_an_error_status_not_a_failure() {
let (outcome, states) =

172
swarm-controller/src/main.rs
View file

@ -254,14 +254,20 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind {
}
}
/// Placeholder resource name — same rationale and same "no variants until a
/// real node needs one" shape as [`SwarmNodeKind`].
/// A named counting semaphore a node acquires before it starts. Every name
/// has capacity 1: the scheduler's default, which no caller overrides.
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
enum SwarmResourceKind {}
enum SwarmResourceKind {
/// The swarm builder: one `BuildAgent` node runs at a time, and the rest
/// wait pending, where a newer push can replace them.
Builder,
}
impl hive_jobq_wire::WireResource for SwarmResourceKind {
fn name(&self) -> String {
match *self {}
match self {
SwarmResourceKind::Builder => "builder".to_owned(),
}
}
}
@ -1960,29 +1966,58 @@ fn claiming_hives(
/// whichever hive the agent runs: the check builds on the builder's own
/// agent base.
fn queue_config_build(state: &AppState, changed: config_pr::ChangedConfigPr) {
let config_pr::ChangedConfigPr { agent, rev, pr } = changed;
let inserted = state
.jobq
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.insert_job(None, |b| {
vec![
b.node(SwarmNodeKind::BuildAgent {
agent: agent.clone(),
rev: rev.clone(),
pr,
})
.guid(),
]
});
match inserted {
Ok(_) => tracing::info!(%agent, %rev, pr, "config-pr head: build queued"),
let (agent, rev, pr) = (changed.agent.clone(), changed.rev.clone(), changed.pr);
let queued = queue_build(
&mut state
.jobq
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner),
changed,
);
match queued {
Ok(superseded) => {
tracing::info!(%agent, %rev, pr, superseded, "config-pr head: build queued");
}
Err(e) => {
tracing::warn!(%agent, %rev, pr, error = %e, "config-pr head: queueing the build failed");
}
}
}
/// Cancel every pending build of the same PR, then insert the new one behind
/// [`SwarmResourceKind::Builder`]. A build already running is left to finish:
/// its status lands on its own commit, not on the new head. Returns how many
/// pending builds were cancelled.
fn queue_build(
sched: &mut hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>,
changed: config_pr::ChangedConfigPr,
) -> Result<usize> {
let config_pr::ChangedConfigPr { agent, rev, pr } = changed;
let stale: Vec<hive_jobq::NodeId> = sched
.graph()
.nodes()
.filter(|n| n.state == hive_jobq::State::Pending)
.filter(|n| {
matches!(&n.payload, SwarmNodeKind::BuildAgent { agent: a, pr: p, .. } if *a == agent && *p == pr)
})
.map(|n| n.id)
.collect();
let superseded = stale
.into_iter()
.filter(|&id| sched.cancel_node(id))
.count();
sched
.insert_job(None, |b| {
vec![
b.node(SwarmNodeKind::BuildAgent { agent, rev, pr })
.needs(SwarmResourceKind::Builder)
.guid(),
]
})
.map_err(|e| anyhow::anyhow!("{e}"))?;
Ok(superseded)
}
/// Queue the deploy of a merged config PR on the one hive that runs its agent.
///
/// The hive is found by scanning every hive's wanted state, since nothing
@ -4147,6 +4182,101 @@ mod tests {
}
}
fn changed(agent: &str, rev: &str, pr: u64) -> crate::config_pr::ChangedConfigPr {
crate::config_pr::ChangedConfigPr {
agent: agent.to_owned(),
rev: rev.to_owned(),
pr,
}
}
fn build_state(
sched: &hive_jobq::scheduler::Scheduler<SwarmNodeKind, super::SwarmResourceKind>,
rev: &str,
) -> hive_jobq::State {
sched
.graph()
.nodes()
.find(|n| matches!(&n.payload, SwarmNodeKind::BuildAgent { rev: r, .. } if r == rev))
.map(|n| n.state)
.expect("a build of that rev")
}
/// A newer head of the same PR replaces its pending build; another PR's
/// build, and the same PR's running build, are left alone.
#[tokio::test]
async fn a_newer_push_replaces_the_prs_pending_build_only() {
let sched =
std::sync::Arc::new(std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new(
hive_jobq::Graph::new(),
hive_jobq::resources::ResourceTable::new(),
)));
let queue = |c| super::queue_build(&mut sched.lock().unwrap(), c).expect("queued");
assert_eq!(queue(changed("atlas", "r1", 3)), 0);
// r1 takes the builder and stays running: its future is never awaited.
let _running = hive_jobq::scheduler::Scheduler::claim_next(&sched, |_, _, b| async move {
(b, hive_jobq::scheduler::Outcome::Done)
})
.expect("r1 is runnable");
assert_eq!(
queue(changed("atlas", "r2", 3)),
0,
"r1 is running, not pending"
);
assert_eq!(queue(changed("argus", "o1", 4)), 0);
assert_eq!(queue(changed("atlas", "r3", 3)), 1, "r2 was pending");
let guard = sched.lock().unwrap();
assert_eq!(build_state(&guard, "r1"), hive_jobq::State::Running);
assert_eq!(build_state(&guard, "r2"), hive_jobq::State::Cancelled);
assert_eq!(build_state(&guard, "r3"), hive_jobq::State::Pending);
assert_eq!(build_state(&guard, "o1"), hive_jobq::State::Pending);
}
/// Builds share one builder slot: a second build is not claimable while
/// the first runs, and is once it has finished.
#[tokio::test]
async fn one_build_runs_at_a_time() {
let sched =
std::sync::Arc::new(std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new(
hive_jobq::Graph::new(),
hive_jobq::resources::ResourceTable::new(),
)));
for c in [changed("atlas", "a1", 1), changed("argus", "b1", 2)] {
super::queue_build(&mut sched.lock().unwrap(), c).expect("queued");
}
let run = |sched| {
hive_jobq::scheduler::Scheduler::claim_next(sched, |_, _, b| async move {
(b, hive_jobq::scheduler::Outcome::Done)
})
};
let first = run(&sched).expect("one build is runnable");
assert!(
run(&sched).is_none(),
"a second build started beside the first"
);
first.await.1.expect("no growth");
run(&sched)
.expect("the second build runs once the first is done")
.await
.1
.expect("no growth");
}
#[test]
fn a_new_agent_named_like_a_service_forge_account_is_refused() {
let raw = nix_reserved();
for name in [
crate::forge::builder_account::LOGIN,
crate::forge::classifier_account::LOGIN,
] {
let mut warnings = Vec::new();
let broken = super::broken_name_rules(name, Some(&raw), &mut warnings);
let err = super::name_verdict(name, &broken, Ok(false), &mut warnings).expect_err(name);
assert!(err.contains("nix/reserved-names.nix"), "{name:?}: {err}");
}
}
/// One case per shape rule, each named in the refusal. Checked against a
/// list without them, so the literal half cannot carry this test.
#[test]