//! Runner updates (oxygenhub#322, OxygenFlow docs/jetstream.md §8.2): //! which build each runner should run, under the policy the site, its //! owner, its test groups and the runner itself set, the nearest setting //! winning. Off unless some level turns updates on. use crate::db::runner_updates::{self as rdb, Group, Policy, Soak}; use crate::db::runners::Runner; use crate::db::{self, Pool}; use flow_core::api::{UpdateOffer, UpdatePolicy}; use std::collections::{BTreeSet, HashMap}; /// The site setting naming the repository whose releases carry runner /// builds (`owner/name`), each an asset `flow-OS-ARCH`. pub const SETTING_REPO: &str = "runner_updates_repo"; /// The site settings for proving a version in a wave (oxygenflow#84): the /// jobs each runner of the wave that took it must finish, or the hours /// since its first job there after which one job is enough. pub const SETTING_SOAK_JOBS: &str = "runner_updates_soak_jobs"; pub const SETTING_SOAK_HOURS: &str = "runner_updates_soak_hours"; pub const SOAK_JOBS: i64 = 5; pub const SOAK_HOURS: i64 = 24; /// Share an offered build with the swarm (oxygenflow#84): cut the release /// asset into the hub's image chunks and keep its manifest, named, as an /// image is, by the build's BLAKE3, so a runner in a swarm fetches it from /// its site's peers and the hub together, as it fetches images, and the /// hub serves it about once per site. Done once, in the background, the /// first time the build is offered; a runner asking before it is done /// fetches the build over HTTP. pub fn share(app: &crate::Shared, offer: &UpdateOffer) { static SHARING: std::sync::Mutex> = std::sync::Mutex::new(BTreeSet::new()); let store = app.images(); let hash = offer.blake3.clone(); if !jet::images::is_hash(&hash) || store.manifest(&hash).is_some() { return; } if !SHARING.lock().unwrap_or_else(|p| p.into_inner()).insert(hash.clone()) { return; } let (app, offer) = (app.clone(), offer.clone()); tokio::spawn(async move { let path = crate::web::releases::blob_path(&app, &hash); let arch = offer.asset.rsplit('-').next().map(|a| a.trim_end_matches(".exe").to_string()); let want = hash.clone(); let made = tokio::task::spawn_blocking(move || -> Result { let m = jet::images::scan_file(&path, arch, |c, bytes| Ok(store.put_chunk(c.hash(), bytes)?)) .map_err(|e| format!("reading {}: {e}", path.display()))?; if m.hash != want { return Err(format!("the asset hashes to {}, not {}", &m.hash[..12], &want[..12])); } store.verify_and_keep(&m)?; Ok(m) }) .await .unwrap_or_else(|e| Err(e.to_string())); match made { Ok(m) => match crate::images::record(&app.db, &m, &format!("runner build flow {}", offer.version)).await { Ok(()) => tracing::info!( "runner updates: flow {} ({}) shared with the swarm: {} chunks", offer.version, offer.asset, m.chunks.len() ), Err(e) => tracing::warn!("runner updates: recording flow {} for the swarm: {e}", offer.version), }, Err(e) => tracing::warn!("runner updates: sharing flow {} ({}) with the swarm: {e}", offer.version, offer.asset), } SHARING.lock().unwrap_or_else(|p| p.into_inner()).remove(&hash); }); } /// A runner's place, 0 to 99, for percentage groups: a stable hash of its /// id, so a group's share is the same machines every rollout, and a /// smaller share sits inside a larger one. pub fn bucket(runner_id: i64) -> i64 { let h = blake3::hash(format!("runner {runner_id}").as_bytes()); let mut first = [0u8; 8]; first.copy_from_slice(&h.as_bytes()[..8]); (u64::from_le_bytes(first) % 100) as i64 } /// Whether a group may hold the runner at all: a site group holds any, an /// owner's group only that owner's runners. pub fn in_reach(g: &Group, r: &Runner) -> bool { g.owner.is_empty() || g.owner == r.scope_owner } /// Every level's settings and the groups, read once. #[derive(Clone, Debug, Default)] pub struct Levels { pub policies: HashMap<(String, String), Policy>, pub groups: Vec, /// Chosen members, as (group, runner). pub chosen: BTreeSet<(i64, i64)>, } impl Levels { pub async fn load(pool: &Pool) -> sqlx::Result { let policies = rdb::policies(pool).await?.into_iter().map(|p| ((p.scope.clone(), p.key.clone()), p)).collect(); Ok(Levels { policies, groups: rdb::groups(pool).await?, chosen: rdb::members(pool).await?.into_iter().collect() }) } pub fn member(&self, g: &Group, r: &Runner) -> bool { in_reach(g, r) && (self.chosen.contains(&(g.id, r.id)) || g.percent.is_some_and(|p| bucket(r.id) < p)) } /// The runner's groups, nearest first: its owner's, then the site's. pub fn groups_of(&self, r: &Runner) -> Vec<&Group> { let mut gs: Vec<&Group> = self.groups.iter().filter(|g| self.member(g, r)).collect(); gs.sort_by_key(|g| (g.owner.is_empty(), g.name.clone())); gs } fn get(&self, scope: &str, key: &str) -> Option<&Policy> { self.policies.get(&(scope.to_string(), key.to_string())) } /// The wave `r` takes a version in: the first of its groups' waves; /// `None` after every wave. pub fn wave_of(&self, r: &Runner) -> Option { self.groups_of(r).iter().filter_map(|g| g.wave).min() } /// The waves before `r`'s, in order, each with its runners among `all`: /// the waves of the groups that may hold `r` (the site's, and its /// owner's), a runner counted in its own first wave. pub fn waves_before<'a>(&self, r: &Runner, all: &'a [Runner]) -> Vec<(i64, Vec<&'a Runner>)> { let mine = self.wave_of(r); let mut waves: Vec = self.groups.iter().filter(|g| in_reach(g, r)).filter_map(|g| g.wave).filter(|w| mine.is_none_or(|m| *w < m)).collect(); waves.sort(); waves.dedup(); waves .into_iter() .map(|w| { let members = all .iter() .filter(|o| self.wave_of(o) == Some(w) && self.groups_of(o).iter().any(|g| g.wave == Some(w) && in_reach(g, r))) .collect(); (w, members) }) .collect() } /// The levels that apply to `r`, nearest first, each with its name. fn chain(&self, r: &Runner) -> Vec<(String, &Policy)> { let mut out = vec![]; if let Some(p) = self.get("runner", &r.id.to_string()) { out.push(("runner".to_string(), p)); } for g in self.groups_of(r) { if let Some(p) = self.get("group", &g.id.to_string()) { out.push((format!("group {}", g.name), p)); } } if !r.scope_owner.is_empty() { if let Some(p) = self.get("owner", &r.scope_owner) { out.push((format!("owner {}", r.scope_owner), p)); } } if let Some(p) = self.get("site", "") { out.push(("site".to_string(), p)); } out } /// The policy for `r`: each setting from the nearest level that sets it. pub fn resolve(&self, r: &Runner) -> UpdatePolicy { let chain = self.chain(r); fn pick(chain: &[(String, &Policy)], f: impl Fn(&Policy) -> Option) -> (Option, String) { chain.iter().find_map(|(name, p)| f(p).map(|v| (Some(v), name.clone()))).unwrap_or((None, "default".into())) } let (updates, u_from) = pick(&chain, |p| p.updates); let (channel, c_from) = pick(&chain, |p| p.channel.clone().filter(|c| !c.is_empty())); let (window, w_from) = pick(&chain, |p| Some((p.window_from? as u8, p.window_to? as u8))); let (dry_run, d_from) = pick(&chain, |p| p.dry_run); UpdatePolicy { updates: updates.unwrap_or(false), channel: channel.unwrap_or_else(|| "stable".into()), window, dry_run: dry_run.unwrap_or(false), from: vec![u_from, c_from, w_from, d_from], } } } /// Whether hour `h` (UTC) falls in `[from, to)`, which may wrap past midnight. pub fn in_window((from, to): (u8, u8), h: u8) -> bool { if from <= to { from <= h && h < to } else { h >= from || h < to } } /// Parse `02-05` (hours, UTC) as a window; blank is none. pub fn parse_window(s: &str) -> Result, String> { let s = s.trim(); if s.is_empty() { return Ok(None); } let (a, b) = s.split_once('-').ok_or("a window is two hours, from-to, e.g. 02-05")?; let hour = |x: &str| x.trim().trim_end_matches(":00").parse::().ok().filter(|h| *h < 24); match (hour(a), hour(b)) { (Some(a), Some(b)) if a != b => Ok(Some((a, b))), _ => Err("a window is two different hours from 0 to 23, from-to, e.g. 02-05".into()), } } /// A channel as written: `stable`, `canary`, or a version. pub fn valid_channel(c: &str) -> bool { matches!(c, "stable" | "canary") || (!c.is_empty() && c.split('.').all(|p| !p.is_empty() && p.bytes().all(|b| b.is_ascii_digit()))) } /// The version in a runner's or a tag's words: `0.3.1 (build)` → `0.3.1`, /// `v0.3.1` → `0.3.1`. pub fn version_of(s: &str) -> String { s.split_whitespace().next().unwrap_or("").trim_start_matches('v').to_string() } fn older(a: &str, b: &str) -> bool { let parts = |v: &str| v.split('.').map(|p| p.parse::().unwrap_or(0)).collect::>(); parts(a) < parts(b) } /// The version a runner's report says rolled back: `rolled back from flow /// 0.4.0 to 0.3.1: …` (the runner's words, oxygenflow spec §11.5.3) → `0.4.0`. pub fn rolled_back_version(note: &str) -> Option { let v = note.strip_prefix("rolled back from flow ")?.split_whitespace().next()?; v.split('.').all(|p| !p.is_empty() && p.bytes().all(|b| b.is_ascii_digit())).then(|| v.to_string()) } /// The release asset a runner on `os` and `arch` runs: `flow-linux-x86_64`, /// `flow-windows-x86_64.exe`. pub fn asset_for(os: &str, arch: &str) -> String { format!("flow-{os}-{arch}{}", if os == "windows" { ".exe" } else { "" }) } /// Where a wave stands with a version (oxygenflow#84). #[derive(Clone, Debug, PartialEq, Eq)] pub enum WaveState { /// Every runner of the wave that took it has finished its soak jobs on /// it, or the soak time has passed since its first job there; or the /// wave has no runners. Proved, /// Not yet: `on` of `runners` run it, with `jobs` jobs between them. Soaking { runners: usize, on: usize, jobs: i64 }, /// A job on it ended in error on this runner: the version stops here. Stopped { runner_id: i64 }, } /// Where a wave of `members` stands with a version, from each runner's /// [`Soak`] on it, at `now` (ms). pub fn wave_state(members: &[i64], soaks: &[Soak], soak_jobs: i64, soak_ms: i64, now: i64) -> WaveState { let mine: Vec<&Soak> = soaks.iter().filter(|s| members.contains(&s.runner_id) && s.jobs > 0).collect(); if let Some(s) = mine.iter().find(|s| s.errors > 0) { return WaveState::Stopped { runner_id: s.runner_id }; } if members.is_empty() { return WaveState::Proved; } let jobs = mine.iter().map(|s| s.jobs).sum(); let soaked = mine.iter().filter_map(|s| s.first_ms).min().is_some_and(|first| now - first >= soak_ms); if !mine.is_empty() && (soaked || mine.iter().all(|s| s.jobs >= soak_jobs)) { WaveState::Proved } else { WaveState::Soaking { runners: members.len(), on: mine.len(), jobs } } } /// What the waves before `r`'s say about `version`: `Ok(None)` go ahead; /// `Ok(Some(why))` wait; `Err((runner, why))` it stopped on that runner. async fn wave_gate( pool: &Pool, levels: &Levels, r: &Runner, version: &str, now: i64, ) -> sqlx::Result, (String, String)>> { let all = db::runners::list(pool).await?; let waves = levels.waves_before(r, &all); if waves.is_empty() { return Ok(Ok(None)); } let setting = |v: Option, d: i64| v.and_then(|s| s.trim().parse::().ok()).filter(|n| *n > 0).unwrap_or(d); let soak_jobs = setting(db::accounts::setting(pool, SETTING_SOAK_JOBS).await?, SOAK_JOBS); let soak_hours = setting(db::accounts::setting(pool, SETTING_SOAK_HOURS).await?, SOAK_HOURS); let soaks = rdb::soaks(pool, version).await?; for (w, members) in waves { let ids: Vec = members.iter().map(|m| m.id).collect(); match wave_state(&ids, &soaks, soak_jobs, soak_hours * 3_600_000, now) { WaveState::Proved => continue, WaveState::Stopped { runner_id } => { let name = members.iter().find(|m| m.id == runner_id).map(|m| m.name.clone()).unwrap_or_default(); let job = rdb::first_error(pool, version, runner_id).await?.unwrap_or_else(|| "a job".into()); return Ok(Err((name.clone(), format!("wave {w} stopped it on {name}: {job} ended in error")))); } WaveState::Soaking { runners, on, jobs } => { return Ok(Ok(Some(format!( "flow {version} waits for wave {w} to prove it: {on} of {runners} runner{} on it, {jobs} job{} (each needs {soak_jobs} without an error, or {soak_hours} h)", if runners == 1 { "" } else { "s" }, if jobs == 1 { "" } else { "s" }, )))); } } } Ok(Ok(None)) } /// The build `r` should install under `policy` at hour `hour` (UTC), or /// why there is none. A version a wave stopped is held back from every /// runner when `act` (a runner asking; a page only says so). pub async fn offer( pool: &Pool, levels: &Levels, r: &Runner, policy: &UpdatePolicy, hour: u8, now: i64, act: bool, ) -> sqlx::Result<(Option, String)> { if !policy.updates { return Ok((None, format!("updates are off ({})", policy.from[0]))); } // A phone's runner is part of the OxygenHub app (oxygenflow#84, // oxygenhub-android#7): it updates when the app does, never from a // runner build. if r.os == "android" { return Ok(( None, format!("follows the OxygenHub app's releases: runs flow {} (update the app to update it)", version_of(&r.version)), )); } let Some(spec) = db::accounts::setting(pool, SETTING_REPO).await?.filter(|s| !s.trim().is_empty()) else { return Ok((None, "no runner releases repository is set (Admin › Runners › Updates)".into())); }; let Some((owner, name)) = spec.trim().split_once('/') else { return Ok((None, format!("the runner releases repository {spec:?} is not owner/name"))); }; let Some(repo) = db::repos::get(pool, owner, name).await? else { return Ok((None, format!("the runner releases repository {spec} does not exist"))); }; let releases = db::releases::list(pool, repo.id).await?; let release = match policy.channel.as_str() { "stable" => releases.iter().find(|x| !x.prerelease), "canary" => releases.first(), v => releases.iter().find(|x| version_of(&x.tag) == v), }; let Some(release) = release else { return Ok((None, format!("{spec} has no release for the {} channel", policy.channel))); }; let (want, have) = (version_of(&release.tag), version_of(&r.version)); if want == have { return Ok((None, format!("up to date: flow {have} ({} channel)", policy.channel))); } if let Some(h) = rdb::held(pool, &want).await? { return Ok((None, format!("flow {want} is held: {}; lift the hold on Admin › Runners › Updates", h.why))); } let going_back = older(&want, &have); if going_back && matches!(policy.channel.as_str(), "stable" | "canary") { return Ok((None, format!("runs flow {have}, newer than the {} channel's {want}", policy.channel))); } // Waves (oxygenflow#84): a later wave takes a version once those before it proved it. if !going_back { match wave_gate(pool, levels, r, &want, now).await? { Ok(None) => {} Ok(Some(why)) => return Ok((None, why)), Err((runner, why)) => { if act { rdb::hold(pool, &want, &runner, &why).await?; db::accounts::audit(pool, None, "runner.update.held", &want, &why).await.ok(); } return Ok((None, format!("flow {want} is held: {why}; lift the hold on Admin › Runners › Updates"))); } } } let file = asset_for(&r.os, &r.arch); let assets = db::releases::assets(pool, release.id).await?; let Some(a) = assets.iter().find(|a| a.name == file) else { return Ok((None, format!("{spec} {} has no {file}", release.tag))); }; if let Some(w) = policy.window.filter(|w| !in_window(*w, hour)) { return Ok((None, format!("flow {want} waits for its window, {:02}:00–{:02}:00 UTC", w.0, w.1))); } let note = format!( "flow {want} ({} channel){}", policy.channel, if policy.dry_run { ": dry run, check it and say what would happen" } else { "" } ); Ok(( Some(UpdateOffer { version: want, tag: release.tag.clone(), asset: a.name.clone(), url: format!("/{owner}/{name}/asset/{}/{}", a.id, a.name), blake3: a.hash.clone(), size: a.size as u64, older: going_back, }), note, )) } #[cfg(test)] mod tests { use super::*; fn runner(id: i64, owner: &str) -> Runner { Runner { id, scope_owner: owner.into(), version: "0.3.1 (build)".into(), ..Runner::default() } } #[test] fn a_smaller_share_sits_inside_a_larger_one_and_is_about_its_size() { let ten: Vec = (1..=1000).filter(|id| bucket(*id) < 10).collect(); let fifty: Vec = (1..=1000).filter(|id| bucket(*id) < 50).collect(); assert!(ten.iter().all(|id| fifty.contains(id))); assert!((60..=140).contains(&ten.len()), "{}", ten.len()); assert_eq!(bucket(7), bucket(7), "stable"); } #[test] fn the_nearest_level_wins_each_setting() { let mut l = Levels::default(); let pol = |scope: &str, key: &str, f: &dyn Fn(&mut Policy)| { let mut p = Policy { scope: scope.into(), key: key.into(), ..Policy::default() }; f(&mut p); ((scope.to_string(), key.to_string()), p) }; l.policies.extend([ pol("site", "", &|p| { p.updates = Some(false); p.channel = Some("stable".into()); }), pol("owner", "acme", &|p| p.updates = Some(true)), pol("group", "1", &|p| p.channel = Some("canary".into())), pol("runner", "3", &|p| p.updates = Some(false)), ]); l.groups.push(Group { id: 1, owner: "acme".into(), name: "canary".into(), percent: None, created_ms: 0, wave: None }); l.chosen.insert((1, 2)); l.chosen.insert((1, 9)); let site_runner = l.resolve(&runner(1, "")); assert!(!site_runner.updates, "off at the site"); assert_eq!(site_runner.from, ["site", "site", "default", "default"]); let canary = l.resolve(&runner(2, "acme")); assert!(canary.updates && canary.channel == "canary"); assert_eq!(canary.from[..2], ["owner acme".to_string(), "group canary".to_string()]); let held = l.resolve(&runner(3, "acme")); assert!(!held.updates, "the runner's own setting wins"); assert_eq!(held.channel, "stable"); assert!(l.groups_of(&runner(9, "other")).is_empty(), "an owner's group holds only its own runners"); assert!(!Levels::default().resolve(&runner(4, "")).updates, "off unless chosen"); } #[test] fn a_percentage_group_takes_its_share_of_the_runners_in_reach() { let mut l = Levels::default(); l.groups.push(Group { id: 5, owner: String::new(), name: "farm 20%".into(), percent: Some(20), created_ms: 0, wave: None }); let n = (1..=500).filter(|id| !l.groups_of(&runner(*id, "")).is_empty()).count(); assert!((60..=140).contains(&n), "{n}"); } #[test] fn a_wave_proves_a_version_by_its_jobs_or_its_time_and_an_error_stops_it() { let soak = |runner_id, jobs, errors, first_ms| Soak { runner_id, jobs, errors, first_ms: Some(first_ms) }; let hour = 3_600_000; // Nobody has taken it yet: not proved. assert_eq!(wave_state(&[1, 2], &[], 5, 24 * hour, 0), WaveState::Soaking { runners: 2, on: 0, jobs: 0 }); // One runner took it and ran 3 of its 5; the other has not updated. let s = [soak(1, 3, 0, 0), soak(9, 50, 0, 0)]; assert_eq!(wave_state(&[1, 2], &s, 5, 24 * hour, hour), WaveState::Soaking { runners: 2, on: 1, jobs: 3 }); // Five each on every runner that took it proves it. assert_eq!(wave_state(&[1, 2], &[soak(1, 5, 0, 0)], 5, 24 * hour, hour), WaveState::Proved); // Or a day since the first job there, with no error. assert_eq!(wave_state(&[1, 2], &[soak(1, 1, 0, 0)], 5, 24 * hour, 24 * hour), WaveState::Proved); // An error anywhere in the wave stops it, however many passed. assert_eq!( wave_state(&[1, 2], &[soak(1, 40, 0, 0), soak(2, 2, 1, 0)], 5, 24 * hour, 48 * hour), WaveState::Stopped { runner_id: 2 } ); // Another wave's errors are not this one's. assert_eq!(wave_state(&[1], &[soak(1, 5, 0, 0), soak(2, 2, 1, 0)], 5, 24 * hour, hour), WaveState::Proved); // An empty wave holds nothing up. assert_eq!(wave_state(&[], &[], 5, 24 * hour, 0), WaveState::Proved); } #[test] fn a_runner_waits_for_the_waves_before_its_own() { let mut l = Levels::default(); let group = |id, name: &str, wave| Group { id, owner: String::new(), name: name.into(), percent: None, created_ms: 0, wave }; l.groups.extend([group(1, "first", Some(1)), group(2, "second", Some(2)), group(3, "watchers", None)]); l.chosen.extend([(1, 1), (2, 2), (2, 1), (3, 3)]); let all: Vec = (1..=4).map(|id| runner(id, "")).collect(); let ids = |ws: Vec<(i64, Vec<&Runner>)>| { ws.into_iter().map(|(w, rs)| (w, rs.iter().map(|r| r.id).collect::>())).collect::>() }; assert_eq!(l.wave_of(&all[0]), Some(1), "its first wave"); assert!(l.waves_before(&all[0], &all).is_empty(), "wave 1 goes first"); assert_eq!(ids(l.waves_before(&all[1], &all)), vec![(1, vec![1])]); assert_eq!(l.wave_of(&all[2]), None, "a group with no wave"); assert_eq!(ids(l.waves_before(&all[2], &all)), vec![(1, vec![1]), (2, vec![2])], "in no wave: after every wave"); assert_eq!(ids(l.waves_before(&all[3], &all)), vec![(1, vec![1]), (2, vec![2])]); // An owner's waves gate only that owner's runners. l.groups.push(Group { id: 4, owner: "acme".into(), name: "acme first".into(), percent: None, created_ms: 0, wave: Some(1) }); l.chosen.insert((4, 5)); let mut all = all; all.push(runner(5, "acme")); all.push(runner(6, "acme")); assert_eq!(ids(l.waves_before(&all[5], &all)), vec![(1, vec![1, 5]), (2, vec![2])]); assert_eq!(ids(l.waves_before(&all[3], &all)), vec![(1, vec![1]), (2, vec![2])], "not another owner's"); } #[test] fn windows_and_channels_read_as_written() { assert_eq!(parse_window("02-05"), Ok(Some((2, 5)))); assert_eq!(parse_window(" 22:00-04:00 "), Ok(Some((22, 4)))); assert_eq!(parse_window(""), Ok(None)); assert!(parse_window("5").is_err() && parse_window("3-3").is_err() && parse_window("1-24").is_err()); assert!(in_window((2, 5), 2) && in_window((2, 5), 4) && !in_window((2, 5), 5)); assert!(in_window((22, 4), 23) && in_window((22, 4), 1) && !in_window((22, 4), 12)); assert!(valid_channel("stable") && valid_channel("canary") && valid_channel("0.3.4")); assert!(!valid_channel("latest") && !valid_channel("0.3.") && !valid_channel("")); assert_eq!(version_of("v0.3.2"), "0.3.2"); assert_eq!(version_of("0.3.1 (local build)"), "0.3.1"); assert_eq!(asset_for("linux", "riscv64"), "flow-linux-riscv64"); assert_eq!(asset_for("windows", "x86_64"), "flow-windows-x86_64.exe"); assert_eq!( rolled_back_version("rolled back from flow 0.4.0 to 0.3.1: 3 starts without reaching the hub").as_deref(), Some("0.4.0") ); assert_eq!(rolled_back_version("runs flow 0.4.0 (was 0.3.1); proven"), None); } }