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