//! 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, /// The repositories, by `/`, it is checking a batch of. #[serde(default)] pub batches: Vec, /// 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>, /// The milliseconds its steps took since, by owner, of public /// repositories and of private ones. #[serde(default)] pub used: BTreeMap, #[serde(default)] pub used_private: BTreeMap, /// The workflows asked for by name to run again since, by repository /// and ref. #[serde(default)] pub wanted: BTreeMap>, /// The runs dropped since, by repository. #[serde(default)] pub dropped: BTreeMap, } 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 { 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::(&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 = 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, now: &BTreeMap) -> BTreeMap { 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 ¬e.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 ¬e.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 ¬e.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, ¬e.used), (true, ¬e.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, ¬e, None, false)); assert!(!take(&mut state, ¬e, 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, ¬e, Some(¬e), 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, ¬e, Some(¬e), 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"), ¬e, false).unwrap(); store::write(&path(&state, "them"), ¬e, 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![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)); } }