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

pin internal/meta's hyperhive on the controller's rev and rebuild agents when it moves

Browse source 
The swarm controller offers its own running rev (services.hyperhive.c0re.
hyperhiveRev) as the hyperhive pin of internal/meta and takes it only when
the forge compare API shows it descends from the committed pin; merges to
hyperhive main no longer move the pin, and the first commit pins main's head.
When a commit moves the pin, every placed agent gets a TriggerDeploy.

A deploy after a meta update no longer waits on it succeeding (after_any), so
a down builder does not hold back hives that render locally.
This commit is contained in:
atlas 2026-10-24 14:29:12 +02:00
commit 22e2d30581
10 changed files with 1027 additions and 148 deletions
Download patch file Download diff file Expand all files Collapse all files

90
swarm-controller/src/forge/meta_repo.rs
View file

@ -9,7 +9,7 @@ use forgejo_api::structs::{
};
use super::objects::SHARED_ORG;
use super::{Client, DefaultBranchFiles};
use super::{Client, DefaultBranchFiles, is_repo_not_found};
use crate::meta_flake::{META_REPO, MetaFile, MetaForge};
impl MetaForge for Client {
@ -36,6 +36,27 @@ impl MetaForge for Client {
.await
}
async fn commits_ahead(
&self,
owner: &str,
repo: &str,
base: &str,
head: &str,
) -> Result<Option<u64>> {
let basehead = format!("{base}...{head}");
match self
.api
.call(|api| api.repo_compare_diff(owner, repo, &basehead).into_future())
.await
{
Ok(compare) => Ok(Some(
u64::try_from(compare.total_commits.unwrap_or(0)).unwrap_or(0),
)),
Err(e) if is_repo_not_found(&e) => Ok(None),
Err(e) => Err(e).with_context(|| format!("compare {basehead} in {owner}/{repo}")),
}
}
async fn commit_meta(&self, message: &str, files: Vec<MetaFile>) -> Result<()> {
let ops: Vec<ChangeFileOperation> = files
.into_iter()
@ -75,3 +96,70 @@ impl MetaForge for Client {
Ok(())
}
}
#[cfg(test)]
mod tests {
use std::sync::{Arc, Mutex};
use super::*;
/// A stand-in forge answering Forgejo's compare endpoint: `total_commits`
/// of 3 for `aaa...bbb`, 0 for `bbb...aaa`, 404 for anything else.
/// Records every request's path.
async fn compare_forge() -> (Client, Arc<Mutex<Vec<String>>>) {
use axum::http::{StatusCode, Uri};
let seen = Arc::new(Mutex::new(Vec::new()));
let recorded = Arc::clone(&seen);
let app = axum::Router::new().fallback(move |uri: Uri| {
let recorded = Arc::clone(&recorded);
async move {
recorded.lock().unwrap().push(uri.path().to_owned());
match uri.path() {
"/api/v1/repos/o/r/compare/aaa...bbb" => (
StatusCode::OK,
r#"{"total_commits":3,"commits":[],"files":[]}"#,
),
"/api/v1/repos/o/r/compare/bbb...aaa" => {
(StatusCode::OK, r#"{"total_commits":0,"commits":[]}"#)
}
_ => (StatusCode::NOT_FOUND, r#"{"message":"not found"}"#),
}
}
});
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let url = url::Url::parse(&format!("http://{}/", listener.local_addr().unwrap())).unwrap();
tokio::spawn(async move { axum::serve(listener, app).await });
let api = crate::forge::api::tests::fixed(url.clone(), "stub-token");
(Client { api, url }, seen)
}
#[tokio::test]
async fn commits_ahead_reads_the_compare_endpoint_as_base_dots_head() {
let (client, seen) = compare_forge().await;
assert_eq!(
client.commits_ahead("o", "r", "aaa", "bbb").await.unwrap(),
Some(3)
);
assert_eq!(
client.commits_ahead("o", "r", "bbb", "aaa").await.unwrap(),
Some(0)
);
assert_eq!(
*seen.lock().unwrap(),
[
"/api/v1/repos/o/r/compare/aaa...bbb",
"/api/v1/repos/o/r/compare/bbb...aaa"
]
);
}
#[tokio::test]
async fn a_commit_the_forge_lacks_is_no_count_not_an_error() {
let (client, _) = compare_forge().await;
assert_eq!(
client.commits_ahead("o", "r", "aaa", "zzz").await.unwrap(),
None
);
}
}

364
swarm-controller/src/jobq.rs
View file

@ -134,12 +134,12 @@ pub(crate) enum SwarmNodeKind {
/// agent creation, where the hive deploys what it seeds.
rev: Option<String>,
},
/// Pin the swarm meta flake (`internal/meta`) to the forge's current
/// heads: render, lock on the swarm builder, commit. See `meta_flake`.
/// Pin the swarm meta flake (`internal/meta`): render, lock on the swarm
/// builder, commit. See `meta_flake`. When the commit moves the hyperhive
/// pin, a `TriggerDeploy` per placed agent follows.
///
/// Carries nothing: it reads every head when it runs, so one pending node
/// covers every trigger queued behind it. Hives build from that repo, so a
/// deploy waits for this node.
/// covers every trigger queued behind it.
UpdateMeta,
}
@ -395,7 +395,7 @@ pub(crate) async fn run_swarm_node(
),
Some(client) => publish_deploy(&client, &hive, &agent, rev).await,
},
SwarmNodeKind::UpdateMeta => update_meta(&deps, forge).await,
SwarmNodeKind::UpdateMeta => finish_update_meta(update_meta(&deps, forge).await, &builder),
};
(builder, outcome)
}
@ -565,64 +565,104 @@ pub(crate) async fn build_agent(
.await
}
/// The `UpdateMeta` arm, lifted out for the same reason.
/// The `UpdateMeta` arm, lifted out for the same reason: what the update
/// reported, and the placements it rendered for. `Err` is the node's failure.
async fn update_meta(
deps: &WorkerDeps,
forge: Option<Arc<forge::Client>>,
) -> hive_jobq::scheduler::Outcome {
use hive_jobq::scheduler::Outcome;
) -> Result<
(
meta_flake::Report,
std::collections::BTreeSet<(String, String)>,
),
String,
> {
let Some(client) = forge else {
return Outcome::Failed(
return Err(
"no forge access here (SWARM_CONTROLLER_FORGE_URL unset, or its token \
not in the swarm secret store), so internal/meta cannot be written"
.to_owned(),
);
};
let Some(meta) = meta_deps() else {
return Outcome::Failed(format!(
return Err(format!(
"{} is unset, so there are no swarm values to render internal/meta from",
meta_flake::SETTINGS_ENV
));
};
let (Some(queue), Some(wanted)) = (&deps.queue, &deps.wanted) else {
return Outcome::Failed(
"no swarm queue is configured on this host, so neither the placed agents nor a \
lock can be had"
return Err(
"no swarm queue is configured on this host, so neither the placed agents \
nor a lock can be had"
.to_owned(),
);
};
let agents = match placed_agents(queue, wanted, &meta.hives).await {
Ok(agents) => agents,
Err(e) => return Outcome::Failed(format!("{e:#}")),
};
match forge_node(
let placements = placements(wanted, &meta.hives)
.await
.map_err(|e| format!("{e:#}"))?;
let agents = placed_agents(queue, &placements)
.await
.map_err(|e| format!("{e:#}"))?;
let report = forge_node(
"update_meta",
meta_flake::update(
client.as_ref(),
queue,
&meta.settings,
&agents,
&meta.hives,
),
meta_flake::update(client.as_ref(), queue, &meta.settings, &agents, &meta.hives),
)
.await
{
Ok(updated) => {
tracing::info!(?updated, agents = agents.len(), "update_meta: done");
.map_err(|e| format!("{e:#}"))?;
tracing::info!(?report, agents = agents.len(), "update_meta: done");
Ok((report, placements))
}
/// The `UpdateMeta` node's outcome from what [`update_meta`] reported, with the
/// deploys a moved pin calls for declared into `builder` (the scheduler inserts
/// them under the node when it completes). Synchronous: the builder is not
/// `Sync`, so it cannot be held across an await.
fn finish_update_meta(
result: Result<
(
meta_flake::Report,
std::collections::BTreeSet<(String, String)>,
),
String,
>,
builder: &hive_jobq::builder::JobBuilder<SwarmNodeKind, SwarmResourceKind>,
) -> hive_jobq::scheduler::Outcome {
use hive_jobq::scheduler::Outcome;
match result {
Ok((report, placements)) => {
fan_out_if_moved(builder, &report, &placements);
Outcome::Done
}
Err(e) => Outcome::Failed(format!("{e:#}")),
Err(e) => Outcome::Failed(e),
}
}
/// Every agent some hive in `hives` places, with its archetype.
async fn placed_agents(
queue: &async_nats::Client,
/// Declare one rebuild of each `(hive, agent)` placement when `report` says
/// the agents' hyperhive moved: nothing else tells them to build again.
fn fan_out_if_moved(
grown: &hive_jobq::builder::JobBuilder<SwarmNodeKind, SwarmResourceKind>,
report: &meta_flake::Report,
placements: &std::collections::BTreeSet<(String, String)>,
) {
if !report.hyperhive_moved {
return;
}
for (hive, agent) in placements {
let _ = grown.node(SwarmNodeKind::TriggerDeploy {
hive: hive.clone(),
agent: agent.clone(),
rev: None,
});
}
}
/// Every `(hive, agent)` a hive in `hives` places.
async fn placements(
wanted: &wanted::WantedWriter,
hives: &[String],
) -> Result<Vec<meta_flake::Placed>> {
let mut names = std::collections::BTreeSet::new();
) -> Result<std::collections::BTreeSet<(String, String)>> {
let mut placed = std::collections::BTreeSet::new();
for hive in hives {
let Some(declared) = wanted
.view(hive)
@ -631,14 +671,24 @@ async fn placed_agents(
else {
continue;
};
names.extend(
placed.extend(
declared
.agents
.into_iter()
.filter(|(_, w)| crate::routes::agents::places_agent(w.state))
.map(|(name, _)| name),
.map(|(name, _)| (hive.clone(), name)),
);
}
Ok(placed)
}
/// Every agent in `placements`, with its archetype.
async fn placed_agents(
queue: &async_nats::Client,
placements: &std::collections::BTreeSet<(String, String)>,
) -> Result<Vec<meta_flake::Placed>> {
let names: std::collections::BTreeSet<&String> =
placements.iter().map(|(_, agent)| agent).collect();
let archetypes = swarm_queue_client::agent_archetype::open_or_create(queue)
.await
.map_err(|e| anyhow::anyhow!("open the agent-archetype bucket: {e}"))?;
@ -646,10 +696,13 @@ async fn placed_agents(
for name in names {
let (archetype, _, _) = crate::schedules::read_or_default::<
Option<swarm_wire::agent_archetype::ArchetypeRef>,
>(&archetypes, &name)
>(&archetypes, name)
.await
.map_err(|e| e.context(format!("read the archetype of {name}")))?;
placed.push(meta_flake::Placed { name, archetype });
placed.push(meta_flake::Placed {
name: name.clone(),
archetype,
});
}
Ok(placed)
}
@ -983,22 +1036,7 @@ pub(crate) async fn queue_merged_config_deploy(
.jobq
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.insert_job(None, |b| {
// The hive builds the merged commit from `internal/meta`, so the
// flake must pin it before the hive is told.
let update_meta = b
.node(SwarmNodeKind::UpdateMeta)
.needs(SwarmResourceKind::Meta);
let root = update_meta.guid();
let _deploy = b
.node(SwarmNodeKind::TriggerDeploy {
hive: hive.clone(),
agent: agent.clone(),
rev: Some(rev.clone()),
})
.after_ok(update_meta);
vec![root]
});
.insert_job(None, |b| merged_deploy_job(b, &hive, &agent, &rev));
match inserted {
Ok(_) => tracing::info!(%agent, %rev, %hive, "config-pr merge: deploy queued"),
Err(e) => {
@ -1007,6 +1045,33 @@ pub(crate) async fn queue_merged_config_deploy(
}
}
/// The sub-DAG a merged config PR is: the meta update, then the deploy of
/// `rev` on `hive`.
///
/// `after_any`: a hive builds from `internal/meta` only while it pins the
/// agent's config commit, and renders locally otherwise, as a hive that
/// predates the repo always does. A failed update (no builder, an unreadable
/// config repo) must not hold those hives back.
fn merged_deploy_job(
b: &hive_jobq::builder::JobBuilder<SwarmNodeKind, SwarmResourceKind>,
hive: &str,
agent: &str,
rev: &str,
) -> Vec<hive_jobq::builder::NodeGuid> {
let update_meta = b
.node(SwarmNodeKind::UpdateMeta)
.needs(SwarmResourceKind::Meta);
let root = update_meta.guid();
let _deploy = b
.node(SwarmNodeKind::TriggerDeploy {
hive: hive.to_owned(),
agent: agent.to_owned(),
rev: Some(rev.to_owned()),
})
.after_any(update_meta);
vec![root]
}
/// The sub-DAG one agent creation is: the nodes, and the edges between them.
///
/// A function rather than a closure inside [`create_agent`](crate::routes::agents::create_agent) so the endpoint's
@ -1095,8 +1160,10 @@ pub(crate) fn declare_agent_job(
hive: hive.to_owned(),
agent: agent.to_owned(),
});
// The hive builds the agent from `internal/meta`, which can only carry it
// once the agent is declared on a hive and its config repo has a commit.
// The hive builds the agent from `internal/meta` once it pins the agent's
// config commit, and renders locally until then. `after_any` on the
// update, so one that failed (no builder, an unreadable config repo)
// still lets the hive build the agent that way.
let update_meta = b
.node(SwarmNodeKind::UpdateMeta)
.needs(SwarmResourceKind::Meta)
@ -1131,7 +1198,7 @@ pub(crate) fn declare_agent_job(
agent: agent.to_owned(),
rev: None,
})
.after_ok(update_meta)
.after_any(update_meta)
.after_any(mint_identity)
.after_any(mint_forge_token)
// `after_any` for the same reason: the container reads this token
@ -1907,6 +1974,187 @@ mod tests {
);
}
/// Whether the edge from the `from` node to the `on` node accepts a
/// failed `on`, in a graph built by `declare`.
fn edge_accepts_failure(
declare: impl FnOnce(&hive_jobq::builder::JobBuilder<SwarmNodeKind, SwarmResourceKind>),
from: &str,
on: &str,
) -> bool {
use hive_jobq_wire::WireNode as _;
let mut sched = hive_jobq::scheduler::Scheduler::new(
hive_jobq::Graph::new(),
hive_jobq::resources::ResourceTable::new(),
);
sched
.insert_job(None, |b| {
declare(b);
Vec::new()
})
.expect("the graph inserts");
let graph = sched.graph();
let id_of = |label: &str| {
graph
.nodes()
.find(|n| n.payload.label() == label)
.unwrap_or_else(|| panic!("the graph holds a {label} node"))
.id
};
let (from, on) = (id_of(from), id_of(on));
graph
.node(from)
.expect("just found")
.deps
.iter()
.find_map(|d| match d {
hive_jobq::Dep::Node { id, when } if *id == on => Some(*when),
_ => None,
})
.expect("the edge exists")
.accepts(hive_jobq::TerminalState::Failed)
}
/// A deploy that waits for the meta update must still run when the update
/// fails: a hive that renders locally never needed it.
#[test]
fn a_failed_meta_update_does_not_hold_back_the_deploys() {
let create = |b: &hive_jobq::builder::JobBuilder<SwarmNodeKind, SwarmResourceKind>| {
let _ = super::declare_agent_job(b, "atlas", "pr1ma", false);
};
let merge = |b: &hive_jobq::builder::JobBuilder<SwarmNodeKind, SwarmResourceKind>| {
let _ = super::merged_deploy_job(b, "pr1ma", "atlas", "abc");
};
assert!(edge_accepts_failure(
create,
"trigger_deploy",
"update_meta"
));
assert!(edge_accepts_failure(merge, "trigger_deploy", "update_meta"));
// The same lookup finds the edges that do hold a deploy back.
assert!(!edge_accepts_failure(
create,
"trigger_deploy",
"set_agent_wanted"
));
assert!(!edge_accepts_failure(
create,
"update_meta",
"set_agent_wanted"
));
}
/// The scheduler's side of the same edge: with the update failed, the
/// deploy is the next node to run.
#[tokio::test]
async fn the_merged_deploy_runs_after_a_failed_meta_update() {
use hive_jobq::scheduler::{Outcome, Scheduler};
use hive_jobq_wire::WireNode as _;
let mut sched = Scheduler::new(
hive_jobq::Graph::new(),
hive_jobq::resources::ResourceTable::new(),
);
sched
.insert_job(None, |b| {
super::merged_deploy_job(b, "pr1ma", "atlas", "abc")
})
.expect("the graph inserts");
let sched = std::sync::Arc::new(std::sync::Mutex::new(sched));
let mut ran = Vec::new();
while let Some(runner) = Scheduler::claim_next(&sched, |_id, kind, builder| async move {
let outcome = match kind {
SwarmNodeKind::UpdateMeta => Outcome::Failed("no builder".to_owned()),
_ => Outcome::Done,
};
(builder, outcome)
}) {
runner.await.1.expect("no growth declared");
ran.push(
sched
.lock()
.unwrap()
.graph()
.nodes()
.filter(|n| n.state.is_terminal())
.map(|n| (n.payload.label(), n.state))
.collect::<Vec<_>>(),
);
}
let last = ran.last().expect("both nodes ran");
assert!(
last.contains(&("update_meta".to_owned(), hive_jobq::State::Failed)),
"{last:?}"
);
assert!(
last.contains(&("trigger_deploy".to_owned(), hive_jobq::State::Done)),
"{last:?}"
);
}
/// Run an `UpdateMeta` node that reports `report`, and return the
/// `(hive, agent, rev)` of every deploy it grew.
async fn grown_deploys(report: meta_flake::Report) -> Vec<(String, String, Option<String>)> {
use hive_jobq::scheduler::{Outcome, Scheduler};
let mut sched = Scheduler::new(
hive_jobq::Graph::new(),
hive_jobq::resources::ResourceTable::new(),
);
let update = sched
.append(SwarmNodeKind::UpdateMeta, Vec::new(), None)
.expect("insert");
let sched = std::sync::Arc::new(std::sync::Mutex::new(sched));
let placements: std::collections::BTreeSet<(String, String)> =
[("pr1ma", "atlas"), ("pr1ma", "argus"), ("sec0nd", "iris")]
.into_iter()
.map(|(h, a)| (h.to_owned(), a.to_owned()))
.collect();
let runner = Scheduler::claim_next(&sched, move |_id, _kind, builder| async move {
fan_out_if_moved(&builder, &report, &placements);
(builder, Outcome::Done)
})
.expect("the node just inserted is runnable");
runner.await.1.expect("the grown nodes insert");
let guard = sched.lock().unwrap();
guard
.graph()
.nodes()
.filter_map(|n| match &n.payload {
SwarmNodeKind::TriggerDeploy { hive, agent, rev } => {
assert_eq!(n.parent, Some(update), "grown under the update");
Some((hive.clone(), agent.clone(), rev.clone()))
}
_ => None,
})
.collect()
}
#[tokio::test]
async fn a_moved_pin_rebuilds_every_placed_agent_and_an_unmoved_one_none() {
let moved = grown_deploys(meta_flake::Report {
updated: meta_flake::Updated::Committed,
hyperhive_moved: true,
})
.await;
let mut moved: Vec<_> = moved.into_iter().collect();
moved.sort();
assert_eq!(
moved,
vec![
("pr1ma".to_owned(), "argus".to_owned(), None),
("pr1ma".to_owned(), "atlas".to_owned(), None),
("sec0nd".to_owned(), "iris".to_owned(), None),
]
);
let unmoved = grown_deploys(meta_flake::Report {
updated: meta_flake::Updated::Committed,
hyperhive_moved: false,
})
.await;
assert!(unmoved.is_empty(), "{unmoved:?}");
}
/// Node labels of the graph [`super::declare_agent_job`] builds, and the
/// labels the deploy node waits on.
fn creation_graph(archetype: bool) -> (Vec<String>, Vec<String>) {

530
swarm-controller/src/meta_flake.rs
View file

@ -1,11 +1,15 @@
//! `internal/meta`: the swarm meta flake hives build their agents from.
//! `internal/meta`: the swarm meta flake hives build from.
//!
//! [`update`] renders `flake.nix` with `hive_meta_flake::render_flake`: hyperhive
//! pinned to its forge `main`, each custom agent's config input pinned to its
//! pinned by [`hyperhive_pin`], each custom agent's config input pinned to its
//! `agent-configs/<agent>` `main`, every agent placed on some hive. The swarm
//! builder locks it (`swarm_wire::builder::LockMeta`), and the flake and the
//! lock are committed together in one commit. A render equal to the committed
//! `flake.nix` sends no lock and commits nothing.
//!
//! The hyperhive pin is a ratchet on this controller's running rev: it moves
//! only to a rev that descends from the committed one, so deploying an older
//! controller never rolls the agents back.
use std::collections::BTreeMap;
use std::future::Future;
@ -39,6 +43,10 @@ pub(crate) struct Settings {
git_base: String,
/// The swarm turned the agents' GitHub integration off.
github_disabled: bool,
/// The hyperhive commit this controller runs (`self.rev`); `None` for a
/// dirty tree, which leaves the pin where it is.
#[serde(default)]
hyperhive_rev: Option<String>,
}
impl Settings {
@ -61,12 +69,23 @@ impl Settings {
{
bail!("{SETTINGS_ENV}: {unknown} is not a value the meta flake forwards");
}
if settings.vars.get("HIVE_FORGE_URL").is_none_or(String::is_empty) {
if settings
.vars
.get("HIVE_FORGE_URL")
.is_none_or(String::is_empty)
{
bail!("{SETTINGS_ENV}: HIVE_FORGE_URL is unset");
}
if !settings.git_base.starts_with("https://") {
bail!("{SETTINGS_ENV}: git_base must be an https origin");
}
if let Some(rev) = settings
.hyperhive_rev
.as_deref()
.filter(|r| !hive_meta_flake::is_rev(r))
{
bail!("{SETTINGS_ENV}: hyperhive_rev {rev:?} is not a full commit id");
}
Ok(settings)
}
@ -110,6 +129,22 @@ pub(crate) trait MetaForge {
repo: &str,
paths: &[&str],
) -> impl Future<Output = Result<Option<DefaultBranchFiles>>> + Send;
/// The number of commits `head` has that `base` lacks, as the forge's
/// compare endpoint counts them: `GET /repos/{owner}/{repo}/compare/
/// {base}...{head}`, whose `total_commits` is the commits from the merge
/// base of the two to `head`. `None` when either commit is not on the
/// forge.
///
/// ⚠️ Two histories with no common ancestor count 0, the same as a
/// `head` already contained in `base`. [`is_strict_descendant`] is the
/// reading that tells them apart.
fn commits_ahead(
&self,
owner: &str,
repo: &str,
base: &str,
head: &str,
) -> impl Future<Output = Result<Option<u64>>> + Send;
fn ensure_meta_repo(&self) -> impl Future<Output = Result<()>> + Send;
fn grant_meta_read(&self, login: &str) -> impl Future<Output = Result<()>> + Send;
fn commit_meta(
@ -159,6 +194,16 @@ pub(crate) enum Updated {
Committed,
}
/// What [`update`] reports.
#[derive(Debug, PartialEq, Eq)]
pub(crate) struct Report {
pub(crate) updated: Updated,
/// The commit moved the hyperhive pin from one rev to another, which is
/// what the placed agents are rebuilt for. The first commit moves nothing:
/// no agent was ever told to build from this repo.
pub(crate) hyperhive_moved: bool,
}
/// Bring `internal/meta` up to the forge's current heads for `agents`, and
/// give each of `hives`' forge accounts read on it.
///
@ -172,7 +217,7 @@ pub(crate) async fn update(
settings: &Settings,
agents: &[Placed],
hives: &[String],
) -> Result<Updated> {
) -> Result<Report> {
forge.ensure_meta_repo().await?;
for hive in hives {
let login = swarm_secret_client::forge::hive_login(hive);
@ -180,7 +225,6 @@ pub(crate) async fn update(
tracing::warn!(%hive, error = %format!("{e:#}"), "meta flake: read grant on internal/meta failed");
}
}
let flake = render(forge, settings, agents).await?;
let committed = forge
.head_files(
crate::forge::objects::SHARED_ORG,
@ -194,8 +238,20 @@ pub(crate) async fn update(
.and_then(|c| c.files.iter().find(|f| f.path == path))
};
let flake_file = sha_of("flake.nix");
let pin = hyperhive_pin(
forge,
settings,
flake_file
.map(|f| String::from_utf8_lossy(&f.bytes))
.as_deref(),
)
.await?;
let flake = render(forge, settings, &pin.rev, agents).await?;
if flake_file.is_some_and(|f| f.bytes == flake.as_bytes()) && sha_of("flake.lock").is_some() {
return Ok(Updated::Unchanged);
return Ok(Report {
updated: Updated::Unchanged,
hyperhive_moved: false,
});
}
let request = serde_json::to_vec(&LockMeta {
flake_nix: flake.clone(),
@ -223,33 +279,116 @@ pub(crate) async fn update(
],
)
.await?;
Ok(Updated::Committed)
Ok(Report {
updated: Updated::Committed,
hyperhive_moved: pin.moved,
})
}
/// Whether a `push` webhook body moved hyperhive's `main`, which the flake
/// pins.
pub(crate) fn moves_hyperhive(body: &[u8]) -> bool {
let Ok(v) = serde_json::from_slice::<serde_json::Value>(body) else {
return false;
/// The hyperhive commit the flake pins, and whether that is a move off the
/// committed pin.
struct HyperhivePin {
rev: String,
moved: bool,
}
/// The hyperhive pin for the next render, given the committed `flake.nix`
/// (`None` before the first commit).
///
/// - No committed flake: the forge's `main` head, the one time the pin follows
/// `main`.
/// - Otherwise the committed pin, unless this controller's running rev is a
/// strict descendant of it ([`is_strict_descendant`]), which becomes the
/// pin. A merge to `main` moves nothing; a controller older than, or on
/// another line from, the pin leaves it. A forge that cannot answer the
/// comparison leaves it too, and the next update asks again.
async fn hyperhive_pin(
forge: &impl MetaForge,
settings: &Settings,
committed_flake: Option<&str>,
) -> Result<HyperhivePin> {
let Some(flake) = committed_flake else {
let head = forge
.head_files(HYPERHIVE_OWNER, HYPERHIVE_REPO, &[])
.await?
.with_context(|| format!("{HYPERHIVE_OWNER}/{HYPERHIVE_REPO} is empty or absent"))?;
return Ok(HyperhivePin {
rev: head.commit,
moved: false,
});
};
let repo = v
.get("repository")
.and_then(|r| r.get("full_name"))
.and_then(serde_json::Value::as_str);
let git_ref = v.get("ref").and_then(serde_json::Value::as_str);
repo == Some(&format!("{HYPERHIVE_OWNER}/{HYPERHIVE_REPO}"))
&& git_ref == Some("refs/heads/main")
let pinned = hive_meta_flake::pinned_rev(flake, "hyperhive")
.context("the committed internal/meta/flake.nix pins no hyperhive commit")?;
let keep = || HyperhivePin {
rev: pinned.to_owned(),
moved: false,
};
let Some(running) = settings.hyperhive_rev.as_deref() else {
tracing::info!(%pinned, "meta flake: this controller has no rev; keeping the hyperhive pin");
return Ok(keep());
};
match is_strict_descendant(forge, HYPERHIVE_OWNER, HYPERHIVE_REPO, pinned, running).await {
Ok(true) => {
tracing::info!(%pinned, %running, "meta flake: moving the hyperhive pin to this controller's rev");
Ok(HyperhivePin {
rev: running.to_owned(),
moved: true,
})
}
Ok(false) => {
tracing::info!(%pinned, %running, "meta flake: this controller's rev is not newer than the hyperhive pin; keeping it");
Ok(keep())
}
Err(e) => {
tracing::warn!(%pinned, %running, error = %format!("{e:#}"), "meta flake: cannot compare this controller's rev with the hyperhive pin; keeping it");
Ok(keep())
}
}
}
/// The `flake.nix` for `agents` at the forge's current heads.
async fn render(forge: &impl MetaForge, settings: &Settings, agents: &[Placed]) -> Result<String> {
let hyperhive = forge
.head_files(HYPERHIVE_OWNER, HYPERHIVE_REPO, &[])
/// Whether `descendant` is a commit of `owner/repo` strictly after `ancestor`:
/// it has commits `ancestor` lacks ([`MetaForge::commits_ahead`]) and
/// `ancestor` has none it lacks. Equal commits, a `descendant` that is behind,
/// histories that diverged and histories with nothing in common are all
/// `false`; a commit the forge does not have is too.
async fn is_strict_descendant(
forge: &impl MetaForge,
owner: &str,
repo: &str,
ancestor: &str,
descendant: &str,
) -> Result<bool> {
if ancestor == descendant {
return Ok(false);
}
// Unrelated histories also count 0, so a zero here is `false` whatever
// the other side says.
let Some(ahead) = forge
.commits_ahead(owner, repo, ancestor, descendant)
.await?
.with_context(|| format!("{HYPERHIVE_OWNER}/{HYPERHIVE_REPO} is empty or absent"))?;
else {
return Ok(false);
};
if ahead == 0 {
return Ok(false);
}
let behind = forge
.commits_ahead(owner, repo, descendant, ancestor)
.await?;
Ok(behind == Some(0))
}
/// The `flake.nix` for `agents` at the forge's current config heads, with
/// hyperhive at `hyperhive_rev`.
async fn render(
forge: &impl MetaForge,
settings: &Settings,
hyperhive_rev: &str,
agents: &[Placed],
) -> Result<String> {
let hyperhive_url = hive_meta_flake::pinned_git_url(
&settings.git_url(HYPERHIVE_OWNER, HYPERHIVE_REPO),
Some(&hyperhive.commit),
Some(hyperhive_rev),
)?;
let mut specs = Vec::with_capacity(agents.len());
for agent in agents {
@ -298,6 +437,7 @@ async fn render(forge: &impl MetaForge, settings: &Settings, agents: &[Placed])
#[cfg(test)]
mod tests {
use std::collections::BTreeSet;
use std::sync::Mutex;
use super::*;
@ -305,13 +445,38 @@ mod tests {
const FORGE: &str = "https://forge.t.local";
fn settings() -> Settings {
/// A full commit id made of one hex digit.
fn rev(digit: char) -> String {
digit.to_string().repeat(40)
}
/// The hyperhive history the fake forge knows: `1 <- 2 <- 3`, `2 <- 4`
/// (4 diverged from 3) and `9` on its own. `8` is not on the forge.
fn history() -> BTreeMap<String, Vec<String>> {
[
('1', vec![]),
('2', vec!['1']),
('3', vec!['2']),
('4', vec!['2']),
('9', vec![]),
]
.into_iter()
.map(|(c, parents)| (rev(c), parents.into_iter().map(rev).collect()))
.collect()
}
fn settings_running(running: Option<&str>) -> Settings {
let running = running.map_or("null".to_owned(), |r| format!("\"{r}\""));
Settings::parse(&format!(
r#"{{"vars":{{"HIVE_FORGE_URL":"http://forge.t.local","HYPERHIVE_SWARM_NAME":"t"}},"git_base":"{FORGE}/","github_disabled":true}}"#
r#"{{"vars":{{"HIVE_FORGE_URL":"http://forge.t.local","HYPERHIVE_SWARM_NAME":"t"}},"git_base":"{FORGE}/","github_disabled":true,"hyperhive_rev":{running}}}"#
))
.unwrap()
}
fn settings() -> Settings {
settings_running(None)
}
fn head(commit: &str, files: &[(&str, &str)]) -> DefaultBranchFiles {
DefaultBranchFiles {
branch: "main".to_owned(),
@ -336,6 +501,28 @@ mod tests {
repos: BTreeMap<String, DefaultBranchFiles>,
commits: Mutex<Vec<Commit>>,
grants: Mutex<Vec<String>>,
/// Hyperhive's commits and their parents.
history: BTreeMap<String, Vec<String>>,
/// Every compare asked, as `(base, head)`.
compares: Mutex<Vec<(String, String)>>,
/// Every compare fails, as a forge that is down does.
compare_fails: bool,
}
impl FakeForge {
/// `commit` and everything it descends from; `None` for a commit not
/// on the forge.
fn ancestry(&self, commit: &str) -> Option<BTreeSet<String>> {
self.history.get(commit)?;
let mut seen = BTreeSet::new();
let mut todo = vec![commit.to_owned()];
while let Some(c) = todo.pop() {
if seen.insert(c.clone()) {
todo.extend(self.history.get(&c).cloned().unwrap_or_default());
}
}
Some(seen)
}
}
impl MetaForge for FakeForge {
@ -351,6 +538,31 @@ mod tests {
h
}))
}
/// Forgejo's `GetCompareInfo` for `base...head`: the commits `head`
/// has beyond the merge base, and none at all when the two share no
/// ancestor.
async fn commits_ahead(
&self,
_owner: &str,
_repo: &str,
base: &str,
head: &str,
) -> Result<Option<u64>> {
self.compares
.lock()
.unwrap()
.push((base.to_owned(), head.to_owned()));
if self.compare_fails {
bail!("the forge is down");
}
let (Some(base), Some(head)) = (self.ancestry(base), self.ancestry(head)) else {
return Ok(None);
};
if base.is_disjoint(&head) {
return Ok(Some(0));
}
Ok(Some(head.difference(&base).count() as u64))
}
async fn ensure_meta_repo(&self) -> Result<()> {
Ok(())
}
@ -395,9 +607,10 @@ mod tests {
}
}
/// A forge whose hyperhive `main` is at commit `1`, with no `internal/meta`.
fn forge_with(extra: &[(&str, DefaultBranchFiles)]) -> FakeForge {
let mut repos = BTreeMap::new();
repos.insert("hyperhive/hyperhive".to_owned(), head("hh0", &[]));
repos.insert("hyperhive/hyperhive".to_owned(), head(&rev('1'), &[]));
repos.insert(
"agent-configs/argus".to_owned(),
head(
@ -413,10 +626,23 @@ mod tests {
}
FakeForge {
repos,
history: history(),
..FakeForge::default()
}
}
/// A forge with `internal/meta` committed, pinning hyperhive at `pin`.
async fn forge_pinned_at(pin: &str) -> FakeForge {
let base = forge_with(&[]);
let flake = render(&base, &settings(), pin, &argus_and_reviewer())
.await
.unwrap();
forge_with(&[(
"internal/meta",
head("m1", &[("flake.nix", &flake), ("flake.lock", "{}")]),
)])
}
fn argus_and_reviewer() -> Vec<Placed> {
vec![
Placed {
@ -433,6 +659,23 @@ mod tests {
]
}
/// The hyperhive commit the one lock sent to `locker` pins.
fn locked_pin(locker: &FakeLocker) -> String {
let sent = locker.sent.lock().unwrap();
let [flake] = sent.as_slice() else {
panic!("one lock: {sent:?}")
};
hive_meta_flake::pinned_rev(flake, "hyperhive")
.expect("a hyperhive pin")
.to_owned()
}
async fn run(forge: &FakeForge, locker: &FakeLocker, settings: &Settings) -> Report {
update(forge, locker, settings, &argus_and_reviewer(), &[])
.await
.unwrap()
}
#[tokio::test]
async fn a_new_render_is_locked_and_committed_with_its_lock_in_one_commit() {
let forge = forge_with(&[]);
@ -446,19 +689,29 @@ mod tests {
)
.await
.unwrap();
assert_eq!(got, Updated::Committed);
assert_eq!(
got,
Report {
updated: Updated::Committed,
hyperhive_moved: false
}
);
let sent = locker.sent.lock().unwrap();
let [flake] = sent.as_slice() else {
panic!("one lock: {sent:?}")
};
assert!(flake.contains(&format!(
"hyperhive.url = \"git+{FORGE}/hyperhive/hyperhive.git?rev=hh0\";"
"hyperhive.url = \"git+{FORGE}/hyperhive/hyperhive.git?rev={}\";",
rev('1')
)));
assert!(flake.contains(&format!(
"agent-argus.url = \"git+{FORGE}/agent-configs/argus.git?rev=a1\";"
)));
assert!(flake.contains("agent-argus.inputs.nixpkgs.follows = \"nixpkgs\";"));
assert!(!flake.contains("agent-rev.url"), "an archetype agent has no input");
assert!(
!flake.contains("agent-rev.url"),
"an archetype agent has no input"
);
assert!(flake.contains("services.hyperhive.agent.github.enable = false;"));
assert!(flake.contains("HYPERHIVE_SWARM_NAME = \"t\";"));
let commits = forge.commits.lock().unwrap();
@ -480,35 +733,25 @@ mod tests {
#[tokio::test]
async fn an_unchanged_render_sends_no_lock_and_commits_nothing() {
let first = forge_with(&[]);
let forge = forge_pinned_at(&rev('1')).await;
let locker = FakeLocker::locked();
update(&first, &locker, &settings(), &argus_and_reviewer(), &[])
.await
.unwrap();
let flake = locker.sent.lock().unwrap()[0].clone();
let forge = forge_with(&[(
"internal/meta",
head("m1", &[("flake.nix", &flake), ("flake.lock", "{}")]),
)]);
let locker = FakeLocker::locked();
let got = update(&forge, &locker, &settings(), &argus_and_reviewer(), &[])
.await
.unwrap();
assert_eq!(got, Updated::Unchanged);
let got = run(&forge, &locker, &settings()).await;
assert_eq!(got.updated, Updated::Unchanged);
assert!(!got.hyperhive_moved);
assert!(locker.sent.lock().unwrap().is_empty());
assert!(forge.commits.lock().unwrap().is_empty());
}
#[tokio::test]
async fn a_moved_head_replaces_both_committed_blobs() {
async fn a_moved_config_head_replaces_both_committed_blobs() {
let base = forge_with(&[]);
let old = render(&base, &settings(), &rev('1'), &[]).await.unwrap();
let forge = forge_with(&[(
"internal/meta",
head("m1", &[("flake.nix", "old"), ("flake.lock", "{}")]),
head("m1", &[("flake.nix", &old), ("flake.lock", "{}")]),
)]);
let locker = FakeLocker::locked();
update(&forge, &locker, &settings(), &argus_and_reviewer(), &[])
.await
.unwrap();
run(&forge, &locker, &settings()).await;
let commits = forge.commits.lock().unwrap();
let shas: Vec<_> = commits[0].1.iter().map(|(_, _, sha)| sha.clone()).collect();
assert_eq!(
@ -550,34 +793,179 @@ mod tests {
assert!(forge.commits.lock().unwrap().is_empty());
}
#[test]
fn only_a_push_to_hyperhive_main_moves_the_pin() {
let push = |repo: &str, git_ref: &str| {
serde_json::to_vec(&serde_json::json!({
"ref": git_ref,
"repository": { "full_name": repo },
"commits": [],
}))
.unwrap()
#[tokio::test]
async fn the_first_commit_pins_the_forge_main_head_whatever_the_controller_runs() {
let forge = forge_with(&[]);
let locker = FakeLocker::locked();
let got = run(&forge, &locker, &settings_running(Some(&rev('3')))).await;
assert_eq!(locked_pin(&locker), rev('1'));
assert!(
!got.hyperhive_moved,
"no agent was told to build from it yet"
);
assert!(
forge.compares.lock().unwrap().is_empty(),
"nothing to compare with before the first commit"
);
}
#[tokio::test]
async fn a_controller_on_a_descendant_moves_the_pin_and_says_so() {
let forge = forge_pinned_at(&rev('2')).await;
let locker = FakeLocker::locked();
let got = run(&forge, &locker, &settings_running(Some(&rev('3')))).await;
assert_eq!(locked_pin(&locker), rev('3'));
assert_eq!(
got,
Report {
updated: Updated::Committed,
hyperhive_moved: true
}
);
}
/// Each running rev that must not move its pin; the control is the test
/// above, where a descendant does.
#[tokio::test]
async fn a_controller_that_is_not_strictly_newer_leaves_the_pin() {
for (name, pin, running) in [
("the same commit", rev('2'), rev('2')),
("an ancestor", rev('2'), rev('1')),
("a diverged commit", rev('3'), rev('4')),
("an unrelated history", rev('2'), rev('9')),
("a commit the forge lacks", rev('2'), rev('8')),
] {
let forge = forge_pinned_at(&pin).await;
let locker = FakeLocker::locked();
let got = run(&forge, &locker, &settings_running(Some(&running))).await;
assert_eq!(got.updated, Updated::Unchanged, "{name}");
assert!(!got.hyperhive_moved, "{name}");
assert!(locker.sent.lock().unwrap().is_empty(), "{name}");
assert!(forge.commits.lock().unwrap().is_empty(), "{name}");
}
}
#[tokio::test]
async fn a_merge_to_hyperhive_main_does_not_move_the_pin() {
let mut forge = forge_pinned_at(&rev('2')).await;
forge
.repos
.insert("hyperhive/hyperhive".to_owned(), head(&rev('3'), &[]));
let locker = FakeLocker::locked();
let got = run(&forge, &locker, &settings_running(Some(&rev('2')))).await;
assert_eq!(got.updated, Updated::Unchanged);
assert!(forge.commits.lock().unwrap().is_empty());
}
#[tokio::test]
async fn a_controller_without_a_rev_keeps_the_pin_and_asks_nothing() {
let forge = forge_pinned_at(&rev('2')).await;
let locker = FakeLocker::locked();
let got = run(&forge, &locker, &settings()).await;
assert_eq!(got.updated, Updated::Unchanged);
assert!(forge.compares.lock().unwrap().is_empty());
}
#[tokio::test]
async fn a_forge_that_cannot_compare_keeps_the_pin_and_still_commits_the_rest() {
let mut forge = forge_pinned_at(&rev('2')).await;
forge.compare_fails = true;
// A new agent makes the render differ from the committed flake.
let mut agents = argus_and_reviewer();
agents.push(Placed {
name: "newcomer".to_owned(),
archetype: Some(ArchetypeRef {
source: "hyperhive".to_owned(),
name: "reviewer".to_owned(),
}),
});
let locker = FakeLocker::locked();
let got = update(
&forge,
&locker,
&settings_running(Some(&rev('3'))),
&agents,
&[],
)
.await
.unwrap();
assert_eq!(locked_pin(&locker), rev('2'));
assert!(!got.hyperhive_moved);
assert_eq!(got.updated, Updated::Committed);
}
#[tokio::test]
async fn a_committed_flake_without_a_hyperhive_pin_is_an_error_not_a_first_commit() {
let forge = forge_with(&[(
"internal/meta",
head("m1", &[("flake.nix", "{ }"), ("flake.lock", "{}")]),
)]);
let err = update(&forge, &FakeLocker::locked(), &settings(), &[], &[])
.await
.unwrap_err();
assert!(format!("{err:#}").contains("pins no hyperhive commit"));
}
/// The fake is Forgejo's `GetCompareInfo`, so the descendant test rests
/// on these: a `base...head` count of 0 is both "head is contained in
/// base" and "no common ancestor", and only the reverse count tells a
/// divergence from a descendant.
#[tokio::test]
async fn descent_needs_commits_ahead_one_way_and_none_the_other() {
let forge = forge_with(&[]);
let ahead = |base: char, head: char| {
let forge = &forge;
async move {
forge
.commits_ahead("o", "r", &rev(base), &rev(head))
.await
.unwrap()
}
};
assert!(moves_hyperhive(&push("hyperhive/hyperhive", "refs/heads/main")));
assert!(!moves_hyperhive(&push("hyperhive/hyperhive", "refs/heads/x")));
assert!(!moves_hyperhive(&push("agents/foo", "refs/heads/main")));
assert!(!moves_hyperhive(b"not json"));
assert_eq!(ahead('2', '3').await, Some(1));
assert_eq!(ahead('1', '3').await, Some(2));
assert_eq!(ahead('3', '2').await, Some(0), "contained in base");
assert_eq!(ahead('2', '9').await, Some(0), "no common ancestor");
assert_eq!(ahead('3', '4').await, Some(1), "diverged: both ways");
assert_eq!(ahead('4', '3').await, Some(1), "diverged: both ways");
assert_eq!(ahead('2', '8').await, None);
let descends = |ancestor: char, descendant: char| {
let forge = &forge;
async move {
is_strict_descendant(forge, "o", "r", &rev(ancestor), &rev(descendant))
.await
.unwrap()
}
};
assert!(descends('2', '3').await);
assert!(descends('1', '3').await);
assert!(!descends('3', '2').await);
assert!(!descends('2', '2').await);
assert!(!descends('3', '4').await);
assert!(!descends('2', '9').await, "unrelated histories");
assert!(!descends('2', '8').await);
}
#[test]
fn settings_refuse_an_unknown_value_a_missing_forge_and_a_plain_http_base() {
let parse = |vars: &str, git_base: &str| {
fn settings_refuse_an_unknown_value_a_missing_forge_a_plain_http_base_and_a_short_rev() {
let parse = |vars: &str, git_base: &str, rev: &str| {
Settings::parse(&format!(
r#"{{"vars":{vars},"git_base":"{git_base}","github_disabled":false}}"#
r#"{{"vars":{vars},"git_base":"{git_base}","github_disabled":false,"hyperhive_rev":{rev}}}"#
))
};
assert!(parse(r#"{"HIVE_FORGE_URL":"x"}"#, "https://f").is_ok());
let unknown = parse(r#"{"HIVE_FORGE_URL":"x","HIVE_AGENT_NATS_URL":"y"}"#, "https://f")
.unwrap_err();
let ok = r#"{"HIVE_FORGE_URL":"x"}"#;
assert!(parse(ok, "https://f", "null").is_ok());
let full = format!("\"{}\"", rev('a'));
assert!(parse(ok, "https://f", &full).is_ok());
let unknown = parse(
r#"{"HIVE_FORGE_URL":"x","HIVE_AGENT_NATS_URL":"y"}"#,
"https://f",
"null",
)
.unwrap_err();
assert!(format!("{unknown:#}").contains("HIVE_AGENT_NATS_URL"));
assert!(parse("{}", "https://f").is_err());
assert!(parse(r#"{"HIVE_FORGE_URL":"x"}"#, "http://f").is_err());
assert!(parse("{}", "https://f", "null").is_err());
assert!(parse(ok, "http://f", "null").is_err());
assert!(parse(ok, "https://f", "\"main\"").is_err());
}
}

5
swarm-controller/src/startup.rs
View file

@ -285,8 +285,9 @@ fn spawn_forge_workers(
auth: Option<Arc<auth::AuthBridge>>,
queue: Option<async_nats::Client>,
) {
// `internal/meta` appears, and catches up with every head that moved while
// this daemon was down.
// `internal/meta` appears, catches up with every config head that moved
// while this daemon was down, and takes this daemon's rev as the hyperhive
// pin when that is newer.
if let Err(e) = crate::jobq::queue_meta_update(jobq) {
tracing::warn!(error = %format!("{e:#}"), "update_meta: queueing failed");
}

5
swarm-controller/src/webhook.rs
View file

@ -443,11 +443,6 @@ pub(super) async fn post_webhook_forge(
}
}
DeliveryKind::VcsActivity => {
if crate::meta_flake::moves_hyperhive(&body)
&& let Err(e) = crate::jobq::queue_meta_update(&state.jobq)
{
tracing::warn!(error = %format!("{e:#}"), "webhook: queueing update_meta failed");
}
if let Some(activity) = crate::vcs_metrics::parse_push(&body) {
crate::vcs_metrics::record_push(&activity);
} else {