OXYGENHUB

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

414 of 428 lines run, 149 of 185 functions · run #541 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, 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
/// 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
/// The site settings for proving a version in a wave (oxygenflow#84): the
16
/// jobs each runner of the wave that took it must finish, or the hours
17
/// since its first job there after which one job is enough.
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
/// Share an offered build with the swarm (oxygenflow#84): cut the release
24
/// asset into the hub's image chunks and keep its manifest, named, as an
25
/// image is, by the build's BLAKE3, so a runner in a swarm fetches it from
26
/// its site's peers and the hub together, as it fetches images, and the
27
/// hub serves it about once per site. Done once, in the background, the
28
/// first time the build is offered; a runner asking before it is done
29
/// fetches the build over HTTP.
304×
pub fn share(app: &crate::Shared, offer: &UpdateOffer) {
31
    static SHARING: std::sync::Mutex<BTreeSet<String>> = std::sync::Mutex::new(BTreeSet::new());
324×
    let store = app.images();
334×
    let hash = offer.blake3.clone();
344×
    if !jet::images::is_hash(&hash) || store.manifest(&hash).is_some() {
351×
        return;
363×
    }
373×
    if !SHARING.lock().unwrap_or_else(|p| p.into_inner()).insert(hash.clone()) {
380×
        return;
393×
    }
403×
    let (app, offer) = (app.clone(), offer.clone());
413×
    tokio::spawn(async move {
423×
        let path = crate::web::releases::blob_path(&app, &hash);
433×
        let arch = offer.asset.rsplit('-').next().map(|a| a.trim_end_matches(".exe").to_string());
443×
        let want = hash.clone();
453×
        let made = tokio::task::spawn_blocking(move || -> Result<jet::images::Manifest, String> {
464×
            let m = jet::images::scan_file(&path, arch, |c, bytes| Ok(store.put_chunk(c.hash(), bytes)?))
473×
                .map_err(|e| format!("reading {}: {e}", path.display()))?;
481×
            if m.hash != want {
490×
                return Err(format!("the asset hashes to {}, not {}", &m.hash[..12], &want[..12]));
501×
            }
511×
            store.verify_and_keep(&m)?;
521×
            Ok(m)
533×
        })
543×
        .await
553×
        .unwrap_or_else(|e| Err(e.to_string()));
563×
        match made {
571×
            Ok(m) => match crate::images::record(&app.db, &m, &format!("runner build flow {}", offer.version)).await {
581×
                Ok(()) => tracing::info!(
59
                    "runner updates: flow {} ({}) shared with the swarm: {} chunks",
60
                    offer.version,
61
                    offer.asset,
620×
                    m.chunks.len()
63
                ),
640×
                Err(e) => tracing::warn!("runner updates: recording flow {} for the swarm: {e}", offer.version),
65
            },
662×
            Err(e) => tracing::warn!("runner updates: sharing flow {} ({}) with the swarm: {e}", offer.version, offer.asset),
67
        }
683×
        SHARING.lock().unwrap_or_else(|p| p.into_inner()).remove(&hash);
693×
    });
704×
}
71
72
/// A runner's place, 0 to 99, for percentage groups: a stable hash of its
73
/// id, so a group's share is the same machines every rollout, and a
74
/// smaller share sits inside a larger one.
752520×
pub fn bucket(runner_id: i64) -> i64 {
762520×
    let h = blake3::hash(format!("runner {runner_id}").as_bytes());
772520×
    let mut first = [0u8; 8];
782520×
    first.copy_from_slice(&h.as_bytes()[..8]);
792520×
    (u64::from_le_bytes(first) % 100) as i64
802520×
}
81
82
/// Whether a group may hold the runner at all: a site group holds any, an
83
/// owner's group only that owner's runners.
841114×
pub fn in_reach(g: &Group, r: &Runner) -> bool {
851114×
    g.owner.is_empty() || g.owner == r.scope_owner
861114×
}
87
88
/// Every level's settings and the groups, read once.
89
#[derive(Clone, Debug, Default)]
90
pub struct Levels {
91
    pub policies: HashMap<(String, String), Policy>,
92
    pub groups: Vec<Group>,
93
    /// Chosen members, as (group, runner).
94
    pub chosen: BTreeSet<(i64, i64)>,
95
}
96
97
impl Levels {
9880×
    pub async fn load(pool: &Pool) -> sqlx::Result<Levels> {
99101×
        let policies = rdb::policies(pool).await?.into_iter().map(|p| ((p.scope.clone(), p.key.clone()), p)).collect();
10080×
        Ok(Levels { policies, groups: rdb::groups(pool).await?, chosen: rdb::members(pool).await?.into_iter().collect() })
10180×
    }
102
1031032×
    pub fn member(&self, g: &Group, r: &Runner) -> bool {
1041032×
        in_reach(g, r) && (self.chosen.contains(&(g.id, r.id)) || g.percent.is_some_and(|p| bucket(r.id) < p))
1051032×
    }
106
107
    /// The runner's groups, nearest first: its owner's, then the site's.
108802×
    pub fn groups_of(&self, r: &Runner) -> Vec<&Group> {
109967×
        let mut gs: Vec<&Group> = self.groups.iter().filter(|g| self.member(g, r)).collect();
110802×
        gs.sort_by_key(|g| (g.owner.is_empty(), g.name.clone()));
111802×
        gs
112802×
    }
113
114275×
    fn get(&self, scope: &str, key: &str) -> Option<&Policy> {
115275×
        self.policies.get(&(scope.to_string(), key.to_string()))
116275×
    }
117
118
    /// The wave `r` takes a version in: the first of its groups' waves;
119
    /// `None` after every wave.
120104×
    pub fn wave_of(&self, r: &Runner) -> Option<i64> {
121104×
        self.groups_of(r).iter().filter_map(|g| g.wave).min()
122104×
    }
123
124
    /// The waves before `r`'s, in order, each with its runners among `all`:
125
    /// the waves of the groups that may hold `r` (the site's, and its
126
    /// owner's), a runner counted in its own first wave.
12740×
    pub fn waves_before<'a>(&self, r: &Runner, all: &'a [Runner]) -> Vec<(i64, Vec<&'a Runner>)> {
12840×
        let mine = self.wave_of(r);
12940×
        let mut waves: Vec<i64> =
13062×
            self.groups.iter().filter(|g| in_reach(g, r)).filter_map(|g| g.wave).filter(|w| mine.is_none_or(|m| *w < m)).collect();
13140×
        waves.sort();
13240×
        waves.dedup();
13340×
        waves
13440×
            .into_iter()
13540×
            .map(|w| {
13618×
                let members = all
13718×
                    .iter()
13862×
                    .filter(|o| self.wave_of(o) == Some(w) && self.groups_of(o).iter().any(|g| g.wave == Some(w) && in_reach(g, r)))
13918×
                    .collect();
14018×
                (w, members)
14118×
            })
14240×
            .collect()
14340×
    }
144
145
    /// The levels that apply to `r`, nearest first, each with its name.
146100×
    fn chain(&self, r: &Runner) -> Vec<(String, &Policy)> {
147100×
        let mut out = vec![];
148100×
        if let Some(p) = self.get("runner", &r.id.to_string()) {
14921×
            out.push(("runner".to_string(), p));
15079×
        }
151100×
        for g in self.groups_of(r) {
15251×
            if let Some(p) = self.get("group", &g.id.to_string()) {
15327×
                out.push((format!("group {}", g.name), p));
15427×
            }
155
        }
156100×
        if !r.scope_owner.is_empty() {
15724×
            if let Some(p) = self.get("owner", &r.scope_owner) {
15822×
                out.push((format!("owner {}", r.scope_owner), p));
15922×
            }
16076×
        }
161100×
        if let Some(p) = self.get("site", "") {
16262×
            out.push(("site".to_string(), p));
16362×
        }
164100×
        out
165100×
    }
166
167
    /// The policy for `r`: each setting from the nearest level that sets it.
168100×
    pub fn resolve(&self, r: &Runner) -> UpdatePolicy {
169100×
        let chain = self.chain(r);
170400×
        fn pick<T>(chain: &[(String, &Policy)], f: impl Fn(&Policy) -> Option<T>) -> (Option<T>, String) {
171435×
            chain.iter().find_map(|(name, p)| f(p).map(|v| (Some(v), name.clone()))).unwrap_or((None, "default".into()))
172400×
        }
173100×
        let (updates, u_from) = pick(&chain, |p| p.updates);
174107×
        let (channel, c_from) = pick(&chain, |p| p.channel.clone().filter(|c| !c.is_empty()));
175116×
        let (window, w_from) = pick(&chain, |p| Some((p.window_from? as u8, p.window_to? as u8)));
176100×
        let (dry_run, d_from) = pick(&chain, |p| p.dry_run);
177
        UpdatePolicy {
178100×
            updates: updates.unwrap_or(false),
179100×
            channel: channel.unwrap_or_else(|| "stable".into()),
180100×
            window,
181100×
            dry_run: dry_run.unwrap_or(false),
182100×
            from: vec![u_from, c_from, w_from, d_from],
183
        }
184100×
    }
185
}
186
187
/// Whether hour `h` (UTC) falls in `[from, to)`, which may wrap past midnight.
18817×
pub fn in_window((from, to): (u8, u8), h: u8) -> bool {
18917×
    if from <= to {
19014×
        from <= h && h < to
191
    } else {
1923×
        h >= from || h < to
193
    }
19417×
}
195
196
/// Parse `02-05` (hours, UTC) as a window; blank is none.
19717×
pub fn parse_window(s: &str) -> Result<Option<(u8, u8)>, String> {
19817×
    let s = s.trim();
19917×
    if s.is_empty() {
20010×
        return Ok(None);
2017×
    }
2027×
    let (a, b) = s.split_once('-').ok_or("a window is two hours, from-to, e.g. 02-05")?;
20310×
    let hour = |x: &str| x.trim().trim_end_matches(":00").parse::<u8>().ok().filter(|h| *h < 24);
2045×
    match (hour(a), hour(b)) {
2054×
        (Some(a), Some(b)) if a != b => Ok(Some((a, b))),
2062×
        _ => Err("a window is two different hours from 0 to 23, from-to, e.g. 02-05".into()),
207
    }
20817×
}
209
210
/// A channel as written: `stable`, `canary`, or a version.
21111×
pub fn valid_channel(c: &str) -> bool {
21215×
    matches!(c, "stable" | "canary") || (!c.is_empty() && c.split('.').all(|p| !p.is_empty() && p.bytes().all(|b| b.is_ascii_digit())))
21311×
}
214
215
/// The version in a runner's or a tag's words: `0.3.1 (build)` → `0.3.1`,
216
/// `v0.3.1` → `0.3.1`.
217259×
pub fn version_of(s: &str) -> String {
218259×
    s.split_whitespace().next().unwrap_or("").trim_start_matches('v').to_string()
219259×
}
220
22134×
fn older(a: &str, b: &str) -> bool {
222204×
    let parts = |v: &str| v.split('.').map(|p| p.parse::<u64>().unwrap_or(0)).collect::<Vec<_>>();
22334×
    parts(a) < parts(b)
22434×
}
225
226
/// The version a runner's report says rolled back: `rolled back from flow
227
/// 0.4.0 to 0.3.1: …` (the runner's words, oxygenflow spec §11.5.3) → `0.4.0`.
2284×
pub fn rolled_back_version(note: &str) -> Option<String> {
2294×
    let v = note.strip_prefix("rolled back from flow ")?.split_whitespace().next()?;
2306×
    v.split('.').all(|p| !p.is_empty() && p.bytes().all(|b| b.is_ascii_digit())).then(|| v.to_string())
2314×
}
232
233
/// The release asset a runner on `os` and `arch` runs: `flow-linux-x86_64`,
234
/// `flow-windows-x86_64.exe`.
23529×
pub fn asset_for(os: &str, arch: &str) -> String {
23629×
    format!("flow-{os}-{arch}{}", if os == "windows" { ".exe" } else { "" })
23729×
}
238
239
/// Where a wave stands with a version (oxygenflow#84).
240
#[derive(Clone, Debug, PartialEq, Eq)]
241
pub enum WaveState {
242
    /// Every runner of the wave that took it has finished its soak jobs on
243
    /// it, or the soak time has passed since its first job there; or the
244
    /// wave has no runners.
245
    Proved,
246
    /// Not yet: `on` of `runners` run it, with `jobs` jobs between them.
247
    Soaking { runners: usize, on: usize, jobs: i64 },
248
    /// A job on it ended in error on this runner: the version stops here.
249
    Stopped { runner_id: i64 },
250
}
251
252
/// Where a wave of `members` stands with a version, from each runner's
253
/// [`Soak`] on it, at `now` (ms).
25416×
pub fn wave_state(members: &[i64], soaks: &[Soak], soak_jobs: i64, soak_ms: i64, now: i64) -> WaveState {
25516×
    let mine: Vec<&Soak> = soaks.iter().filter(|s| members.contains(&s.runner_id) && s.jobs > 0).collect();
25616×
    if let Some(s) = mine.iter().find(|s| s.errors > 0) {
2573×
        return WaveState::Stopped { runner_id: s.runner_id };
25813×
    }
25913×
    if members.is_empty() {
2601×
        return WaveState::Proved;
26112×
    }
26212×
    let jobs = mine.iter().map(|s| s.jobs).sum();
26312×
    let soaked = mine.iter().filter_map(|s| s.first_ms).min().is_some_and(|first| now - first >= soak_ms);
26412×
    if !mine.is_empty() && (soaked || mine.iter().all(|s| s.jobs >= soak_jobs)) {
2655×
        WaveState::Proved
266
    } else {
2677×
        WaveState::Soaking { runners: members.len(), on: mine.len(), jobs }
268
    }
26916×
}
270
271
/// What the waves before `r`'s say about `version`: `Ok(None)` go ahead;
272
/// `Ok(Some(why))` wait; `Err((runner, why))` it stopped on that runner.
27334×
async fn wave_gate(
27434×
    pool: &Pool,
27534×
    levels: &Levels,
27634×
    r: &Runner,
27734×
    version: &str,
27834×
    now: i64,
27934×
) -> sqlx::Result<Result<Option<String>, (String, String)>> {
28034×
    let all = db::runners::list(pool).await?;
28134×
    let waves = levels.waves_before(r, &all);
28234×
    if waves.is_empty() {
28325×
        return Ok(Ok(None));
2849×
    }
28518×
    let setting = |v: Option<String>, d: i64| v.and_then(|s| s.trim().parse::<i64>().ok()).filter(|n| *n > 0).unwrap_or(d);
2869×
    let soak_jobs = setting(db::accounts::setting(pool, SETTING_SOAK_JOBS).await?, SOAK_JOBS);
2879×
    let soak_hours = setting(db::accounts::setting(pool, SETTING_SOAK_HOURS).await?, SOAK_HOURS);
2889×
    let soaks = rdb::soaks(pool, version).await?;
2899×
    for (w, members) in waves {
2909×
        let ids: Vec<i64> = members.iter().map(|m| m.id).collect();
2919×
        match wave_state(&ids, &soaks, soak_jobs, soak_hours * 3_600_000, now) {
2922×
            WaveState::Proved => continue,
2932×
            WaveState::Stopped { runner_id } => {
2942×
                let name = members.iter().find(|m| m.id == runner_id).map(|m| m.name.clone()).unwrap_or_default();
2952×
                let job = rdb::first_error(pool, version, runner_id).await?.unwrap_or_else(|| "a job".into());
2962×
                return Ok(Err((name.clone(), format!("wave {w} stopped it on {name}: {job} ended in error"))));
297
            }
2985×
            WaveState::Soaking { runners, on, jobs } => {
2995×
                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)",
3015×
                    if runners == 1 { "" } else { "s" },
3025×
                    if jobs == 1 { "" } else { "s" },
303
                ))));
304
            }
305
        }
306
    }
3072×
    Ok(Ok(None))
30834×
}
309
310
/// The build `r` should install under `policy` at hour `hour` (UTC), or
311
/// why there is none. A version a wave stopped is held back from every
312
/// runner when `act` (a runner asking; a page only says so).
31396×
pub async fn offer(
31496×
    pool: &Pool,
31596×
    levels: &Levels,
31696×
    r: &Runner,
31796×
    policy: &UpdatePolicy,
31896×
    hour: u8,
31996×
    now: i64,
32096×
    act: bool,
32196×
) -> sqlx::Result<(Option<UpdateOffer>, String)> {
32296×
    if !policy.updates {
32322×
        return Ok((None, format!("updates are off ({})", policy.from[0])));
32474×
    }
325
    // A phone's runner is part of the OxygenHub app (oxygenflow#84,
326
    // oxygenhub-android#7): it updates when the app does, never from a
327
    // runner build.
32874×
    if r.os == "android" {
3292×
        return Ok((
3302×
            None,
3312×
            format!("follows the OxygenHub app's releases: runs flow {} (update the app to update it)", version_of(&r.version)),
3322×
        ));
33372×
    }
33472×
    let Some(spec) = db::accounts::setting(pool, SETTING_REPO).await?.filter(|s| !s.trim().is_empty()) else {
33516×
        return Ok((None, "no runner releases repository is set (Admin › Runners › Updates)".into()));
336
    };
33756×
    let Some((owner, name)) = spec.trim().split_once('/') else {
3380×
        return Ok((None, format!("the runner releases repository {spec:?} is not owner/name")));
339
    };
34056×
    let Some(repo) = db::repos::get(pool, owner, name).await? else {
3410×
        return Ok((None, format!("the runner releases repository {spec} does not exist")));
342
    };
34356×
    let releases = db::releases::list(pool, repo.id).await?;
34456×
    let release = match policy.channel.as_str() {
34556×
        "stable" => releases.iter().find(|x| !x.prerelease),
34616×
        "canary" => releases.first(),
34716×
        v => releases.iter().find(|x| version_of(&x.tag) == v),
348
    };
34956×
    let Some(release) = release else {
35016×
        return Ok((None, format!("{spec} has no release for the {} channel", policy.channel)));
351
    };
35240×
    let (want, have) = (version_of(&release.tag), version_of(&r.version));
35340×
    if want == have {
3540×
        return Ok((None, format!("up to date: flow {have} ({} channel)", policy.channel)));
35540×
    }
35640×
    if let Some(h) = rdb::held(pool, &want).await? {
3576×
        return Ok((None, format!("flow {want} is held: {}; lift the hold on Admin › Runners › Updates", h.why)));
35834×
    }
35934×
    let going_back = older(&want, &have);
36034×
    if going_back && matches!(policy.channel.as_str(), "stable" | "canary") {
3610×
        return Ok((None, format!("runs flow {have}, newer than the {} channel's {want}", policy.channel)));
36234×
    }
363
    // Waves (oxygenflow#84): a later wave takes a version once those before it proved it.
36434×
    if !going_back {
36534×
        match wave_gate(pool, levels, r, &want, now).await? {
36627×
            Ok(None) => {}
3675×
            Ok(Some(why)) => return Ok((None, why)),
3682×
            Err((runner, why)) => {
3692×
                if act {
3701×
                    rdb::hold(pool, &want, &runner, &why).await?;
3711×
                    db::accounts::audit(pool, None, "runner.update.held", &want, &why).await.ok();
3721×
                }
3732×
                return Ok((None, format!("flow {want} is held: {why}; lift the hold on Admin › Runners › Updates")));
374
            }
375
        }
3760×
    }
37727×
    let file = asset_for(&r.os, &r.arch);
37827×
    let assets = db::releases::assets(pool, release.id).await?;
37927×
    let Some(a) = assets.iter().find(|a| a.name == file) else {
3800×
        return Ok((None, format!("{spec} {} has no {file}", release.tag)));
381
    };
38227×
    if let Some(w) = policy.window.filter(|w| !in_window(*w, hour)) {
38311×
        return Ok((None, format!("flow {want} waits for its window, {:02}:00–{:02}:00 UTC", w.0, w.1)));
38416×
    }
38516×
    let note = format!(
386
        "flow {want} ({} channel){}",
387
        policy.channel,
38816×
        if policy.dry_run { ": dry run, check it and say what would happen" } else { "" }
389
    );
39016×
    Ok((
39116×
        Some(UpdateOffer {
39216×
            version: want,
39316×
            tag: release.tag.clone(),
39416×
            asset: a.name.clone(),
39516×
            url: format!("/{owner}/{name}/asset/{}/{}", a.id, a.name),
39616×
            blake3: a.hash.clone(),
39716×
            size: a.size as u64,
39816×
            older: going_back,
39916×
        }),
40016×
        note,
40116×
    ))
40296×
}
403
404
#[cfg(test)]
405
mod tests {
406
    use super::*;
407
408511×
    fn runner(id: i64, owner: &str) -> Runner {
409511×
        Runner { id, scope_owner: owner.into(), version: "0.3.1 (build)".into(), ..Runner::default() }
410511×
    }
411
412
    #[test]
4131×
    fn a_smaller_share_sits_inside_a_larger_one_and_is_about_its_size() {
4141000×
        let ten: Vec<i64> = (1..=1000).filter(|id| bucket(*id) < 10).collect();
4151000×
        let fifty: Vec<i64> = (1..=1000).filter(|id| bucket(*id) < 50).collect();
41696×
        assert!(ten.iter().all(|id| fifty.contains(id)));
4171×
        assert!((60..=140).contains(&ten.len()), "{}", ten.len());
4181×
        assert_eq!(bucket(7), bucket(7), "stable");
4191×
    }
420
421
    #[test]
4221×
    fn the_nearest_level_wins_each_setting() {
4231×
        let mut l = Levels::default();
4244×
        let pol = |scope: &str, key: &str, f: &dyn Fn(&mut Policy)| {
4254×
            let mut p = Policy { scope: scope.into(), key: key.into(), ..Policy::default() };
4264×
            f(&mut p);
4274×
            ((scope.to_string(), key.to_string()), p)
4284×
        };
4291×
        l.policies.extend([
4301×
            pol("site", "", &|p| {
4311×
                p.updates = Some(false);
4321×
                p.channel = Some("stable".into());
4331×
            }),
4341×
            pol("owner", "acme", &|p| p.updates = Some(true)),
4351×
            pol("group", "1", &|p| p.channel = Some("canary".into())),
4361×
            pol("runner", "3", &|p| p.updates = Some(false)),
437
        ]);
4381×
        l.groups.push(Group { id: 1, owner: "acme".into(), name: "canary".into(), percent: None, created_ms: 0, wave: None });
4391×
        l.chosen.insert((1, 2));
4401×
        l.chosen.insert((1, 9));
441
4421×
        let site_runner = l.resolve(&runner(1, ""));
4431×
        assert!(!site_runner.updates, "off at the site");
4441×
        assert_eq!(site_runner.from, ["site", "site", "default", "default"]);
445
4461×
        let canary = l.resolve(&runner(2, "acme"));
4471×
        assert!(canary.updates && canary.channel == "canary");
4481×
        assert_eq!(canary.from[..2], ["owner acme".to_string(), "group canary".to_string()]);
449
4501×
        let held = l.resolve(&runner(3, "acme"));
4511×
        assert!(!held.updates, "the runner's own setting wins");
4521×
        assert_eq!(held.channel, "stable");
453
4541×
        assert!(l.groups_of(&runner(9, "other")).is_empty(), "an owner's group holds only its own runners");
4551×
        assert!(!Levels::default().resolve(&runner(4, "")).updates, "off unless chosen");
4561×
    }
457
458
    #[test]
4591×
    fn a_percentage_group_takes_its_share_of_the_runners_in_reach() {
4601×
        let mut l = Levels::default();
4611×
        l.groups.push(Group { id: 5, owner: String::new(), name: "farm 20%".into(), percent: Some(20), created_ms: 0, wave: None });
462500×
        let n = (1..=500).filter(|id| !l.groups_of(&runner(*id, "")).is_empty()).count();
4631×
        assert!((60..=140).contains(&n), "{n}");
4641×
    }
465
466
    #[test]
4671×
    fn a_wave_proves_a_version_by_its_jobs_or_its_time_and_an_error_stops_it() {
4688×
        let soak = |runner_id, jobs, errors, first_ms| Soak { runner_id, jobs, errors, first_ms: Some(first_ms) };
4691×
        let hour = 3_600_000;
470
        // Nobody has taken it yet: not proved.
4711×
        assert_eq!(wave_state(&[1, 2], &[], 5, 24 * hour, 0), WaveState::Soaking { runners: 2, on: 0, jobs: 0 });
472
        // One runner took it and ran 3 of its 5; the other has not updated.
4731×
        let s = [soak(1, 3, 0, 0), soak(9, 50, 0, 0)];
4741×
        assert_eq!(wave_state(&[1, 2], &s, 5, 24 * hour, hour), WaveState::Soaking { runners: 2, on: 1, jobs: 3 });
475
        // Five each on every runner that took it proves it.
4761×
        assert_eq!(wave_state(&[1, 2], &[soak(1, 5, 0, 0)], 5, 24 * hour, hour), WaveState::Proved);
477
        // Or a day since the first job there, with no error.
4781×
        assert_eq!(wave_state(&[1, 2], &[soak(1, 1, 0, 0)], 5, 24 * hour, 24 * hour), WaveState::Proved);
479
        // An error anywhere in the wave stops it, however many passed.
4801×
        assert_eq!(
4811×
            wave_state(&[1, 2], &[soak(1, 40, 0, 0), soak(2, 2, 1, 0)], 5, 24 * hour, 48 * hour),
482
            WaveState::Stopped { runner_id: 2 }
483
        );
484
        // Another wave's errors are not this one's.
4851×
        assert_eq!(wave_state(&[1], &[soak(1, 5, 0, 0), soak(2, 2, 1, 0)], 5, 24 * hour, hour), WaveState::Proved);
486
        // An empty wave holds nothing up.
4871×
        assert_eq!(wave_state(&[], &[], 5, 24 * hour, 0), WaveState::Proved);
4881×
    }
489
490
    #[test]
4911×
    fn a_runner_waits_for_the_waves_before_its_own() {
4921×
        let mut l = Levels::default();
4933×
        let group = |id, name: &str, wave| Group { id, owner: String::new(), name: name.into(), percent: None, created_ms: 0, wave };
4941×
        l.groups.extend([group(1, "first", Some(1)), group(2, "second", Some(2)), group(3, "watchers", None)]);
4951×
        l.chosen.extend([(1, 1), (2, 2), (2, 1), (3, 3)]);
4964×
        let all: Vec<Runner> = (1..=4).map(|id| runner(id, "")).collect();
4975×
        let ids = |ws: Vec<(i64, Vec<&Runner>)>| {
4989×
            ws.into_iter().map(|(w, rs)| (w, rs.iter().map(|r| r.id).collect::<Vec<_>>())).collect::<Vec<_>>()
4995×
        };
5001×
        assert_eq!(l.wave_of(&all[0]), Some(1), "its first wave");
5011×
        assert!(l.waves_before(&all[0], &all).is_empty(), "wave 1 goes first");
5021×
        assert_eq!(ids(l.waves_before(&all[1], &all)), vec![(1, vec![1])]);
5031×
        assert_eq!(l.wave_of(&all[2]), None, "a group with no wave");
5041×
        assert_eq!(ids(l.waves_before(&all[2], &all)), vec![(1, vec![1]), (2, vec![2])], "in no wave: after every wave");
5051×
        assert_eq!(ids(l.waves_before(&all[3], &all)), vec![(1, vec![1]), (2, vec![2])]);
506
        // An owner's waves gate only that owner's runners.
5071×
        l.groups.push(Group { id: 4, owner: "acme".into(), name: "acme first".into(), percent: None, created_ms: 0, wave: Some(1) });
5081×
        l.chosen.insert((4, 5));
5091×
        let mut all = all;
5101×
        all.push(runner(5, "acme"));
5111×
        all.push(runner(6, "acme"));
5121×
        assert_eq!(ids(l.waves_before(&all[5], &all)), vec![(1, vec![1, 5]), (2, vec![2])]);
5131×
        assert_eq!(ids(l.waves_before(&all[3], &all)), vec![(1, vec![1]), (2, vec![2])], "not another owner's");
5141×
    }
515
516
    #[test]
5171×
    fn windows_and_channels_read_as_written() {
5181×
        assert_eq!(parse_window("02-05"), Ok(Some((2, 5))));
5191×
        assert_eq!(parse_window(" 22:00-04:00 "), Ok(Some((22, 4))));
5201×
        assert_eq!(parse_window(""), Ok(None));
5211×
        assert!(parse_window("5").is_err() && parse_window("3-3").is_err() && parse_window("1-24").is_err());
5221×
        assert!(in_window((2, 5), 2) && in_window((2, 5), 4) && !in_window((2, 5), 5));
5231×
        assert!(in_window((22, 4), 23) && in_window((22, 4), 1) && !in_window((22, 4), 12));
5241×
        assert!(valid_channel("stable") && valid_channel("canary") && valid_channel("0.3.4"));
5251×
        assert!(!valid_channel("latest") && !valid_channel("0.3.") && !valid_channel(""));
5261×
        assert_eq!(version_of("v0.3.2"), "0.3.2");
5271×
        assert_eq!(version_of("0.3.1 (local build)"), "0.3.1");
5281×
        assert_eq!(asset_for("linux", "riscv64"), "flow-linux-riscv64");
5291×
        assert_eq!(asset_for("windows", "x86_64"), "flow-windows-x86_64.exe");
5301×
        assert_eq!(
5311×
            rolled_back_version("rolled back from flow 0.4.0 to 0.3.1: 3 starts without reaching the hub").as_deref(),
532
            Some("0.4.0")
533
        );
5341×
        assert_eq!(rolled_back_version("runs flow 0.4.0 (was 0.3.1); proven"), None);
5351×
    }
536
}