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

read the running rev and the hyperhive flake from /etc files, not the unit

Browse source
This commit is contained in:
atlas 2026-11-06 04:44:27 +01:00 • committed by mara
commit aed7c8451a
2 changed files with 416 additions and 85 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -792,7 +792,14 @@ async fn update_meta(
.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,
meta_flake::running_rev().as_deref(),
&agents,
&meta.hives,
),
)
.await
.map_err(|e| format!("{e:#}"))?;

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

@ -13,6 +13,7 @@
use std::collections::BTreeMap;
use std::future::Future;
use std::path::Path;
use anyhow::{Context, Result, bail};
use serde::Deserialize;
@ -30,6 +31,11 @@ pub(crate) const META_REPO: &str = "meta";
/// The env var holding [`Settings`] as JSON, set by `swarm-controller.nix`.
pub(crate) const SETTINGS_ENV: &str = "SWARM_CONTROLLER_META_FLAKE";
/// The file `swarm-controller.nix` writes the running rev to, which every
/// [`update`] reads again through [`running_rev`]. The unit does not carry the
/// rev: it differs on every commit, and a changed unit restarts the controller.
const REV_FILE: &str = "/etc/hyperhive/swarm-controller.json";
/// Where hyperhive itself lives on the forge.
const HYPERHIVE_OWNER: &str = "hyperhive";
const HYPERHIVE_REPO: &str = "hyperhive";
@ -46,12 +52,49 @@ 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.
}
/// The contents of [`REV_FILE`].
#[derive(Deserialize)]
struct RevFile {
/// The hyperhive commit this controller runs (`self.rev`); `null` for a
/// dirty tree.
#[serde(default)]
hyperhive_rev: Option<String>,
}
/// The hyperhive commit this controller runs, from [`REV_FILE`]; `None` when
/// the file or the value is absent, empty or not a full commit id, which leaves
/// the hyperhive pin where it is.
pub(crate) fn running_rev() -> Option<String> {
read_rev_file(Path::new(REV_FILE))
}
fn read_rev_file(path: &Path) -> Option<String> {
let raw = match std::fs::read_to_string(path) {
Ok(raw) => raw,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return None,
Err(e) => {
tracing::warn!(path = %path.display(), error = %e, "meta flake: cannot read the running rev file");
return None;
}
};
let file: RevFile = match serde_json::from_str(&raw) {
Ok(file) => file,
Err(e) => {
tracing::warn!(path = %path.display(), error = %e, "meta flake: the running rev file does not parse");
return None;
}
};
let rev = file.hyperhive_rev.filter(|r| !r.is_empty())?;
if hive_meta_flake::is_rev(&rev) {
Some(rev)
} else {
tracing::warn!(path = %path.display(), %rev, "meta flake: the running rev is not a full commit id");
None
}
}
impl Settings {
/// [`Settings`] from [`SETTINGS_ENV`], `None` when it is unset.
pub(crate) fn from_env() -> Result<Option<Self>> {
@ -82,13 +125,6 @@ impl Settings {
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)
}
@ -216,6 +252,9 @@ pub(crate) struct Report {
/// Bring `internal/meta` up to the forge's current heads for `agents`, and
/// give each of `hives`' forge accounts, and the swarm builder's, read on it.
///
/// `running_rev` is the rev this controller runs ([`running_rev`]), read by the
/// caller for each update; `None` leaves the hyperhive pin.
///
/// # Errors
/// A head that cannot be read, a lock the builder refuses, or a refused commit.
/// A failed read grant is logged, not returned: the commit is what hives wait
@ -224,6 +263,7 @@ pub(crate) async fn update(
forge: &impl MetaForge,
locker: &impl LockSink,
settings: &Settings,
running_rev: Option<&str>,
agents: &[Placed],
hives: &[String],
) -> Result<Report> {
@ -253,7 +293,7 @@ pub(crate) async fn update(
let flake_file = sha_of("flake.nix");
let pin = hyperhive_pin(
forge,
settings,
running_rev,
flake_file
.map(|f| String::from_utf8_lossy(&f.bytes))
.as_deref(),
@ -319,7 +359,7 @@ struct HyperhivePin {
/// comparison leaves it too, and the next update asks again.
async fn hyperhive_pin(
forge: &impl MetaForge,
settings: &Settings,
running: Option<&str>,
committed_flake: Option<&str>,
) -> Result<HyperhivePin> {
let Some(flake) = committed_flake else {
@ -338,7 +378,7 @@ async fn hyperhive_pin(
rev: pinned.to_owned(),
moved: false,
};
let Some(running) = settings.hyperhive_rev.as_deref() else {
let Some(running) = running else {
tracing::info!(%pinned, "meta flake: this controller has no rev; keeping the hyperhive pin");
return Ok(keep());
};
@ -488,18 +528,13 @@ mod tests {
.collect()
}
fn settings_running(running: Option<&str>) -> Settings {
let running = running.map_or("null".to_owned(), |r| format!("\"{r}\""));
fn settings() -> Settings {
Settings::parse(&format!(
r#"{{"vars":{{"HIVE_FORGE_URL":"http://forge.t.local","HYPERHIVE_SWARM_NAME":"t"}},"git_base":"{FORGE}/","github_disabled":true,"hyperhive_rev":{running}}}"#
r#"{{"vars":{{"HIVE_FORGE_URL":"http://forge.t.local","HYPERHIVE_SWARM_NAME":"t"}},"git_base":"{FORGE}/","github_disabled":true}}"#
))
.unwrap()
}
fn settings() -> Settings {
settings_running(None)
}
fn head(commit: &str, files: &[(&str, &str)]) -> DefaultBranchFiles {
DefaultBranchFiles {
branch: "main".to_owned(),
@ -700,10 +735,17 @@ mod tests {
.to_owned()
}
async fn run(forge: &FakeForge, locker: &FakeLocker, settings: &Settings) -> Report {
update(forge, locker, settings, &argus_and_reviewer(), &[])
.await
.unwrap()
async fn run(forge: &FakeForge, locker: &FakeLocker, running: Option<&str>) -> Report {
update(
forge,
locker,
&settings(),
running,
&argus_and_reviewer(),
&[],
)
.await
.unwrap()
}
#[tokio::test]
@ -714,6 +756,7 @@ mod tests {
&forge,
&locker,
&settings(),
None,
&argus_and_reviewer(),
&["pr1ma".to_owned()],
)
@ -769,7 +812,7 @@ mod tests {
async fn an_unchanged_render_sends_no_lock_and_commits_nothing() {
let forge = forge_pinned_at(&rev('1')).await;
let locker = FakeLocker::locked();
let got = run(&forge, &locker, &settings()).await;
let got = run(&forge, &locker, None).await;
assert_eq!(got.updated, Updated::Unchanged);
assert_eq!(got.pin, rev('1'));
assert!(!got.hyperhive_moved);
@ -786,7 +829,7 @@ mod tests {
head("m1", &[("flake.nix", &old), ("flake.lock", "{}")]),
)]);
let locker = FakeLocker::locked();
run(&forge, &locker, &settings()).await;
run(&forge, &locker, None).await;
let commits = forge.commits.lock().unwrap();
let shas: Vec<_> = commits[0].1.iter().map(|(_, _, sha)| sha.clone()).collect();
assert_eq!(
@ -807,9 +850,16 @@ mod tests {
}),
sent: Mutex::default(),
};
let err = update(&forge, &locker, &settings(), &argus_and_reviewer(), &[])
.await
.unwrap_err();
let err = update(
&forge,
&locker,
&settings(),
None,
&argus_and_reviewer(),
&[],
)
.await
.unwrap_err();
assert!(format!("{err:#}").contains("no such input"));
assert!(forge.commits.lock().unwrap().is_empty());
}
@ -823,7 +873,7 @@ mod tests {
archetype: None,
});
let locker = FakeLocker::locked();
let got = update(&forge, &locker, &settings(), &agents, &[])
let got = update(&forge, &locker, &settings(), None, &agents, &[])
.await
.unwrap();
assert_eq!(got.updated, Updated::Committed);
@ -842,7 +892,7 @@ mod tests {
#[tokio::test]
async fn a_config_head_is_read_from_main_not_the_default_branch() {
let forge = forge_with(&[]);
run(&forge, &FakeLocker::locked(), &settings()).await;
run(&forge, &FakeLocker::locked(), None).await;
let reads = forge.head_reads.lock().unwrap();
assert!(
reads.contains(&("agent-configs/argus".to_owned(), Some("main".to_owned()))),
@ -858,7 +908,7 @@ mod tests {
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;
let got = run(&forge, &locker, Some(&rev('3'))).await;
assert_eq!(locked_pin(&locker), rev('1'));
assert!(
!got.hyperhive_moved,
@ -874,7 +924,7 @@ mod tests {
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;
let got = run(&forge, &locker, Some(&rev('3'))).await;
assert_eq!(locked_pin(&locker), rev('3'));
assert_eq!(
got,
@ -899,7 +949,7 @@ mod tests {
] {
let forge = forge_pinned_at(&pin).await;
let locker = FakeLocker::locked();
let got = run(&forge, &locker, &settings_running(Some(&running))).await;
let got = run(&forge, &locker, Some(&running)).await;
assert_eq!(got.updated, Updated::Unchanged, "{name}");
assert!(!got.hyperhive_moved, "{name}");
assert!(locker.sent.lock().unwrap().is_empty(), "{name}");
@ -914,7 +964,7 @@ mod tests {
.repos
.insert("hyperhive/hyperhive".to_owned(), head(&rev('3'), &[]));
let locker = FakeLocker::locked();
let got = run(&forge, &locker, &settings_running(Some(&rev('2')))).await;
let got = run(&forge, &locker, Some(&rev('2'))).await;
assert_eq!(got.updated, Updated::Unchanged);
assert!(forge.commits.lock().unwrap().is_empty());
}
@ -923,7 +973,7 @@ mod tests {
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;
let got = run(&forge, &locker, None).await;
assert_eq!(got.updated, Updated::Unchanged);
assert!(forge.compares.lock().unwrap().is_empty());
}
@ -942,15 +992,9 @@ mod tests {
}),
});
let locker = FakeLocker::locked();
let got = update(
&forge,
&locker,
&settings_running(Some(&rev('3'))),
&agents,
&[],
)
.await
.unwrap();
let got = update(&forge, &locker, &settings(), Some(&rev('3')), &agents, &[])
.await
.unwrap();
assert_eq!(locked_pin(&locker), rev('2'));
assert!(!got.hyperhive_moved);
assert_eq!(got.updated, Updated::Committed);
@ -962,7 +1006,7 @@ mod tests {
"internal/meta",
head("m1", &[("flake.nix", "{ }"), ("flake.lock", "{}")]),
)]);
let err = update(&forge, &FakeLocker::locked(), &settings(), &[], &[])
let err = update(&forge, &FakeLocker::locked(), &settings(), None, &[], &[])
.await
.unwrap_err();
assert!(format!("{err:#}").contains("pins no hyperhive commit"));
@ -1009,25 +1053,54 @@ mod tests {
}
#[test]
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| {
fn settings_refuse_an_unknown_value_a_missing_forge_and_a_plain_http_base() {
let parse = |vars: &str, git_base: &str| {
Settings::parse(&format!(
r#"{{"vars":{vars},"git_base":"{git_base}","github_disabled":false,"hyperhive_rev":{rev}}}"#
r#"{{"vars":{vars},"git_base":"{git_base}","github_disabled":false}}"#
))
};
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());
assert!(parse(ok, "https://f").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", "null").is_err());
assert!(parse(ok, "http://f", "null").is_err());
assert!(parse(ok, "https://f", "\"main\"").is_err());
assert!(parse("{}", "https://f").is_err());
assert!(parse(ok, "http://f").is_err());
}
/// The rev file's content as `read_rev_file` sees it; `None` writes no file.
fn rev_from(content: Option<&str>) -> Option<String> {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("swarm-controller.json");
if let Some(content) = content {
std::fs::write(&path, content).unwrap();
}
read_rev_file(&path)
}
#[test]
fn the_running_rev_is_the_files_full_commit_id() {
let json = format!(r#"{{"hyperhive_rev":"{}"}}"#, rev('a'));
assert_eq!(rev_from(Some(&json)), Some(rev('a')));
}
#[test]
fn no_file_no_value_an_empty_value_or_a_bad_value_is_no_rev() {
// Control: `the_running_rev_is_the_files_full_commit_id` reads a rev
// through this same function.
for (name, content) in [
("missing file", None),
("null", Some(r#"{"hyperhive_rev":null}"#)),
("no key", Some("{}")),
("empty value", Some(r#"{"hyperhive_rev":""}"#)),
("short rev", Some(r#"{"hyperhive_rev":"abc123"}"#)),
("a branch", Some(r#"{"hyperhive_rev":"main"}"#)),
("not JSON", Some("not json")),
] {
assert_eq!(rev_from(content), None, "{name}");
}
}
}