| 1 | |
|
| 2 | |
|
| 3 | |
|
| 4 | |
|
| 5 | |
|
| 6 | | use crate::db::runner_updates::{self as rdb, Group, Policy, Soak}; |
| 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 | | pub const SETTING_SOAK_JOBS: &str = "runner_updates_soak_jobs"; |
| 19 | | pub const SETTING_SOAK_HOURS: &str = "runner_updates_soak_hours"; |
| 20 | | pub const SOAK_JOBS: i64 = 5; |
| 21 | | pub const SOAK_HOURS: i64 = 24; |
| 22 | |
|
| 23 | |
|
| 24 | |
|
| 25 | |
|
| 26 | |
|
| 27 | |
|
| 28 | |
|
| 29 | |
|
| 30 | 4× | pub fn share(app: &crate::Shared, offer: &UpdateOffer) { |
| 31 | | static SHARING: std::sync::Mutex<BTreeSet<String>> = std::sync::Mutex::new(BTreeSet::new()); |
| 32 | 4× | let store = app.images(); |
| 33 | 4× | let hash = offer.blake3.clone(); |
| 34 | 4× | if !jet::images::is_hash(&hash) || store.manifest(&hash).is_some() { |
| 35 | 1× | return; |
| 36 | 3× | } |
| 37 | 3× | if !SHARING.lock().unwrap_or_else(|p| p.into_inner()).insert(hash.clone()) { |
| 38 | 0× | return; |
| 39 | 3× | } |
| 40 | 3× | let (app, offer) = (app.clone(), offer.clone()); |
| 41 | 3× | tokio::spawn(async move { |
| 42 | 3× | let path = crate::web::releases::blob_path(&app, &hash); |
| 43 | 3× | let arch = offer.asset.rsplit('-').next().map(|a| a.trim_end_matches(".exe").to_string()); |
| 44 | 3× | let want = hash.clone(); |
| 45 | 3× | let made = tokio::task::spawn_blocking(move || -> Result<jet::images::Manifest, String> { |
| 46 | 4× | let m = jet::images::scan_file(&path, arch, |c, bytes| Ok(store.put_chunk(c.hash(), bytes)?)) |
| 47 | 3× | .map_err(|e| format!("reading {}: {e}", path.display()))?; |
| 48 | 1× | if m.hash != want { |
| 49 | 0× | return Err(format!("the asset hashes to {}, not {}", &m.hash[..12], &want[..12])); |
| 50 | 1× | } |
| 51 | 1× | store.verify_and_keep(&m)?; |
| 52 | 1× | Ok(m) |
| 53 | 3× | }) |
| 54 | 3× | .await |
| 55 | 3× | .unwrap_or_else(|e| Err(e.to_string())); |
| 56 | 3× | match made { |
| 57 | 1× | Ok(m) => match crate::images::record(&app.db, &m, &format!("runner build flow {}", offer.version)).await { |
| 58 | 1× | Ok(()) => tracing::info!( |
| 59 | | "runner updates: flow {} ({}) shared with the swarm: {} chunks", |
| 60 | | offer.version, |
| 61 | | offer.asset, |
| 62 | 0× | m.chunks.len() |
| 63 | | ), |
| 64 | 0× | Err(e) => tracing::warn!("runner updates: recording flow {} for the swarm: {e}", offer.version), |
| 65 | | }, |
| 66 | 2× | Err(e) => tracing::warn!("runner updates: sharing flow {} ({}) with the swarm: {e}", offer.version, offer.asset), |
| 67 | | } |
| 68 | 3× | SHARING.lock().unwrap_or_else(|p| p.into_inner()).remove(&hash); |
| 69 | 3× | }); |
| 70 | 4× | } |
| 71 | |
|
| 72 | |
|
| 73 | |
|
| 74 | |
|
| 75 | 2520× | pub fn bucket(runner_id: i64) -> i64 { |
| 76 | 2520× | let h = blake3::hash(format!("runner {runner_id}").as_bytes()); |
| 77 | 2520× | let mut first = [0u8; 8]; |
| 78 | 2520× | first.copy_from_slice(&h.as_bytes()[..8]); |
| 79 | 2520× | (u64::from_le_bytes(first) % 100) as i64 |
| 80 | 2520× | } |
| 81 | |
|
| 82 | |
|
| 83 | |
|
| 84 | 1114× | pub fn in_reach(g: &Group, r: &Runner) -> bool { |
| 85 | 1114× | g.owner.is_empty() || g.owner == r.scope_owner |
| 86 | 1114× | } |
| 87 | |
|
| 88 | |
|
| 89 | | #[derive(Clone, Debug, Default)] |
| 90 | | pub struct Levels { |
| 91 | | pub policies: HashMap<(String, String), Policy>, |
| 92 | | pub groups: Vec<Group>, |
| 93 | | |
| 94 | | pub chosen: BTreeSet<(i64, i64)>, |
| 95 | | } |
| 96 | |
|
| 97 | | impl Levels { |
| 98 | 78× | pub async fn load(pool: &Pool) -> sqlx::Result<Levels> { |
| 99 | 97× | let policies = rdb::policies(pool).await?.into_iter().map(|p| ((p.scope.clone(), p.key.clone()), p)).collect(); |
| 100 | 78× | Ok(Levels { policies, groups: rdb::groups(pool).await?, chosen: rdb::members(pool).await?.into_iter().collect() }) |
| 101 | 78× | } |
| 102 | |
|
| 103 | 1032× | pub fn member(&self, g: &Group, r: &Runner) -> bool { |
| 104 | 1032× | in_reach(g, r) && (self.chosen.contains(&(g.id, r.id)) || g.percent.is_some_and(|p| bucket(r.id) < p)) |
| 105 | 1032× | } |
| 106 | |
|
| 107 | | |
| 108 | 794× | pub fn groups_of(&self, r: &Runner) -> Vec<&Group> { |
| 109 | 967× | let mut gs: Vec<&Group> = self.groups.iter().filter(|g| self.member(g, r)).collect(); |
| 110 | 794× | gs.sort_by_key(|g| (g.owner.is_empty(), g.name.clone())); |
| 111 | 794× | gs |
| 112 | 794× | } |
| 113 | |
|
| 114 | 267× | fn get(&self, scope: &str, key: &str) -> Option<&Policy> { |
| 115 | 267× | self.policies.get(&(scope.to_string(), key.to_string())) |
| 116 | 267× | } |
| 117 | |
|
| 118 | | |
| 119 | | |
| 120 | 103× | pub fn wave_of(&self, r: &Runner) -> Option<i64> { |
| 121 | 103× | self.groups_of(r).iter().filter_map(|g| g.wave).min() |
| 122 | 103× | } |
| 123 | |
|
| 124 | | |
| 125 | | |
| 126 | | |
| 127 | 39× | pub fn waves_before<'a>(&self, r: &Runner, all: &'a [Runner]) -> Vec<(i64, Vec<&'a Runner>)> { |
| 128 | 39× | let mine = self.wave_of(r); |
| 129 | 39× | let mut waves: Vec<i64> = |
| 130 | 62× | self.groups.iter().filter(|g| in_reach(g, r)).filter_map(|g| g.wave).filter(|w| mine.is_none_or(|m| *w < m)).collect(); |
| 131 | 39× | waves.sort(); |
| 132 | 39× | waves.dedup(); |
| 133 | 39× | waves |
| 134 | 39× | .into_iter() |
| 135 | 39× | .map(|w| { |
| 136 | 18× | let members = all |
| 137 | 18× | .iter() |
| 138 | 62× | .filter(|o| self.wave_of(o) == Some(w) && self.groups_of(o).iter().any(|g| g.wave == Some(w) && in_reach(g, r))) |
| 139 | 18× | .collect(); |
| 140 | 18× | (w, members) |
| 141 | 18× | }) |
| 142 | 39× | .collect() |
| 143 | 39× | } |
| 144 | |
|
| 145 | | |
| 146 | 96× | fn chain(&self, r: &Runner) -> Vec<(String, &Policy)> { |
| 147 | 96× | let mut out = vec![]; |
| 148 | 96× | if let Some(p) = self.get("runner", &r.id.to_string()) { |
| 149 | 20× | out.push(("runner".to_string(), p)); |
| 150 | 76× | } |
| 151 | 96× | for g in self.groups_of(r) { |
| 152 | 51× | if let Some(p) = self.get("group", &g.id.to_string()) { |
| 153 | 27× | out.push((format!("group {}", g.name), p)); |
| 154 | 27× | } |
| 155 | | } |
| 156 | 96× | if !r.scope_owner.is_empty() { |
| 157 | 24× | if let Some(p) = self.get("owner", &r.scope_owner) { |
| 158 | 22× | out.push((format!("owner {}", r.scope_owner), p)); |
| 159 | 22× | } |
| 160 | 72× | } |
| 161 | 96× | if let Some(p) = self.get("site", "") { |
| 162 | 58× | out.push(("site".to_string(), p)); |
| 163 | 58× | } |
| 164 | 96× | out |
| 165 | 96× | } |
| 166 | |
|
| 167 | | |
| 168 | 96× | pub fn resolve(&self, r: &Runner) -> UpdatePolicy { |
| 169 | 96× | let chain = self.chain(r); |
| 170 | 384× | fn pick<T>(chain: &[(String, &Policy)], f: impl Fn(&Policy) -> Option<T>) -> (Option<T>, String) { |
| 171 | 418× | chain.iter().find_map(|(name, p)| f(p).map(|v| (Some(v), name.clone()))).unwrap_or((None, "default".into())) |
| 172 | 384× | } |
| 173 | 96× | let (updates, u_from) = pick(&chain, |p| p.updates); |
| 174 | 103× | let (channel, c_from) = pick(&chain, |p| p.channel.clone().filter(|c| !c.is_empty())); |
| 175 | 111× | let (window, w_from) = pick(&chain, |p| Some((p.window_from? as u8, p.window_to? as u8))); |
| 176 | 96× | let (dry_run, d_from) = pick(&chain, |p| p.dry_run); |
| 177 | | UpdatePolicy { |
| 178 | 96× | updates: updates.unwrap_or(false), |
| 179 | 96× | channel: channel.unwrap_or_else(|| "stable".into()), |
| 180 | 96× | window, |
| 181 | 96× | dry_run: dry_run.unwrap_or(false), |
| 182 | 96× | from: vec![u_from, c_from, w_from, d_from], |
| 183 | | } |
| 184 | 96× | } |
| 185 | | } |
| 186 | |
|
| 187 | |
|
| 188 | 17× | pub fn in_window((from, to): (u8, u8), h: u8) -> bool { |
| 189 | 17× | if from <= to { |
| 190 | 14× | from <= h && h < to |
| 191 | | } else { |
| 192 | 3× | h >= from || h < to |
| 193 | | } |
| 194 | 17× | } |
| 195 | |
|
| 196 | |
|
| 197 | 16× | pub fn parse_window(s: &str) -> Result<Option<(u8, u8)>, String> { |
| 198 | 16× | let s = s.trim(); |
| 199 | 16× | if s.is_empty() { |
| 200 | 9× | return Ok(None); |
| 201 | 7× | } |
| 202 | 7× | let (a, b) = s.split_once('-').ok_or("a window is two hours, from-to, e.g. 02-05")?; |
| 203 | 10× | let hour = |x: &str| x.trim().trim_end_matches(":00").parse::<u8>().ok().filter(|h| *h < 24); |
| 204 | 5× | match (hour(a), hour(b)) { |
| 205 | 4× | (Some(a), Some(b)) if a != b => Ok(Some((a, b))), |
| 206 | 2× | _ => Err("a window is two different hours from 0 to 23, from-to, e.g. 02-05".into()), |
| 207 | | } |
| 208 | 16× | } |
| 209 | |
|
| 210 | |
|
| 211 | 11× | pub fn valid_channel(c: &str) -> bool { |
| 212 | 15× | matches!(c, "stable" | "canary") || (!c.is_empty() && c.split('.').all(|p| !p.is_empty() && p.bytes().all(|b| b.is_ascii_digit()))) |
| 213 | 11× | } |
| 214 | |
|
| 215 | |
|
| 216 | |
|
| 217 | 254× | pub fn version_of(s: &str) -> String { |
| 218 | 254× | s.split_whitespace().next().unwrap_or("").trim_start_matches('v').to_string() |
| 219 | 254× | } |
| 220 | |
|
| 221 | 33× | fn older(a: &str, b: &str) -> bool { |
| 222 | 198× | let parts = |v: &str| v.split('.').map(|p| p.parse::<u64>().unwrap_or(0)).collect::<Vec<_>>(); |
| 223 | 33× | parts(a) < parts(b) |
| 224 | 33× | } |
| 225 | |
|
| 226 | |
|
| 227 | |
|
| 228 | 4× | pub fn rolled_back_version(note: &str) -> Option<String> { |
| 229 | 4× | let v = note.strip_prefix("rolled back from flow ")?.split_whitespace().next()?; |
| 230 | 6× | v.split('.').all(|p| !p.is_empty() && p.bytes().all(|b| b.is_ascii_digit())).then(|| v.to_string()) |
| 231 | 4× | } |
| 232 | |
|
| 233 | |
|
| 234 | |
|
| 235 | 28× | pub fn asset_for(os: &str, arch: &str) -> String { |
| 236 | 28× | format!("flow-{os}-{arch}{}", if os == "windows" { ".exe" } else { "" }) |
| 237 | 28× | } |
| 238 | |
|
| 239 | |
|
| 240 | | #[derive(Clone, Debug, PartialEq, Eq)] |
| 241 | | pub enum WaveState { |
| 242 | | |
| 243 | | |
| 244 | | |
| 245 | | Proved, |
| 246 | | |
| 247 | | Soaking { runners: usize, on: usize, jobs: i64 }, |
| 248 | | |
| 249 | | Stopped { runner_id: i64 }, |
| 250 | | } |
| 251 | |
|
| 252 | |
|
| 253 | |
|
| 254 | 16× | pub fn wave_state(members: &[i64], soaks: &[Soak], soak_jobs: i64, soak_ms: i64, now: i64) -> WaveState { |
| 255 | 16× | let mine: Vec<&Soak> = soaks.iter().filter(|s| members.contains(&s.runner_id) && s.jobs > 0).collect(); |
| 256 | 16× | if let Some(s) = mine.iter().find(|s| s.errors > 0) { |
| 257 | 3× | return WaveState::Stopped { runner_id: s.runner_id }; |
| 258 | 13× | } |
| 259 | 13× | if members.is_empty() { |
| 260 | 1× | return WaveState::Proved; |
| 261 | 12× | } |
| 262 | 12× | let jobs = mine.iter().map(|s| s.jobs).sum(); |
| 263 | 12× | let soaked = mine.iter().filter_map(|s| s.first_ms).min().is_some_and(|first| now - first >= soak_ms); |
| 264 | 12× | if !mine.is_empty() && (soaked || mine.iter().all(|s| s.jobs >= soak_jobs)) { |
| 265 | 5× | WaveState::Proved |
| 266 | | } else { |
| 267 | 7× | WaveState::Soaking { runners: members.len(), on: mine.len(), jobs } |
| 268 | | } |
| 269 | 16× | } |
| 270 | |
|
| 271 | |
|
| 272 | |
|
| 273 | 33× | async fn wave_gate( |
| 274 | 33× | pool: &Pool, |
| 275 | 33× | levels: &Levels, |
| 276 | 33× | r: &Runner, |
| 277 | 33× | version: &str, |
| 278 | 33× | now: i64, |
| 279 | 33× | ) -> sqlx::Result<Result<Option<String>, (String, String)>> { |
| 280 | 33× | let all = db::runners::list(pool).await?; |
| 281 | 33× | let waves = levels.waves_before(r, &all); |
| 282 | 33× | if waves.is_empty() { |
| 283 | 24× | return Ok(Ok(None)); |
| 284 | 9× | } |
| 285 | 18× | let setting = |v: Option<String>, d: i64| v.and_then(|s| s.trim().parse::<i64>().ok()).filter(|n| *n > 0).unwrap_or(d); |
| 286 | 9× | let soak_jobs = setting(db::accounts::setting(pool, SETTING_SOAK_JOBS).await?, SOAK_JOBS); |
| 287 | 9× | let soak_hours = setting(db::accounts::setting(pool, SETTING_SOAK_HOURS).await?, SOAK_HOURS); |
| 288 | 9× | let soaks = rdb::soaks(pool, version).await?; |
| 289 | 9× | for (w, members) in waves { |
| 290 | 9× | let ids: Vec<i64> = members.iter().map(|m| m.id).collect(); |
| 291 | 9× | match wave_state(&ids, &soaks, soak_jobs, soak_hours * 3_600_000, now) { |
| 292 | 2× | WaveState::Proved => continue, |
| 293 | 2× | WaveState::Stopped { runner_id } => { |
| 294 | 2× | let name = members.iter().find(|m| m.id == runner_id).map(|m| m.name.clone()).unwrap_or_default(); |
| 295 | 2× | let job = rdb::first_error(pool, version, runner_id).await?.unwrap_or_else(|| "a job".into()); |
| 296 | 2× | return Ok(Err((name.clone(), format!("wave {w} stopped it on {name}: {job} ended in error")))); |
| 297 | | } |
| 298 | 5× | WaveState::Soaking { runners, on, jobs } => { |
| 299 | 5× | return Ok(Ok(Some(format!( |
| 300 | | "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)", |
| 301 | 5× | if runners == 1 { "" } else { "s" }, |
| 302 | 5× | if jobs == 1 { "" } else { "s" }, |
| 303 | | )))); |
| 304 | | } |
| 305 | | } |
| 306 | | } |
| 307 | 2× | Ok(Ok(None)) |
| 308 | 33× | } |
| 309 | |
|
| 310 | |
|
| 311 | |
|
| 312 | |
|
| 313 | 92× | pub async fn offer( |
| 314 | 92× | pool: &Pool, |
| 315 | 92× | levels: &Levels, |
| 316 | 92× | r: &Runner, |
| 317 | 92× | policy: &UpdatePolicy, |
| 318 | 92× | hour: u8, |
| 319 | 92× | now: i64, |
| 320 | 92× | act: bool, |
| 321 | 92× | ) -> sqlx::Result<(Option<UpdateOffer>, String)> { |
| 322 | 92× | if !policy.updates { |
| 323 | 22× | return Ok((None, format!("updates are off ({})", policy.from[0]))); |
| 324 | 70× | } |
| 325 | 70× | let Some(spec) = db::accounts::setting(pool, SETTING_REPO).await?.filter(|s| !s.trim().is_empty()) else { |
| 326 | 16× | return Ok((None, "no runner releases repository is set (Admin › Runners › Updates)".into())); |
| 327 | | }; |
| 328 | 54× | let Some((owner, name)) = spec.trim().split_once('/') else { |
| 329 | 0× | return Ok((None, format!("the runner releases repository {spec:?} is not owner/name"))); |
| 330 | | }; |
| 331 | 54× | let Some(repo) = db::repos::get(pool, owner, name).await? else { |
| 332 | 0× | return Ok((None, format!("the runner releases repository {spec} does not exist"))); |
| 333 | | }; |
| 334 | 54× | let releases = db::releases::list(pool, repo.id).await?; |
| 335 | 54× | let release = match policy.channel.as_str() { |
| 336 | 54× | "stable" => releases.iter().find(|x| !x.prerelease), |
| 337 | 15× | "canary" => releases.first(), |
| 338 | 15× | v => releases.iter().find(|x| version_of(&x.tag) == v), |
| 339 | | }; |
| 340 | 54× | let Some(release) = release else { |
| 341 | 15× | return Ok((None, format!("{spec} has no release for the {} channel", policy.channel))); |
| 342 | | }; |
| 343 | 39× | let (want, have) = (version_of(&release.tag), version_of(&r.version)); |
| 344 | 39× | if want == have { |
| 345 | 0× | return Ok((None, format!("up to date: flow {have} ({} channel)", policy.channel))); |
| 346 | 39× | } |
| 347 | 39× | if let Some(h) = rdb::held(pool, &want).await? { |
| 348 | 6× | return Ok((None, format!("flow {want} is held: {}; lift the hold on Admin › Runners › Updates", h.why))); |
| 349 | 33× | } |
| 350 | 33× | let going_back = older(&want, &have); |
| 351 | 33× | if going_back && matches!(policy.channel.as_str(), "stable" | "canary") { |
| 352 | 0× | return Ok((None, format!("runs flow {have}, newer than the {} channel's {want}", policy.channel))); |
| 353 | 33× | } |
| 354 | | |
| 355 | 33× | if !going_back { |
| 356 | 33× | match wave_gate(pool, levels, r, &want, now).await? { |
| 357 | 26× | Ok(None) => {} |
| 358 | 5× | Ok(Some(why)) => return Ok((None, why)), |
| 359 | 2× | Err((runner, why)) => { |
| 360 | 2× | if act { |
| 361 | 1× | rdb::hold(pool, &want, &runner, &why).await?; |
| 362 | 1× | db::accounts::audit(pool, None, "runner.update.held", &want, &why).await.ok(); |
| 363 | 1× | } |
| 364 | 2× | return Ok((None, format!("flow {want} is held: {why}; lift the hold on Admin › Runners › Updates"))); |
| 365 | | } |
| 366 | | } |
| 367 | 0× | } |
| 368 | 26× | let file = asset_for(&r.os, &r.arch); |
| 369 | 26× | let assets = db::releases::assets(pool, release.id).await?; |
| 370 | 26× | let Some(a) = assets.iter().find(|a| a.name == file) else { |
| 371 | 0× | return Ok((None, format!("{spec} {} has no {file}", release.tag))); |
| 372 | | }; |
| 373 | 26× | if let Some(w) = policy.window.filter(|w| !in_window(*w, hour)) { |
| 374 | 11× | return Ok((None, format!("flow {want} waits for its window, {:02}:00–{:02}:00 UTC", w.0, w.1))); |
| 375 | 15× | } |
| 376 | 15× | let note = format!( |
| 377 | | "flow {want} ({} channel){}", |
| 378 | | policy.channel, |
| 379 | 15× | if policy.dry_run { ": dry run, check it and say what would happen" } else { "" } |
| 380 | | ); |
| 381 | 15× | Ok(( |
| 382 | 15× | Some(UpdateOffer { |
| 383 | 15× | version: want, |
| 384 | 15× | tag: release.tag.clone(), |
| 385 | 15× | asset: a.name.clone(), |
| 386 | 15× | url: format!("/{owner}/{name}/asset/{}/{}", a.id, a.name), |
| 387 | 15× | blake3: a.hash.clone(), |
| 388 | 15× | size: a.size as u64, |
| 389 | 15× | older: going_back, |
| 390 | 15× | }), |
| 391 | 15× | note, |
| 392 | 15× | )) |
| 393 | 92× | } |
| 394 | |
|
| 395 | | #[cfg(test)] |
| 396 | | mod tests { |
| 397 | | use super::*; |
| 398 | |
|
| 399 | 511× | fn runner(id: i64, owner: &str) -> Runner { |
| 400 | 511× | Runner { id, scope_owner: owner.into(), version: "0.3.1 (build)".into(), ..Runner::default() } |
| 401 | 511× | } |
| 402 | |
|
| 403 | | #[test] |
| 404 | 1× | fn a_smaller_share_sits_inside_a_larger_one_and_is_about_its_size() { |
| 405 | 1000× | let ten: Vec<i64> = (1..=1000).filter(|id| bucket(*id) < 10).collect(); |
| 406 | 1000× | let fifty: Vec<i64> = (1..=1000).filter(|id| bucket(*id) < 50).collect(); |
| 407 | 96× | assert!(ten.iter().all(|id| fifty.contains(id))); |
| 408 | 1× | assert!((60..=140).contains(&ten.len()), "{}", ten.len()); |
| 409 | 1× | assert_eq!(bucket(7), bucket(7), "stable"); |
| 410 | 1× | } |
| 411 | |
|
| 412 | | #[test] |
| 413 | 1× | fn the_nearest_level_wins_each_setting() { |
| 414 | 1× | let mut l = Levels::default(); |
| 415 | 4× | let pol = |scope: &str, key: &str, f: &dyn Fn(&mut Policy)| { |
| 416 | 4× | let mut p = Policy { scope: scope.into(), key: key.into(), ..Policy::default() }; |
| 417 | 4× | f(&mut p); |
| 418 | 4× | ((scope.to_string(), key.to_string()), p) |
| 419 | 4× | }; |
| 420 | 1× | l.policies.extend([ |
| 421 | 1× | pol("site", "", &|p| { |
| 422 | 1× | p.updates = Some(false); |
| 423 | 1× | p.channel = Some("stable".into()); |
| 424 | 1× | }), |
| 425 | 1× | pol("owner", "acme", &|p| p.updates = Some(true)), |
| 426 | 1× | pol("group", "1", &|p| p.channel = Some("canary".into())), |
| 427 | 1× | pol("runner", "3", &|p| p.updates = Some(false)), |
| 428 | | ]); |
| 429 | 1× | l.groups.push(Group { id: 1, owner: "acme".into(), name: "canary".into(), percent: None, created_ms: 0, wave: None }); |
| 430 | 1× | l.chosen.insert((1, 2)); |
| 431 | 1× | l.chosen.insert((1, 9)); |
| 432 | |
|
| 433 | 1× | let site_runner = l.resolve(&runner(1, "")); |
| 434 | 1× | assert!(!site_runner.updates, "off at the site"); |
| 435 | 1× | assert_eq!(site_runner.from, ["site", "site", "default", "default"]); |
| 436 | |
|
| 437 | 1× | let canary = l.resolve(&runner(2, "acme")); |
| 438 | 1× | assert!(canary.updates && canary.channel == "canary"); |
| 439 | 1× | assert_eq!(canary.from[..2], ["owner acme".to_string(), "group canary".to_string()]); |
| 440 | |
|
| 441 | 1× | let held = l.resolve(&runner(3, "acme")); |
| 442 | 1× | assert!(!held.updates, "the runner's own setting wins"); |
| 443 | 1× | assert_eq!(held.channel, "stable"); |
| 444 | |
|
| 445 | 1× | assert!(l.groups_of(&runner(9, "other")).is_empty(), "an owner's group holds only its own runners"); |
| 446 | 1× | assert!(!Levels::default().resolve(&runner(4, "")).updates, "off unless chosen"); |
| 447 | 1× | } |
| 448 | |
|
| 449 | | #[test] |
| 450 | 1× | fn a_percentage_group_takes_its_share_of_the_runners_in_reach() { |
| 451 | 1× | let mut l = Levels::default(); |
| 452 | 1× | l.groups.push(Group { id: 5, owner: String::new(), name: "farm 20%".into(), percent: Some(20), created_ms: 0, wave: None }); |
| 453 | 500× | let n = (1..=500).filter(|id| !l.groups_of(&runner(*id, "")).is_empty()).count(); |
| 454 | 1× | assert!((60..=140).contains(&n), "{n}"); |
| 455 | 1× | } |
| 456 | |
|
| 457 | | #[test] |
| 458 | 1× | fn a_wave_proves_a_version_by_its_jobs_or_its_time_and_an_error_stops_it() { |
| 459 | 8× | let soak = |runner_id, jobs, errors, first_ms| Soak { runner_id, jobs, errors, first_ms: Some(first_ms) }; |
| 460 | 1× | let hour = 3_600_000; |
| 461 | | |
| 462 | 1× | assert_eq!(wave_state(&[1, 2], &[], 5, 24 * hour, 0), WaveState::Soaking { runners: 2, on: 0, jobs: 0 }); |
| 463 | | |
| 464 | 1× | let s = [soak(1, 3, 0, 0), soak(9, 50, 0, 0)]; |
| 465 | 1× | assert_eq!(wave_state(&[1, 2], &s, 5, 24 * hour, hour), WaveState::Soaking { runners: 2, on: 1, jobs: 3 }); |
| 466 | | |
| 467 | 1× | assert_eq!(wave_state(&[1, 2], &[soak(1, 5, 0, 0)], 5, 24 * hour, hour), WaveState::Proved); |
| 468 | | |
| 469 | 1× | assert_eq!(wave_state(&[1, 2], &[soak(1, 1, 0, 0)], 5, 24 * hour, 24 * hour), WaveState::Proved); |
| 470 | | |
| 471 | 1× | assert_eq!( |
| 472 | 1× | wave_state(&[1, 2], &[soak(1, 40, 0, 0), soak(2, 2, 1, 0)], 5, 24 * hour, 48 * hour), |
| 473 | | WaveState::Stopped { runner_id: 2 } |
| 474 | | ); |
| 475 | | |
| 476 | 1× | assert_eq!(wave_state(&[1], &[soak(1, 5, 0, 0), soak(2, 2, 1, 0)], 5, 24 * hour, hour), WaveState::Proved); |
| 477 | | |
| 478 | 1× | assert_eq!(wave_state(&[], &[], 5, 24 * hour, 0), WaveState::Proved); |
| 479 | 1× | } |
| 480 | |
|
| 481 | | #[test] |
| 482 | 1× | fn a_runner_waits_for_the_waves_before_its_own() { |
| 483 | 1× | let mut l = Levels::default(); |
| 484 | 3× | let group = |id, name: &str, wave| Group { id, owner: String::new(), name: name.into(), percent: None, created_ms: 0, wave }; |
| 485 | 1× | l.groups.extend([group(1, "first", Some(1)), group(2, "second", Some(2)), group(3, "watchers", None)]); |
| 486 | 1× | l.chosen.extend([(1, 1), (2, 2), (2, 1), (3, 3)]); |
| 487 | 4× | let all: Vec<Runner> = (1..=4).map(|id| runner(id, "")).collect(); |
| 488 | 5× | let ids = |ws: Vec<(i64, Vec<&Runner>)>| { |
| 489 | 9× | ws.into_iter().map(|(w, rs)| (w, rs.iter().map(|r| r.id).collect::<Vec<_>>())).collect::<Vec<_>>() |
| 490 | 5× | }; |
| 491 | 1× | assert_eq!(l.wave_of(&all[0]), Some(1), "its first wave"); |
| 492 | 1× | assert!(l.waves_before(&all[0], &all).is_empty(), "wave 1 goes first"); |
| 493 | 1× | assert_eq!(ids(l.waves_before(&all[1], &all)), vec![(1, vec![1])]); |
| 494 | 1× | assert_eq!(l.wave_of(&all[2]), None, "a group with no wave"); |
| 495 | 1× | assert_eq!(ids(l.waves_before(&all[2], &all)), vec![(1, vec![1]), (2, vec![2])], "in no wave: after every wave"); |
| 496 | 1× | assert_eq!(ids(l.waves_before(&all[3], &all)), vec![(1, vec![1]), (2, vec![2])]); |
| 497 | | |
| 498 | 1× | l.groups.push(Group { id: 4, owner: "acme".into(), name: "acme first".into(), percent: None, created_ms: 0, wave: Some(1) }); |
| 499 | 1× | l.chosen.insert((4, 5)); |
| 500 | 1× | let mut all = all; |
| 501 | 1× | all.push(runner(5, "acme")); |
| 502 | 1× | all.push(runner(6, "acme")); |
| 503 | 1× | assert_eq!(ids(l.waves_before(&all[5], &all)), vec![(1, vec![1, 5]), (2, vec![2])]); |
| 504 | 1× | assert_eq!(ids(l.waves_before(&all[3], &all)), vec![(1, vec![1]), (2, vec![2])], "not another owner's"); |
| 505 | 1× | } |
| 506 | |
|
| 507 | | #[test] |
| 508 | 1× | fn windows_and_channels_read_as_written() { |
| 509 | 1× | assert_eq!(parse_window("02-05"), Ok(Some((2, 5)))); |
| 510 | 1× | assert_eq!(parse_window(" 22:00-04:00 "), Ok(Some((22, 4)))); |
| 511 | 1× | assert_eq!(parse_window(""), Ok(None)); |
| 512 | 1× | assert!(parse_window("5").is_err() && parse_window("3-3").is_err() && parse_window("1-24").is_err()); |
| 513 | 1× | assert!(in_window((2, 5), 2) && in_window((2, 5), 4) && !in_window((2, 5), 5)); |
| 514 | 1× | assert!(in_window((22, 4), 23) && in_window((22, 4), 1) && !in_window((22, 4), 12)); |
| 515 | 1× | assert!(valid_channel("stable") && valid_channel("canary") && valid_channel("0.3.4")); |
| 516 | 1× | assert!(!valid_channel("latest") && !valid_channel("0.3.") && !valid_channel("")); |
| 517 | 1× | assert_eq!(version_of("v0.3.2"), "0.3.2"); |
| 518 | 1× | assert_eq!(version_of("0.3.1 (local build)"), "0.3.1"); |
| 519 | 1× | assert_eq!(asset_for("linux", "riscv64"), "flow-linux-riscv64"); |
| 520 | 1× | assert_eq!(asset_for("windows", "x86_64"), "flow-windows-x86_64.exe"); |
| 521 | 1× | assert_eq!( |
| 522 | 1× | rolled_back_version("rolled back from flow 0.4.0 to 0.3.1: 3 starts without reaching the hub").as_deref(), |
| 523 | | Some("0.4.0") |
| 524 | | ); |
| 525 | 1× | assert_eq!(rolled_back_version("runs flow 0.4.0 (was 0.3.1); proven"), None); |
| 526 | 1× | } |
| 527 | | } |