OXYGENHUB

Coverage / crates/hub/src/runner_updates.rs 96.8%

214 of 221 lines run, 91 of 108 functions · run #469 on main · run missed

1
//! Runner updates (oxygenhub#322, OxygenFlow docs/jetstream.md §8.2):
2
//! which build each runner should run, under the policy the site, its
3
//! owner, its test groups and the runner itself set, the nearest setting
4
//! winning. Off unless some level turns updates on.
5
6
use crate::db::runner_updates::{self as rdb, Group, Policy};
7
use crate::db::runners::Runner;
8
use crate::db::{self, Pool};
9
use flow_core::api::{UpdateOffer, UpdatePolicy};
10
use std::collections::{BTreeSet, HashMap};
11
12
/// The site setting naming the repository whose releases carry runner
13
/// builds (`owner/name`), each an asset `flow-OS-ARCH`.
14
pub const SETTING_REPO: &str = "runner_updates_repo";
15
16
/// A runner's place, 0 to 99, for percentage groups: a stable hash of its
17
/// id, so a group's share is the same machines every rollout, and a
18
/// smaller share sits inside a larger one.
192509×
pub fn bucket(runner_id: i64) -> i64 {
202509×
    let h = blake3::hash(format!("runner {runner_id}").as_bytes());
212509×
    let mut first = [0u8; 8];
222509×
    first.copy_from_slice(&h.as_bytes()[..8]);
232509×
    (u64::from_le_bytes(first) % 100) as i64
242509×
}
25
26
/// Whether a group may hold the runner at all: a site group holds any, an
27
/// owner's group only that owner's runners.
28570×
pub fn in_reach(g: &Group, r: &Runner) -> bool {
29570×
    g.owner.is_empty() || g.owner == r.scope_owner
30570×
}
31
32
/// Every level's settings and the groups, read once.
33
#[derive(Clone, Debug, Default)]
34
pub struct Levels {
35
    pub policies: HashMap<(String, String), Policy>,
36
    pub groups: Vec<Group>,
37
    /// Chosen members, as (group, runner).
38
    pub chosen: BTreeSet<(i64, i64)>,
39
}
40
41
impl Levels {
4222×
    pub async fn load(pool: &Pool) -> sqlx::Result<Levels> {
4341×
        let policies = rdb::policies(pool).await?.into_iter().map(|p| ((p.scope.clone(), p.key.clone()), p)).collect();
4422×
        Ok(Levels { policies, groups: rdb::groups(pool).await?, chosen: rdb::members(pool).await?.into_iter().collect() })
4522×
    }
46
47570×
    pub fn member(&self, g: &Group, r: &Runner) -> bool {
48570×
        in_reach(g, r) && (self.chosen.contains(&(g.id, r.id)) || g.percent.is_some_and(|p| bucket(r.id) < p))
49570×
    }
50
51
    /// The runner's groups, nearest first: its owner's, then the site's.
52565×
    pub fn groups_of(&self, r: &Runner) -> Vec<&Group> {
53565×
        let mut gs: Vec<&Group> = self.groups.iter().filter(|g| self.member(g, r)).collect();
54565×
        gs.sort_by_key(|g| (g.owner.is_empty(), g.name.clone()));
55565×
        gs
56565×
    }
57
5894×
    fn get(&self, scope: &str, key: &str) -> Option<&Policy> {
5994×
        self.policies.get(&(scope.to_string(), key.to_string()))
6094×
    }
61
62
    /// The levels that apply to `r`, nearest first, each with its name.
6338×
    fn chain(&self, r: &Runner) -> Vec<(String, &Policy)> {
6438×
        let mut out = vec![];
6538×
        if let Some(p) = self.get("runner", &r.id.to_string()) {
6611×
            out.push(("runner".to_string(), p));
6727×
        }
6838×
        for g in self.groups_of(r) {
6916×
            if let Some(p) = self.get("group", &g.id.to_string()) {
7015×
                out.push((format!("group {}", g.name), p));
7115×
            }
72
        }
7338×
        if !r.scope_owner.is_empty() {
742×
            if let Some(p) = self.get("owner", &r.scope_owner) {
752×
                out.push((format!("owner {}", r.scope_owner), p));
762×
            }
7736×
        }
7838×
        if let Some(p) = self.get("site", "") {
7926×
            out.push(("site".to_string(), p));
8026×
        }
8138×
        out
8238×
    }
83
84
    /// The policy for `r`: each setting from the nearest level that sets it.
8538×
    pub fn resolve(&self, r: &Runner) -> UpdatePolicy {
8638×
        let chain = self.chain(r);
87152×
        fn pick<T>(chain: &[(String, &Policy)], f: impl Fn(&Policy) -> Option<T>) -> (Option<T>, String) {
88168×
            chain.iter().find_map(|(name, p)| f(p).map(|v| (Some(v), name.clone()))).unwrap_or((None, "default".into()))
89152×
        }
9038×
        let (updates, u_from) = pick(&chain, |p| p.updates);
9140×
        let (channel, c_from) = pick(&chain, |p| p.channel.clone().filter(|c| !c.is_empty()));
9245×
        let (window, w_from) = pick(&chain, |p| Some((p.window_from? as u8, p.window_to? as u8)));
9338×
        let (dry_run, d_from) = pick(&chain, |p| p.dry_run);
94
        UpdatePolicy {
9538×
            updates: updates.unwrap_or(false),
9638×
            channel: channel.unwrap_or_else(|| "stable".into()),
9738×
            window,
9838×
            dry_run: dry_run.unwrap_or(false),
9938×
            from: vec![u_from, c_from, w_from, d_from],
100
        }
10138×
    }
102
}
103
104
/// Whether hour `h` (UTC) falls in `[from, to)`, which may wrap past midnight.
10513×
pub fn in_window((from, to): (u8, u8), h: u8) -> bool {
10613×
    if from <= to {
10710×
        from <= h && h < to
108
    } else {
1093×
        h >= from || h < to
110
    }
11113×
}
112
113
/// Parse `02-05` (hours, UTC) as a window; blank is none.
11411×
pub fn parse_window(s: &str) -> Result<Option<(u8, u8)>, String> {
11511×
    let s = s.trim();
11611×
    if s.is_empty() {
1174×
        return Ok(None);
1187×
    }
1197×
    let (a, b) = s.split_once('-').ok_or("a window is two hours, from-to, e.g. 02-05")?;
12010×
    let hour = |x: &str| x.trim().trim_end_matches(":00").parse::<u8>().ok().filter(|h| *h < 24);
1215×
    match (hour(a), hour(b)) {
1224×
        (Some(a), Some(b)) if a != b => Ok(Some((a, b))),
1232×
        _ => Err("a window is two different hours from 0 to 23, from-to, e.g. 02-05".into()),
124
    }
12511×
}
126
127
/// A channel as written: `stable`, `canary`, or a version.
1288×
pub fn valid_channel(c: &str) -> bool {
12911×
    matches!(c, "stable" | "canary") || (!c.is_empty() && c.split('.').all(|p| !p.is_empty() && p.bytes().all(|b| b.is_ascii_digit())))
1308×
}
131
132
/// The version in a runner's or a tag's words: `0.3.1 (build)` → `0.3.1`,
133
/// `v0.3.1` → `0.3.1`.
13436×
pub fn version_of(s: &str) -> String {
13536×
    s.split_whitespace().next().unwrap_or("").trim_start_matches('v').to_string()
13636×
}
137
13812×
fn older(a: &str, b: &str) -> bool {
13972×
    let parts = |v: &str| v.split('.').map(|p| p.parse::<u64>().unwrap_or(0)).collect::<Vec<_>>();
14012×
    parts(a) < parts(b)
14112×
}
142
143
/// The build `r` should install under `policy` at hour `hour` (UTC), or
144
/// why there is none.
14534×
pub async fn offer(pool: &Pool, r: &Runner, policy: &UpdatePolicy, hour: u8) -> sqlx::Result<(Option<UpdateOffer>, String)> {
14634×
    if !policy.updates {
14712×
        return Ok((None, format!("updates are off ({})", policy.from[0])));
14822×
    }
14922×
    let Some(spec) = db::accounts::setting(pool, SETTING_REPO).await?.filter(|s| !s.trim().is_empty()) else {
1500×
        return Ok((None, "no runner releases repository is set (Admin › Runners › Updates)".into()));
151
    };
15222×
    let Some((owner, name)) = spec.trim().split_once('/') else {
1530×
        return Ok((None, format!("the runner releases repository {spec:?} is not owner/name")));
154
    };
15522×
    let Some(repo) = db::repos::get(pool, owner, name).await? else {
1560×
        return Ok((None, format!("the runner releases repository {spec} does not exist")));
157
    };
15822×
    let releases = db::releases::list(pool, repo.id).await?;
15922×
    let release = match policy.channel.as_str() {
16022×
        "stable" => releases.iter().find(|x| !x.prerelease),
16110×
        "canary" => releases.first(),
16210×
        v => releases.iter().find(|x| version_of(&x.tag) == v),
163
    };
16422×
    let Some(release) = release else {
16510×
        return Ok((None, format!("{spec} has no release for the {} channel", policy.channel)));
166
    };
16712×
    let (want, have) = (version_of(&release.tag), version_of(&r.version));
16812×
    if want == have {
1690×
        return Ok((None, format!("up to date: flow {have} ({} channel)", policy.channel)));
17012×
    }
17112×
    let going_back = older(&want, &have);
17212×
    if going_back && matches!(policy.channel.as_str(), "stable" | "canary") {
1730×
        return Ok((None, format!("runs flow {have}, newer than the {} channel's {want}", policy.channel)));
17412×
    }
17512×
    let file = format!("flow-{}-{}", r.os, r.arch);
17612×
    let assets = db::releases::assets(pool, release.id).await?;
17712×
    let Some(a) = assets.iter().find(|a| a.name == file) else {
1780×
        return Ok((None, format!("{spec} {} has no {file}", release.tag)));
179
    };
18012×
    if let Some(w) = policy.window.filter(|w| !in_window(*w, hour)) {
1817×
        return Ok((None, format!("flow {want} waits for its window, {:02}:00–{:02}:00 UTC", w.0, w.1)));
1825×
    }
1835×
    let note = format!(
184
        "flow {want} ({} channel){}",
185
        policy.channel,
1865×
        if policy.dry_run { ": dry run, check it and say what would happen" } else { "" }
187
    );
1885×
    Ok((
1895×
        Some(UpdateOffer {
1905×
            version: want,
1915×
            tag: release.tag.clone(),
1925×
            asset: a.name.clone(),
1935×
            url: format!("/{owner}/{name}/asset/{}/{}", a.id, a.name),
1945×
            blake3: a.hash.clone(),
1955×
            size: a.size as u64,
1965×
            older: going_back,
1975×
        }),
1985×
        note,
1995×
    ))
20034×
}
201
202
#[cfg(test)]
203
mod tests {
204
    use super::*;
205
206505×
    fn runner(id: i64, owner: &str) -> Runner {
207505×
        Runner { id, scope_owner: owner.into(), version: "0.3.1 (build)".into(), ..Runner::default() }
208505×
    }
209
210
    #[test]
2111×
    fn a_smaller_share_sits_inside_a_larger_one_and_is_about_its_size() {
2121000×
        let ten: Vec<i64> = (1..=1000).filter(|id| bucket(*id) < 10).collect();
2131000×
        let fifty: Vec<i64> = (1..=1000).filter(|id| bucket(*id) < 50).collect();
21496×
        assert!(ten.iter().all(|id| fifty.contains(id)));
2151×
        assert!((60..=140).contains(&ten.len()), "{}", ten.len());
2161×
        assert_eq!(bucket(7), bucket(7), "stable");
2171×
    }
218
219
    #[test]
2201×
    fn the_nearest_level_wins_each_setting() {
2211×
        let mut l = Levels::default();
2224×
        let pol = |scope: &str, key: &str, f: &dyn Fn(&mut Policy)| {
2234×
            let mut p = Policy { scope: scope.into(), key: key.into(), ..Policy::default() };
2244×
            f(&mut p);
2254×
            ((scope.to_string(), key.to_string()), p)
2264×
        };
2271×
        l.policies.extend([
2281×
            pol("site", "", &|p| {
2291×
                p.updates = Some(false);
2301×
                p.channel = Some("stable".into());
2311×
            }),
2321×
            pol("owner", "acme", &|p| p.updates = Some(true)),
2331×
            pol("group", "1", &|p| p.channel = Some("canary".into())),
2341×
            pol("runner", "3", &|p| p.updates = Some(false)),
235
        ]);
2361×
        l.groups.push(Group { id: 1, owner: "acme".into(), name: "canary".into(), percent: None, created_ms: 0 });
2371×
        l.chosen.insert((1, 2));
2381×
        l.chosen.insert((1, 9));
239
2401×
        let site_runner = l.resolve(&runner(1, ""));
2411×
        assert!(!site_runner.updates, "off at the site");
2421×
        assert_eq!(site_runner.from, ["site", "site", "default", "default"]);
243
2441×
        let canary = l.resolve(&runner(2, "acme"));
2451×
        assert!(canary.updates && canary.channel == "canary");
2461×
        assert_eq!(canary.from[..2], ["owner acme".to_string(), "group canary".to_string()]);
247
2481×
        let held = l.resolve(&runner(3, "acme"));
2491×
        assert!(!held.updates, "the runner's own setting wins");
2501×
        assert_eq!(held.channel, "stable");
251
2521×
        assert!(l.groups_of(&runner(9, "other")).is_empty(), "an owner's group holds only its own runners");
2531×
        assert!(!Levels::default().resolve(&runner(4, "")).updates, "off unless chosen");
2541×
    }
255
256
    #[test]
2571×
    fn a_percentage_group_takes_its_share_of_the_runners_in_reach() {
2581×
        let mut l = Levels::default();
2591×
        l.groups.push(Group { id: 5, owner: String::new(), name: "farm 20%".into(), percent: Some(20), created_ms: 0 });
260500×
        let n = (1..=500).filter(|id| !l.groups_of(&runner(*id, "")).is_empty()).count();
2611×
        assert!((60..=140).contains(&n), "{n}");
2621×
    }
263
264
    #[test]
2651×
    fn windows_and_channels_read_as_written() {
2661×
        assert_eq!(parse_window("02-05"), Ok(Some((2, 5))));
2671×
        assert_eq!(parse_window(" 22:00-04:00 "), Ok(Some((22, 4))));
2681×
        assert_eq!(parse_window(""), Ok(None));
2691×
        assert!(parse_window("5").is_err() && parse_window("3-3").is_err() && parse_window("1-24").is_err());
2701×
        assert!(in_window((2, 5), 2) && in_window((2, 5), 4) && !in_window((2, 5), 5));
2711×
        assert!(in_window((22, 4), 23) && in_window((22, 4), 1) && !in_window((22, 4), 12));
2721×
        assert!(valid_channel("stable") && valid_channel("canary") && valid_channel("0.3.4"));
2731×
        assert!(!valid_channel("latest") && !valid_channel("0.3.") && !valid_channel(""));
2741×
        assert_eq!(version_of("v0.3.2"), "0.3.2");
2751×
        assert_eq!(version_of("0.3.1 (local build)"), "0.3.1");
2761×
    }
277
}