johnandersen777.bsky.social/knotless

Switch repository

Loading…

//! A runner stopping hands its work over to the one started next, rather
//! than holding it up: a deploy starts the new runner beside the old one,
//! which finishes only the runs it has going.
//!
//! Stopping, a runner first drops each run waiting that it will not run,
//! for the next to run, and writes the state file a last time. From then
//! on it writes a note of its own beside it instead, under the
//! [`dir`] the state file names, every [`BEAT`] and as anything changes:
//! the checks of the runs it still has going, the repositories with a
//! merge queue's batch among them, and what it has kept since that the
//! state file would have held: the pulls' runs that passed, the runs
//! dropped and those asked for by name to run again, as a run asked for
//! is cancelled by the stop, and the time steps took. Once its runs end it says it is done.
//!
//! A runner that starts while such a note is fresh starts beside a runner
//! still going: it leaves that one's checks and containers alone, and
//! the caches' bases it may read; it lends it the jobs its runs hold, so
//! the two take no more of the machine than one; it lands no batch of a
//! repository whose batch that one is still checking; and it takes in
//! what each note says it kept. A note that says its runner is done, or
//! one not written for [`STALE`], is of a runner gone: what it kept is
//! taken in a last time, the time its steps took with it, and the note
//! removed, and what it left going, if anything, is ended as any runner
//! that stopped otherwise leaves it.

use crate::{Asks, Dropped, Passed, State, Used};
use knotless::store;
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::time::Duration;

/// How often a runner handing over writes its note.
pub(crate) const BEAT: Duration = Duration::from_secs(2);

/// How long a note goes unwritten before its runner is taken for gone.
pub(crate) const STALE: Duration = Duration::from_secs(20);

/// What a runner handing over says of itself.
#[derive(Serialize, Deserialize, Default, Clone, Debug, PartialEq)]
pub(crate) struct Note {
    /// When it was written, in milliseconds since the epoch.
    pub beat: u64,
    /// The keys of the checks of the runs it has going, started or not.
    pub going: Vec<String>,
    /// The repositories, by `<did>/<key>`, it is checking a batch of.
    #[serde(default)]
    pub batches: Vec<String>,
    /// Whether its runs have all ended.
    #[serde(default)]
    pub done: bool,
    /// The trusted pulls' runs that passed since it handed over.
    #[serde(default)]
    pub passed: BTreeMap<String, Vec<Passed>>,
    /// The milliseconds its steps took since, by owner, of public
    /// repositories and of private ones.
    #[serde(default)]
    pub used: BTreeMap<String, Used>,
    #[serde(default)]
    pub used_private: BTreeMap<String, Used>,
    /// The workflows asked for by name to run again since, by repository
    /// and ref.
    #[serde(default)]
    pub wanted: BTreeMap<String, BTreeMap<String, Asks>>,
    /// The runs dropped since, by repository.
    #[serde(default)]
    pub dropped: BTreeMap<String, Dropped>,
}

impl Note {
    /// Whether its runner is gone: done, or not heard from for [`STALE`]
    /// as of `now`.
    pub fn gone(&self, now: u64) -> bool {
        self.done || self.beat + STALE.as_millis() as u64 <= now
    }
}

/// Where the notes of runners handing over are kept, beside the state
/// file at `state`.
pub(crate) fn dir(state: &Path) -> PathBuf {
    state.with_extension("handover")
}

/// Where the runner `instance`'s note is.
pub(crate) fn path(state: &Path, instance: &str) -> PathBuf {
    dir(state).join(format!("{instance}.json"))
}

/// Milliseconds since the epoch.
pub(crate) fn now() -> u64 {
    let since = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH);
    since.unwrap_or_default().as_millis() as u64
}

/// The notes beside the state file at `state` but `instance`'s own, by
/// where each is. One that cannot be read is passed over.
pub(crate) fn read(state: &Path, instance: &str) -> BTreeMap<PathBuf, Note> {
    let own = path(state, instance);
    let entries = std::fs::read_dir(dir(state))
        .into_iter()
        .flatten()
        .flatten();
    let notes = entries.map(|e| e.path()).filter(|p| *p != own);
    let notes = notes.filter(|p| p.extension().is_some_and(|e| e == "json"));
    notes
        .filter_map(|p| Some((p.clone(), store::read::<Note>(&p)?)))
        .collect()
}

/// What `state` has kept since `then`, the state as the runner handed
/// over: the pulls' runs that passed it did not have, and the time steps
/// took over what it had.
pub(crate) fn since(then: &State, state: &State, note: &mut Note) {
    for (repo, passed) in &state.passed {
        let had = then.passed.get(repo);
        let new = passed
            .iter()
            .filter(|p| !had.is_some_and(|h| h.contains(p)));
        let new: Vec<Passed> = new.cloned().collect();
        if !new.is_empty() {
            note.passed.insert(repo.clone(), new);
        }
    }
    note.used = more(&then.used, &state.used);
    note.used_private = more(&then.used_private, &state.used_private);
    for (repo, refs) in &state.wanted {
        for (name, asks) in refs {
            let had = then.wanted.get(repo).and_then(|r| r.get(name));
            for (wf, env) in asks {
                if had.is_none_or(|h| !h.contains_key(wf)) {
                    let refs = note.wanted.entry(repo.clone()).or_default();
                    let asks = refs.entry(name.clone()).or_default();
                    asks.insert(wf.clone(), env.clone());
                }
            }
        }
    }
    for (repo, of) in &state.dropped {
        for (uri, ats) in of {
            for (at, wfs) in ats {
                let had = then.dropped.get(repo).and_then(|o| o.get(uri)?.get(at));
                for wf in wfs.iter().filter(|wf| had.is_none_or(|h| !h.contains(*wf))) {
                    let of = note.dropped.entry(repo.clone()).or_default();
                    let ats = of.entry(uri.clone()).or_default();
                    ats.entry(at.clone()).or_default().insert(wf.clone());
                }
            }
        }
    }
}

/// What each owner used in `now` over what they had in `then`.
fn more(then: &BTreeMap<String, Used>, now: &BTreeMap<String, Used>) -> BTreeMap<String, Used> {
    let mut more = BTreeMap::new();
    for (owner, days) in now {
        let had = then.get(owner);
        for (day, ms) in days {
            let was = had.and_then(|h| h.get(day)).copied().unwrap_or_default();
            if *ms > was {
                more.entry(owner.clone())
                    .or_insert_with(Used::new)
                    .insert(*day, ms - was);
            }
        }
    }
    more
}

/// Takes into `state` what `note` says its runner kept: each pull's run
/// that passed, run dropped and workflow asked for again that neither
/// `state` has nor the note as `seen` before said, as one taken then may
/// since have run or been let go; and once it is `gone`, the time its
/// steps took. Says whether that changed anything.
pub(crate) fn take(state: &mut State, note: &Note, seen: Option<&Note>, gone: bool) -> bool {
    let mut changed = false;
    for (repo, passed) in &note.passed {
        let had = seen.and_then(|s| s.passed.get(repo));
        let kept = state.passed.entry(repo.clone()).or_default();
        for p in passed {
            if !kept.contains(p) && !had.is_some_and(|h| h.contains(p)) {
                kept.push(p.clone());
                changed = true;
            }
        }
        let over = kept.len().saturating_sub(crate::PASSED_MAX);
        kept.drain(..over);
    }
    for (repo, refs) in &note.wanted {
        let had = seen.and_then(|s| s.wanted.get(repo));
        for (name, asks) in refs {
            let had = had.and_then(|h| h.get(name));
            for (wf, env) in asks
                .iter()
                .filter(|(wf, _)| had.is_none_or(|h| !h.contains_key(*wf)))
            {
                let refs = state.wanted.entry(repo.clone()).or_default();
                let asks = refs.entry(name.clone()).or_default();
                changed |= asks.insert(wf.clone(), env.clone()).is_none();
            }
        }
    }
    for (repo, of) in &note.dropped {
        let had = seen.and_then(|s| s.dropped.get(repo));
        for (uri, ats) in of {
            for (at, wfs) in ats {
                let had = had.and_then(|h| h.get(uri)?.get(at));
                for wf in wfs.iter().filter(|wf| had.is_none_or(|h| !h.contains(*wf))) {
                    let of = state.dropped.entry(repo.clone()).or_default();
                    let ats = of.entry(uri.clone()).or_default();
                    changed |= ats.entry(at.clone()).or_default().insert(wf.clone());
                }
            }
        }
    }
    if gone {
        for (private, used) in [(false, &note.used), (true, &note.used_private)] {
            for (owner, days) in used {
                let mine = state.usage(private).entry(owner.clone()).or_default();
                for (day, ms) in days {
                    *mine.entry(*day).or_default() += ms;
                    changed = true;
                }
            }
        }
    }
    changed
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::BTreeSet;

    fn passed(tree: &str) -> Passed {
        Passed {
            tree: tree.into(),
            workflow: "test".into(),
            base: "refs/heads/main".into(),
            rkey: format!("r-{tree}"),
            pull: "queue refs/heads/main".into(),
            at: String::new(),
        }
    }

    #[test]
    fn a_note_says_what_was_kept_since() {
        let mut then = State::default();
        then.passed.insert("r".into(), vec![passed("a")]);
        then.used.insert("o".into(), Used::from([(1, 10)]));
        let mut now = State::default();
        now.passed
            .insert("r".into(), vec![passed("a"), passed("b")]);
        now.used.insert("o".into(), Used::from([(1, 15), (2, 3)]));
        now.used.insert("p".into(), Used::from([(2, 4)]));
        now.used_private.insert("o".into(), Used::from([(2, 1)]));
        let asks = |wf: &str| Asks::from([(wf.to_string(), vec![("K".into(), "v".into())])]);
        then.wanted
            .insert("r".into(), BTreeMap::from([("main".into(), asks("old"))]));
        let mut wanted = asks("old");
        wanted.extend(asks("again"));
        now.wanted
            .insert("r".into(), BTreeMap::from([("main".into(), wanted)]));
        let dropped = |wf: &str| {
            let ats = BTreeMap::from([("c".to_string(), BTreeSet::from([wf.to_string()]))]);
            Dropped::from([("refs/heads/main".to_string(), ats)])
        };
        then.dropped.insert("r".into(), dropped("then"));
        let mut both = dropped("then");
        let at = both.get_mut("refs/heads/main").unwrap();
        at.get_mut("c").unwrap().insert("now".into());
        now.dropped.insert("r".into(), both);
        let mut note = Note::default();
        since(&then, &now, &mut note);
        assert_eq!(note.passed["r"], vec![passed("b")]);
        assert_eq!(note.used["o"], Used::from([(1, 5), (2, 3)]));
        assert_eq!(note.used["p"], Used::from([(2, 4)]));
        assert_eq!(note.used_private["o"], Used::from([(2, 1)]));
        assert_eq!(note.wanted["r"]["main"], asks("again"));
        assert_eq!(note.dropped["r"], dropped("now"));

        // Taken in while its runner goes on, only the runs that passed
        // are, once each, and not again once let go; the time its steps
        // took, once it is gone.
        let mut state = State::default();
        state.used.insert("o".into(), Used::from([(1, 15)]));
        assert!(take(&mut state, &note, None, false));
        assert!(!take(&mut state, &note, None, false));
        assert_eq!(state.passed["r"], vec![passed("b")]);
        assert_eq!(state.wanted["r"]["main"], asks("again"));
        assert_eq!(state.dropped["r"], dropped("now"));
        // Run or let go since, none is taken in again.
        state.passed.clear();
        state.wanted.clear();
        state.dropped.clear();
        assert!(!take(&mut state, &note, Some(&note), false));
        assert_eq!(state.passed["r"], vec![]);
        assert!(state.wanted.is_empty() && state.dropped.is_empty());
        assert_eq!(state.used["o"], Used::from([(1, 15)]));
        assert!(take(&mut state, &note, Some(&note), true));
        assert_eq!(state.used["o"], Used::from([(1, 20), (2, 3)]));
        assert_eq!(state.used_private["o"], Used::from([(2, 1)]));
    }

    #[test]
    fn notes_beside_the_state_file() {
        let tmp = tempfile::tempdir().unwrap();
        let state = tmp.path().join("state.json");
        assert!(read(&state, "me").is_empty());
        let note = Note {
            beat: now(),
            going: vec!["k".into()],
            ..Note::default()
        };
        store::write(&path(&state, "me"), &note, false).unwrap();
        store::write(&path(&state, "them"), &note, false).unwrap();
        std::fs::write(dir(&state).join("them.tmp"), "").unwrap();
        std::fs::write(path(&state, "torn"), "{").unwrap();
        let read = read(&state, "me");
        assert_eq!(read.into_values().collect::<Vec<_>>(), vec![note.clone()]);
        // Fresh, its runner is going; not written for a while, or done,
        // it is gone.
        assert!(!note.gone(now()));
        assert!(note.gone(note.beat + STALE.as_millis() as u64));
        let done = Note { done: true, ..note };
        assert!(done.gone(done.beat));
    }
}