runner_updates.rs 97.9%

237 of 242 lines run, 102 of 118 functions · run #507 on main · run missed

//! 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};
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";

/// 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<Group>,
    /// Chosen members, as (group, runner).
    pub chosen: BTreeSet<(i64, i64)>,
}

impl Levels {
    pub async fn load(pool: &Pool) -> sqlx::Result<Levels> {
        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 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<T>(chain: &[(String, &Policy)], f: impl Fn(&Policy) -> Option<T>) -> (Option<T>, 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<Option<(u8, u8)>, 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::<u8>().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::<u64>().unwrap_or(0)).collect::<Vec<_>>();
    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<String> {
    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 { "" })
}

/// The build `r` should install under `policy` at hour `hour` (UTC), or
/// why there is none.
pub async fn offer(pool: &Pool, r: &Runner, policy: &UpdatePolicy, hour: u8) -> sqlx::Result<(Option<UpdateOffer>, String)> {
    if !policy.updates {
        return Ok((None, format!("updates are off ({})", policy.from[0])));
    }
    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: it rolled back on {} ({}); lift the hold on Admin › Runners › Updates", h.runner, 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)));
    }
    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<i64> = (1..=1000).filter(|id| bucket(*id) < 10).collect();
        let fifty: Vec<i64> = (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 });
        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 });
        let n = (1..=500).filter(|id| !l.groups_of(&runner(*id, "")).is_empty()).count();
        assert!((60..=140).contains(&n), "{n}");
    }

    #[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);
    }
}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

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.
19	2519×	pub fn bucket(runner_id: i64) -> i64 {
20	2519×	let h = blake3::hash(format!("runner {runner_id}").as_bytes());
21	2519×	let mut first = [0u8; 8];
22	2519×	first.copy_from_slice(&h.as_bytes()[..8]);
23	2519×	(u64::from_le_bytes(first) % 100) as i64
24	2519×	}
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.
28	649×	pub fn in_reach(g: &Group, r: &Runner) -> bool {
29	649×	g.owner.is_empty() \|\| g.owner == r.scope_owner
30	649×	}
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 {
42	62×	pub async fn load(pool: &Pool) -> sqlx::Result<Levels> {
43	84×	let policies = rdb::policies(pool).await?.into_iter().map(\|p\| ((p.scope.clone(), p.key.clone()), p)).collect();
44	62×	Ok(Levels { policies, groups: rdb::groups(pool).await?, chosen: rdb::members(pool).await?.into_iter().collect() })
45	62×	}
46
47	649×	pub fn member(&self, g: &Group, r: &Runner) -> bool {
48	649×	in_reach(g, r) && (self.chosen.contains(&(g.id, r.id)) \|\| g.percent.is_some_and(\|p\| bucket(r.id) < p))
49	649×	}
50
51		/// The runner's groups, nearest first: its owner's, then the site's.
52	629×	pub fn groups_of(&self, r: &Runner) -> Vec<&Group> {
53	629×	let mut gs: Vec<&Group> = self.groups.iter().filter(\|g\| self.member(g, r)).collect();
54	629×	gs.sort_by_key(\|g\| (g.owner.is_empty(), g.name.clone()));
55	629×	gs
56	629×	}
57
58	203×	fn get(&self, scope: &str, key: &str) -> Option<&Policy> {
59	203×	self.policies.get(&(scope.to_string(), key.to_string()))
60	203×	}
61
62		/// The levels that apply to `r`, nearest first, each with its name.
63	72×	fn chain(&self, r: &Runner) -> Vec<(String, &Policy)> {
64	72×	let mut out = vec![];
65	72×	if let Some(p) = self.get("runner", &r.id.to_string()) {
66	20×	out.push(("runner".to_string(), p));
67	52×	}
68	72×	for g in self.groups_of(r) {
69	35×	if let Some(p) = self.get("group", &g.id.to_string()) {
70	27×	out.push((format!("group {}", g.name), p));
71	27×	}
72		}
73	72×	if !r.scope_owner.is_empty() {
74	24×	if let Some(p) = self.get("owner", &r.scope_owner) {
75	22×	out.push((format!("owner {}", r.scope_owner), p));
76	22×	}
77	48×	}
78	72×	if let Some(p) = self.get("site", "") {
79	38×	out.push(("site".to_string(), p));
80	38×	}
81	72×	out
82	72×	}
83
84		/// The policy for `r`: each setting from the nearest level that sets it.
85	72×	pub fn resolve(&self, r: &Runner) -> UpdatePolicy {
86	72×	let chain = self.chain(r);
87	288×	fn pick<T>(chain: &[(String, &Policy)], f: impl Fn(&Policy) -> Option<T>) -> (Option<T>, String) {
88	338×	chain.iter().find_map(\|(name, p)\| f(p).map(\|v\| (Some(v), name.clone()))).unwrap_or((None, "default".into()))
89	288×	}
90	72×	let (updates, u_from) = pick(&chain, \|p\| p.updates);
91	83×	let (channel, c_from) = pick(&chain, \|p\| p.channel.clone().filter(\|c\| !c.is_empty()));
92	91×	let (window, w_from) = pick(&chain, \|p\| Some((p.window_from? as u8, p.window_to? as u8)));
93	72×	let (dry_run, d_from) = pick(&chain, \|p\| p.dry_run);
94		UpdatePolicy {
95	72×	updates: updates.unwrap_or(false),
96	72×	channel: channel.unwrap_or_else(\|\| "stable".into()),
97	72×	window,
98	72×	dry_run: dry_run.unwrap_or(false),
99	72×	from: vec![u_from, c_from, w_from, d_from],
100		}
101	72×	}
102		}
103
104		/// Whether hour `h` (UTC) falls in `[from, to)`, which may wrap past midnight.
105	17×	pub fn in_window((from, to): (u8, u8), h: u8) -> bool {
106	17×	if from <= to {
107	14×	from <= h && h < to
108		} else {
109	3×	h >= from \|\| h < to
110		}
111	17×	}
112
113		/// Parse `02-05` (hours, UTC) as a window; blank is none.
114	15×	pub fn parse_window(s: &str) -> Result<Option<(u8, u8)>, String> {
115	15×	let s = s.trim();
116	15×	if s.is_empty() {
117	8×	return Ok(None);
118	7×	}
119	7×	let (a, b) = s.split_once('-').ok_or("a window is two hours, from-to, e.g. 02-05")?;
120	10×	let hour = \|x: &str\| x.trim().trim_end_matches(":00").parse::<u8>().ok().filter(\|h\| *h < 24);
121	5×	match (hour(a), hour(b)) {
122	4×	(Some(a), Some(b)) if a != b => Ok(Some((a, b))),
123	2×	_ => Err("a window is two different hours from 0 to 23, from-to, e.g. 02-05".into()),
124		}
125	15×	}
126
127		/// A channel as written: `stable`, `canary`, or a version.
128	11×	pub fn valid_channel(c: &str) -> bool {
129	15×	matches!(c, "stable" \| "canary") \|\| (!c.is_empty() && c.split('.').all(\|p\| !p.is_empty() && p.bytes().all(\|b\| b.is_ascii_digit())))
130	11×	}
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`.
134	55×	pub fn version_of(s: &str) -> String {
135	55×	s.split_whitespace().next().unwrap_or("").trim_start_matches('v').to_string()
136	55×	}
137
138	16×	fn older(a: &str, b: &str) -> bool {
139	96×	let parts = \|v: &str\| v.split('.').map(\|p\| p.parse::<u64>().unwrap_or(0)).collect::<Vec<_>>();
140	16×	parts(a) < parts(b)
141	16×	}
142
143		/// The version a runner's report says rolled back: `rolled back from flow
144		/// 0.4.0 to 0.3.1: …` (the runner's words, oxygenflow spec §11.5.3) → `0.4.0`.
145	4×	pub fn rolled_back_version(note: &str) -> Option<String> {
146	4×	let v = note.strip_prefix("rolled back from flow ")?.split_whitespace().next()?;
147	6×	v.split('.').all(\|p\| !p.is_empty() && p.bytes().all(\|b\| b.is_ascii_digit())).then(\|\| v.to_string())
148	4×	}
149
150		/// The release asset a runner on `os` and `arch` runs: `flow-linux-x86_64`,
151		/// `flow-windows-x86_64.exe`.
152	18×	pub fn asset_for(os: &str, arch: &str) -> String {
153	18×	format!("flow-{os}-{arch}{}", if os == "windows" { ".exe" } else { "" })
154	18×	}
155
156		/// The build `r` should install under `policy` at hour `hour` (UTC), or
157		/// why there is none.
158	68×	pub async fn offer(pool: &Pool, r: &Runner, policy: &UpdatePolicy, hour: u8) -> sqlx::Result<(Option<UpdateOffer>, String)> {
159	68×	if !policy.updates {
160	18×	return Ok((None, format!("updates are off ({})", policy.from[0])));
161	50×	}
162	50×	let Some(spec) = db::accounts::setting(pool, SETTING_REPO).await?.filter(\|s\| !s.trim().is_empty()) else {
163	16×	return Ok((None, "no runner releases repository is set (Admin › Runners › Updates)".into()));
164		};
165	34×	let Some((owner, name)) = spec.trim().split_once('/') else {
166	0×	return Ok((None, format!("the runner releases repository {spec:?} is not owner/name")));
167		};
168	34×	let Some(repo) = db::repos::get(pool, owner, name).await? else {
169	0×	return Ok((None, format!("the runner releases repository {spec} does not exist")));
170		};
171	34×	let releases = db::releases::list(pool, repo.id).await?;
172	34×	let release = match policy.channel.as_str() {
173	34×	"stable" => releases.iter().find(\|x\| !x.prerelease),
174	15×	"canary" => releases.first(),
175	15×	v => releases.iter().find(\|x\| version_of(&x.tag) == v),
176		};
177	34×	let Some(release) = release else {
178	15×	return Ok((None, format!("{spec} has no release for the {} channel", policy.channel)));
179		};
180	19×	let (want, have) = (version_of(&release.tag), version_of(&r.version));
181	19×	if want == have {
182	0×	return Ok((None, format!("up to date: flow {have} ({} channel)", policy.channel)));
183	19×	}
184	19×	if let Some(h) = rdb::held(pool, &want).await? {
185	3×	return Ok((
186	3×	None,
187	3×	format!("flow {want} is held: it rolled back on {} ({}); lift the hold on Admin › Runners › Updates", h.runner, h.why),
188	3×	));
189	16×	}
190	16×	let going_back = older(&want, &have);
191	16×	if going_back && matches!(policy.channel.as_str(), "stable" \| "canary") {
192	0×	return Ok((None, format!("runs flow {have}, newer than the {} channel's {want}", policy.channel)));
193	16×	}
194	16×	let file = asset_for(&r.os, &r.arch);
195	16×	let assets = db::releases::assets(pool, release.id).await?;
196	16×	let Some(a) = assets.iter().find(\|a\| a.name == file) else {
197	0×	return Ok((None, format!("{spec} {} has no {file}", release.tag)));
198		};
199	16×	if let Some(w) = policy.window.filter(\|w\| !in_window(*w, hour)) {
200	11×	return Ok((None, format!("flow {want} waits for its window, {:02}:00–{:02}:00 UTC", w.0, w.1)));
201	5×	}
202	5×	let note = format!(
203		"flow {want} ({} channel){}",
204		policy.channel,
205	5×	if policy.dry_run { ": dry run, check it and say what would happen" } else { "" }
206		);
207	5×	Ok((
208	5×	Some(UpdateOffer {
209	5×	version: want,
210	5×	tag: release.tag.clone(),
211	5×	asset: a.name.clone(),
212	5×	url: format!("/{owner}/{name}/asset/{}/{}", a.id, a.name),
213	5×	blake3: a.hash.clone(),
214	5×	size: a.size as u64,
215	5×	older: going_back,
216	5×	}),
217	5×	note,
218	5×	))
219	68×	}
220
221		#[cfg(test)]
222		mod tests {
223		use super::*;
224
225	505×	fn runner(id: i64, owner: &str) -> Runner {
226	505×	Runner { id, scope_owner: owner.into(), version: "0.3.1 (build)".into(), ..Runner::default() }
227	505×	}
228
229		#[test]
230	1×	fn a_smaller_share_sits_inside_a_larger_one_and_is_about_its_size() {
231	1000×	let ten: Vec<i64> = (1..=1000).filter(\|id\| bucket(*id) < 10).collect();
232	1000×	let fifty: Vec<i64> = (1..=1000).filter(\|id\| bucket(*id) < 50).collect();
233	96×	assert!(ten.iter().all(\|id\| fifty.contains(id)));
234	1×	assert!((60..=140).contains(&ten.len()), "{}", ten.len());
235	1×	assert_eq!(bucket(7), bucket(7), "stable");
236	1×	}
237
238		#[test]
239	1×	fn the_nearest_level_wins_each_setting() {
240	1×	let mut l = Levels::default();
241	4×	let pol = \|scope: &str, key: &str, f: &dyn Fn(&mut Policy)\| {
242	4×	let mut p = Policy { scope: scope.into(), key: key.into(), ..Policy::default() };
243	4×	f(&mut p);
244	4×	((scope.to_string(), key.to_string()), p)
245	4×	};
246	1×	l.policies.extend([
247	1×	pol("site", "", &\|p\| {
248	1×	p.updates = Some(false);
249	1×	p.channel = Some("stable".into());
250	1×	}),
251	1×	pol("owner", "acme", &\|p\| p.updates = Some(true)),
252	1×	pol("group", "1", &\|p\| p.channel = Some("canary".into())),
253	1×	pol("runner", "3", &\|p\| p.updates = Some(false)),
254		]);
255	1×	l.groups.push(Group { id: 1, owner: "acme".into(), name: "canary".into(), percent: None, created_ms: 0 });
256	1×	l.chosen.insert((1, 2));
257	1×	l.chosen.insert((1, 9));
258
259	1×	let site_runner = l.resolve(&runner(1, ""));
260	1×	assert!(!site_runner.updates, "off at the site");
261	1×	assert_eq!(site_runner.from, ["site", "site", "default", "default"]);
262
263	1×	let canary = l.resolve(&runner(2, "acme"));
264	1×	assert!(canary.updates && canary.channel == "canary");
265	1×	assert_eq!(canary.from[..2], ["owner acme".to_string(), "group canary".to_string()]);
266
267	1×	let held = l.resolve(&runner(3, "acme"));
268	1×	assert!(!held.updates, "the runner's own setting wins");
269	1×	assert_eq!(held.channel, "stable");
270
271	1×	assert!(l.groups_of(&runner(9, "other")).is_empty(), "an owner's group holds only its own runners");
272	1×	assert!(!Levels::default().resolve(&runner(4, "")).updates, "off unless chosen");
273	1×	}
274
275		#[test]
276	1×	fn a_percentage_group_takes_its_share_of_the_runners_in_reach() {
277	1×	let mut l = Levels::default();
278	1×	l.groups.push(Group { id: 5, owner: String::new(), name: "farm 20%".into(), percent: Some(20), created_ms: 0 });
279	500×	let n = (1..=500).filter(\|id\| !l.groups_of(&runner(*id, "")).is_empty()).count();
280	1×	assert!((60..=140).contains(&n), "{n}");
281	1×	}
282
283		#[test]
284	1×	fn windows_and_channels_read_as_written() {
285	1×	assert_eq!(parse_window("02-05"), Ok(Some((2, 5))));
286	1×	assert_eq!(parse_window(" 22:00-04:00 "), Ok(Some((22, 4))));
287	1×	assert_eq!(parse_window(""), Ok(None));
288	1×	assert!(parse_window("5").is_err() && parse_window("3-3").is_err() && parse_window("1-24").is_err());
289	1×	assert!(in_window((2, 5), 2) && in_window((2, 5), 4) && !in_window((2, 5), 5));
290	1×	assert!(in_window((22, 4), 23) && in_window((22, 4), 1) && !in_window((22, 4), 12));
291	1×	assert!(valid_channel("stable") && valid_channel("canary") && valid_channel("0.3.4"));
292	1×	assert!(!valid_channel("latest") && !valid_channel("0.3.") && !valid_channel(""));
293	1×	assert_eq!(version_of("v0.3.2"), "0.3.2");
294	1×	assert_eq!(version_of("0.3.1 (local build)"), "0.3.1");
295	1×	assert_eq!(asset_for("linux", "riscv64"), "flow-linux-riscv64");
296	1×	assert_eq!(asset_for("windows", "x86_64"), "flow-windows-x86_64.exe");
297	1×	assert_eq!(
298	1×	rolled_back_version("rolled back from flow 0.4.0 to 0.3.1: 3 starts without reaching the hub").as_deref(),
299		Some("0.4.0")
300		);
301	1×	assert_eq!(rolled_back_version("runs flow 0.4.0 (was 0.3.1); proven"), None);
302	1×	}
303		}

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

Keyboard