~ knotless Search/
↑↓ move↵ open⇧↵ all results

chadtmiller.com/knotless
Switch repository
Loading…

git that lives in your Atmosphere account

knotless.dev
1 collaborator

knotless 0.13.0 Latest 2 packages, 1 image

1
10

A stopping runner hands over to the one started next

Merged chadtmiller.com asked to merge into main from chadtmiller.com/knotless:run/handover
.changeset/runner-hands-over.mdnew+7−0
View file
@@ -0,0 +1,7 @@
---
default: minor
---
# A stopping runner hands over to the next one
On SIGTERM, knotless-run finishes the runs it has started, along with a merge-queue batch's runs and those at a running deploy's commit. It writes a note beside its state file, so a runner started beside it takes up everything else at once. The new runner leaves the old one's checks and caches alone, and takes no job the old one's runs hold.
crates/knotless-run/src/caches.rs+133−4
View file
@@ -48,6 +48,15 @@
//! newer, so the base never goes back to an older tip. Each base made so
//! is marked with that tip's generation, in [`MARKS`].
//!
//! A runner started while the one before it still has runs going, as a
//! deploy hands over to it, pins every base on disk as it opens
//! ([`Caches::pin`]): the runs it does not know of may read any of them,
//! so none goes until it unpins them, once those runs end. The runner
//! stopping makes no base meanwhile, nor removes one, so the two never
//! make the same generation or remove what the other reads: the one
//! stopping holds still ([`Caches::hold_still`]), its runs' layers going
//! as they end, whether they passed or not.
//!
//! The disk the caches are on is shared, with the site on the same host
//! among others, so the runner keeps a floor of it free: below it, before
//! a run starts, [`Caches::shed`] removes every kept run's layers and
@@ -67,7 +76,7 @@ use std::fs::Metadata;
use std::io;
use std::os::unix::fs::{FileTypeExt, MetadataExt};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use crate::executor;
@@ -105,6 +114,9 @@ pub struct Caches {
bases: Mutex<Bases>,
/// Held while kept layers are made, taken or removed.
kept: Mutex<()>,
/// Set once the runner hands over: from then on it makes no base and
/// removes none.
still: AtomicBool,
}
/// A kept run's layers, when it was kept, and the bytes they take.
@@ -150,6 +162,8 @@ struct Bases {
current: HashMap<String, u64>,
kept: HashMap<String, BTreeSet<u64>>,
readers: HashMap<(String, u64), usize>,
/// Those a runner beside this one may read, kept until unpinned.
pinned: HashSet<(String, u64)>,
next: u64,
marks: HashMap<String, (u64, u64)>,
}
@@ -183,12 +197,16 @@ impl Bases {
/// longer counted as on disk.
fn unused(&mut self, name: &str) -> Vec<u64> {
let current = self.current.get(name).copied();
let readers = &self.readers;
let (readers, pinned) = (&self.readers, &self.pinned);
let kept = self.kept.entry(name.to_string()).or_default();
let read = |g: u64| {
let base = (name.to_string(), g);
readers.contains_key(&base) || pinned.contains(&base)
};
let unused: Vec<u64> = kept
.iter()
.copied()
.filter(|g| Some(*g) != current && !readers.contains_key(&(name.to_string(), *g)))
.filter(|g| Some(*g) != current && !read(*g))
.collect();
kept.retain(|g| !unused.contains(g));
unused
@@ -221,6 +239,17 @@ pub struct Lease {
pub layers: Vec<Layer>,
}
/// The runs whose layers are under the caches at `root`, as a runner
/// stopping beside this one may have mounted them.
pub fn runs_in(root: &Path) -> Vec<String> {
let runs = std::fs::read_dir(root.join("runs"))
.into_iter()
.flatten()
.flatten();
runs.map(|r| r.file_name().to_string_lossy().into_owned())
.collect()
}
impl Caches {
/// The caches under `root`, with what runs of an earlier runner left
/// there gone, but for the layers of the runs `alive` names, whose
@@ -306,9 +335,45 @@ impl Caches {
runs: AtomicU64::new(0),
bases: Mutex::new(bases),
kept: Mutex::new(()),
still: AtomicBool::new(false),
}))
}
/// Makes no base and removes none from now on, as the runner hands
/// over to one started beside it: what its runs write goes as they
/// end, and what it would remove is left for that one.
pub fn hold_still(&self) {
self.still.store(true, Ordering::SeqCst);
}
fn still(&self) -> bool {
self.still.load(Ordering::SeqCst)
}
/// Pins every base on disk, for the runs of a runner beside this one,
/// which may read any: none goes until [`Caches::unpin`].
pub fn pin(&self) {
let mut bases = self.bases.lock().unwrap();
let on_disk = bases.kept.iter();
let pinned = on_disk.flat_map(|(name, gens)| gens.iter().map(|g| (name.clone(), *g)));
bases.pinned = pinned.collect();
}
/// Lets the bases pinned go, once no runner beside this one reads
/// them: those no run of this one's reads, and none starts from, go
/// now. With none pinned, it does nothing.
pub fn unpin(&self) {
let names: Vec<String> = {
let mut bases = self.bases.lock().unwrap();
let pinned = std::mem::take(&mut bases.pinned);
let names = bases.kept.keys().filter(|_| !pinned.is_empty());
names.cloned().collect()
};
for name in names {
self.sweep(&name);
}
}
/// Drops the caches of the repository `repo` ([`name`]), as of one
/// deleted: each base no run reads now goes, and those that runs do go
/// once they end, none being current any more.
@@ -358,6 +423,9 @@ impl Caches {
/// drops them, for the next run of each to start afresh. Says how
/// many bases and kept runs went.
pub fn shed(&self) -> Result<(usize, usize)> {
if self.still() {
return Ok((0, 0));
}
let mut runs = 0;
{
let _kept = self.kept.lock().unwrap();
@@ -370,7 +438,8 @@ impl Caches {
let mut gone = 0;
let current: Vec<(String, u64)> = bases.current.clone().into_iter().collect();
for (name, generation) in current {
if bases.readers.contains_key(&(name.clone(), generation)) {
let base = (name.clone(), generation);
if bases.readers.contains_key(&base) || bases.pinned.contains(&base) {
continue;
}
bases.current.remove(&name);
@@ -454,6 +523,9 @@ impl Caches {
/// Removes the layers kept of each of `runs`, as of runs a push can no
/// longer take, and those past their bounds, as [`Caches::bound`].
pub fn forget(&self, runs: &[String], budget: u64) -> Result<()> {
if self.still() {
return Ok(());
}
self.forgetting(runs, budget, now())
}
@@ -578,6 +650,9 @@ impl Caches {
/// Removes the bases of `name` that are no longer used.
fn sweep(&self, name: &str) {
if self.still() {
return;
}
let unused = self.bases.lock().unwrap().unused(name);
for generation in unused {
let _ = remove(&self.base(name, generation));
@@ -606,6 +681,9 @@ impl Lease {
/// run `run`'s, for a push that takes its check to take, within the
/// bounds of kept layers and `budget` bytes for them all.
pub async fn keep(self, run: &str, budget: u64) -> Result<()> {
if self.caches.still() {
return Ok(());
}
let (caches, run) = (self.caches.clone(), run.to_string());
let kept = move || caches.keeping(self, &run, budget, now());
tokio::task::spawn_blocking(kept).await?
@@ -633,6 +711,9 @@ impl Lease {
/// [`Lease::promote`], or, given the tip a pull merged with, `at`,
/// [`Lease::advance`].
async fn promoting(self, max: u64, at: Option<u64>) -> Result<(Vec<String>, Vec<String>)> {
if self.caches.still() {
return Ok((Vec::new(), Vec::new()));
}
let (caches, layers) = (self.caches.clone(), self.layers.clone());
let building = caches.clone();
let made = tokio::task::spawn_blocking(move || -> Result<Vec<_>> {
@@ -1450,4 +1531,52 @@ mod tests {
// Nothing left to shed but the bases no run reads now.
assert_eq!(caches.shed().unwrap(), (2, 0));
}
#[tokio::test]
async fn pinned_bases_stay_until_unpinned() {
let root = tempfile::tempdir().unwrap();
let wanted = names(&[("t", "/a")]);
{
let before = Caches::open(root.path(), &[], &[]).unwrap();
let main = before.lease(&wanted).unwrap();
write(&main.layers[0].upper.join("x"), "x");
main.promote(1 << 30).await.unwrap();
}
// Opened beside a runner whose runs may read the base on disk, it
// keeps that base as its own runs move on from it, and sheds it
// not, until unpinned.
std::fs::create_dir_all(root.path().join("runs/theirs")).unwrap();
assert_eq!(runs_in(root.path()), ["theirs"]);
let caches = Caches::open(root.path(), &["theirs".into()], &[]).unwrap();
assert!(root.path().join("runs/theirs").exists());
caches.pin();
let main = caches.lease(&wanted).unwrap();
let theirs = main.layers[0].lower.clone();
write(&main.layers[0].upper.join("y"), "y");
main.promote(1 << 30).await.unwrap();
assert!(theirs.exists());
// Shed, its own base goes, as no run reads it, but not theirs.
assert_eq!(caches.shed().unwrap(), (1, 0));
assert!(theirs.exists());
caches.unpin();
assert!(!theirs.exists());
// Held still, as it hands over, it makes no base and removes none:
// what its run wrote goes as the run ends.
caches.hold_still();
let main = caches.lease(&wanted).unwrap();
let (read, upper) = (main.layers[0].lower.clone(), main.layers[0].upper.clone());
write(&upper.join("z"), "z");
main.promote(1 << 30).await.unwrap();
let next = caches.lease(&wanted).unwrap();
assert_eq!(next.layers[0].lower, read);
next.keep("run", 1 << 30).await.unwrap();
assert!(!caches.kept_dir("run").exists());
let next = caches.lease(&wanted).unwrap();
assert_eq!(next.advance(1 << 30, 1).await.unwrap(), (vec![], vec![]));
assert_eq!(caches.shed().unwrap(), (0, 0));
caches.forget(&["run".into()], 0).unwrap();
assert!(read.exists());
assert!(!upper.exists());
}
}
crates/knotless-run/src/handover.rsnew+327−0
View file
@@ -0,0 +1,327 @@
//! 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));
}
}
crates/knotless-run/src/lib.rs+297−45
View file
@@ -85,6 +85,7 @@
pub mod caches;
pub mod check;
pub mod executor;
mod handover;
pub mod live;
mod prune;
mod queue;
@@ -518,7 +519,7 @@ pub fn parse_args(args: &[String]) -> Result<Options, String> {
/// A repository not here yet is being watched for the first time: what its
/// refs, pulls and requests are then is where watching starts, not
/// something to run.
#[derive(Serialize, Deserialize, Default, Debug)]
#[derive(Serialize, Deserialize, Default, Clone, Debug)]
struct State {
repos: BTreeMap<String, BTreeMap<String, u64>>,
#[serde(default)]
@@ -1040,6 +1041,43 @@ pub struct Runner {
/// Counts the runs that have ended and filed how, for those waiting
/// on what they name `after` to look again.
ended: tokio::sync::watch::Sender<u64>,
/// This runner, among those that hand over to each other
/// ([`handover`]).
instance: String,
/// The state as it was when this runner handed over, once it has: its
/// state goes into its note from then on, not the state file.
handing: Mutex<Option<State>>,
/// The notes of the runners stopping beside this one, by where each
/// is, as last read.
before: Mutex<BTreeMap<PathBuf, handover::Note>>,
/// The repositories those runners are checking a batch of.
their_batches: Mutex<HashSet<String>>,
/// The runs not started that a stop has yet to drop or keep.
deciding: Arc<tokio::sync::watch::Sender<usize>>,
}
/// A run not started yet, counted in [`Runner::deciding`] until it starts
/// or ends, or a stop keeps it, to start here.
struct Deciding(Option<Arc<tokio::sync::watch::Sender<usize>>>);
impl Deciding {
fn new(deciding: &Arc<tokio::sync::watch::Sender<usize>>) -> Deciding {
deciding.send_modify(|n| *n += 1);
Deciding(Some(deciding.clone()))
}
/// It is decided: started, or kept to start.
fn done(&mut self) {
if let Some(deciding) = self.0.take() {
deciding.send_modify(|n| *n -= 1);
}
}
}
impl Drop for Deciding {
fn drop(&mut self) {
self.done();
}
}
/// A run going of a workflow whose runs finish, noted in
@@ -1068,6 +1106,10 @@ impl Drop for Finishing<'_> {
/// A run going, or waiting to.
struct Going {
/// The key of its check.
rkey: String,
/// Whether it is a merge queue's batch's.
batch: bool,
/// The commit it is of, and for a pull the base's too.
at: String,
/// The pull it is of.
@@ -1144,6 +1186,7 @@ const CUT_OFF: &str = "as the runner stopped, past what a stop waits for";
/// A run started: its check, under the key it is filed by, its place in
/// line, and what says it is cancelled.
struct Job {
deciding: Deciding,
check: Check,
rkey: String,
rank: Rank,
@@ -1223,12 +1266,28 @@ impl Runner {
.unwrap_or_else(|| ctx.env.cache_dir.join("run").join("caches"));
// No run of this one's is going yet: a container of its own is
// left by one that stopped otherwise, and so are the overlays it
// mounted.
// mounted. Not while a runner stopping beside it has runs going,
// whose containers, overlays and layers those are: they stay, and
// what it leaves goes as it ends, or at the next start.
let now = handover::now();
let beside = handover::read(&state_path, "")
.values()
.any(|n| !n.gone(now));
let me = &reporter.identity.did;
opts.executor
.remove_left(me, &Mutex::default(), opts.work.as_deref())
.await;
let alive = opts.executor.clean_caches().await;
let alive = if beside {
caches::runs_in(&root)
} else {
opts.executor
.remove_left(me, &Mutex::default(), opts.work.as_deref())
.await;
opts.executor.clean_caches().await
};
static STARTED: AtomicU64 = AtomicU64::new(0);
let instance = format!(
"{}-{now}-{}",
std::process::id(),
STARTED.fetch_add(1, Ordering::Relaxed)
);
let runner = Runner {
caches: caches::Caches::open(&root, &alive, &kept)?,
appview: live::Appview::new(&ctx, &reporter.identity),
@@ -1259,7 +1318,16 @@ impl Runner {
finishing: tokio::sync::watch::Sender::new(Vec::new()),
sessions: Mutex::new(HashSet::new()),
ended: tokio::sync::watch::Sender::new(0),
instance,
handing: Mutex::new(None),
before: Mutex::new(BTreeMap::new()),
their_batches: Mutex::new(HashSet::new()),
deciding: Arc::new(tokio::sync::watch::Sender::new(0)),
};
if beside {
runner.caches.pin();
}
runner.heed();
Ok((Arc::new(runner), watches))
}
@@ -1271,6 +1339,7 @@ impl Runner {
if *self.stopping.borrow() {
return Ok(());
}
self.heed();
watch.at("reading the repository");
let snap = watch.remote.snapshot(&self.ctx).await?;
let me = &self.reporter.identity.did;
@@ -1306,7 +1375,9 @@ impl Runner {
watch.at("running pulls");
self.pulls(watch, &repo, &snap).await?;
}
if self.opts.queue && !watch.pulls_only {
// A runner beside this one checking a batch lands it, or not:
// the queue goes on from there once it is gone.
if self.opts.queue && !watch.pulls_only && !self.checking_theirs(watch) {
watch.at("reading the queue");
self.queues(watch, &repo, &snap).await?;
}
@@ -1434,7 +1505,7 @@ impl Runner {
let stopping = *self.stopping.borrow();
let wanted = state.wanted.get_mut(&watch.id()).filter(|_| !stopping);
wanted.map(|refs| refs.remove(&moved.name));
store::write(&self.state_path, &*state, false)?;
self.save(&state)?;
}
Ok(())
}
@@ -1642,7 +1713,7 @@ impl Runner {
None => !before.contains_key(&p.pull),
};
let stale: Vec<String> = passed.filter(gone).map(Passed::rkey).collect();
store::write(&self.state_path, &*state, false)?;
self.save(&state)?;
drop(state);
self.caches.forget(&stale, self.opts.cache_max)
}
@@ -1742,7 +1813,7 @@ impl Runner {
state
.reruns
.insert(id, newest.unwrap_or_default().max(since));
store::write(&self.state_path, &*state, false)
self.save(&state)
}
/// Where the writes in the space of `watch`, a private repository,
@@ -1830,7 +1901,7 @@ impl Runner {
let fired = state.fired.entry(watch.id()).or_default();
fired.retain(|name, _| scheduled.crons.contains_key(name));
state.schedules.insert(watch.id(), scheduled);
store::write(&self.state_path, &*state, false)
self.save(&state)
}
/// The commit `watch`'s schedules were last read at.
@@ -1876,7 +1947,7 @@ impl Runner {
}
}
if !rung.is_empty() {
store::write(&self.state_path, &*state, false)?;
self.save(&state)?;
}
drop(state);
let doors = self.doors.lock().unwrap();
@@ -1904,7 +1975,7 @@ impl Runner {
let refs = state.wanted.entry(watch.id()).or_default();
let asks = refs.entry(subject.ref_name.clone()).or_default();
asks.entry(wf.name.clone()).or_insert_with(|| env.clone());
store::write(&self.state_path, &*state, false)
self.save(&state)
}
/// The refs in `snap` that point past the generation last run, or at
@@ -1925,7 +1996,7 @@ impl Runner {
.map(|r| (r.value.name.clone(), r.value.generation))
.collect();
state.repos.insert(watch.id(), first.clone());
store::write(&self.state_path, &*state, false)?;
self.save(&state)?;
if !watch.discovered {
return Ok(Vec::new());
}
@@ -1938,7 +2009,7 @@ impl Runner {
let wanted = state.wanted.entry(watch.id()).or_default();
wanted.retain(|name, _| live.iter().any(|r| r.name == *name));
let wanted = wanted.clone();
store::write(&self.state_path, &*state, false)?;
self.save(&state)?;
drop(state);
let moved = |r: &RefRecord| {
let past = seen.get(&r.name).is_none_or(|g| *g < r.generation);
@@ -2132,6 +2203,8 @@ impl Runner {
let (sender, why) = tokio::sync::watch::channel(None);
let asked = subject.asked.contains_key(&wf.name);
let going = Going {
rkey: rkey.clone(),
batch: batch.is_some(),
at: subject.at(),
pull: subject.pull.as_ref().map(|p| p.uri.clone()),
asked,
@@ -2176,6 +2249,7 @@ impl Runner {
let (runner, watch, git) = (self.clone(), watch.clone(), git.clone());
let subject = subject.clone();
let job = Job {
deciding: Deciding::new(&self.deciding),
check,
rkey,
rank,
@@ -2227,6 +2301,7 @@ impl Runner {
job: Job,
) -> Result<String> {
let Job {
mut deciding,
mut check,
rkey,
rank,
@@ -2234,12 +2309,14 @@ impl Runner {
mut cancelled,
} = job;
let rkey = rkey.as_str();
// Stopping, one not started yet is dropped, to run on the next start.
// Stopping, one not started yet is dropped, to run on the next
// start, or kept, to run here.
let of = format!("{}\n{}\n{}", watch.id(), subject.keyed(), subject.at());
let batch = subject.batch.is_some();
let waits = self.waits(rank, watch, subject, wf, on_ref, &mut cancelled);
let waited = tokio::select! {
biased;
_ = self.dropping(&of) => {
_ = self.dropping(&of, batch, &mut deciding) => {
telemetry::run("run dropped, to run on the next start", watch, subject, &wf.name, None, None);
return self.drop_run(watch, subject, wf, rkey, check).await;
}
@@ -2260,6 +2337,7 @@ impl Runner {
return self.never_started(watch, rkey, check, why).await;
}
};
deciding.done();
// Its turn on the ref is kept until its check is filed as ended,
// so the next run of a deploy files after it.
let _on_ref = on_ref;
@@ -2409,7 +2487,7 @@ impl Runner {
});
let over = kept.len().saturating_sub(PASSED_MAX);
kept.drain(..over);
store::write(&self.state_path, &*state, false)?;
self.save(&state)?;
drop(state);
self.caches.forget(&superseded, self.opts.cache_max)
}
@@ -2627,7 +2705,7 @@ impl Runner {
let allowance = self.allowance(owner, private);
let now = standing(spent(used, today()), allowance);
let record = minutes_record(owner, used, allowance, today());
store::write(&self.state_path, &*state, false)?;
self.save(&state)?;
let mut published = self.published.lock().unwrap();
let last = published.get(owner).copied();
if private || !due(last, now) {
@@ -2755,7 +2833,7 @@ impl Runner {
let mut state = self.state.lock().unwrap();
state.discovered.remove(&watch.label);
state.schedules.remove(&watch.id());
store::write(&self.state_path, &*state, false)?;
self.save(&state)?;
telemetry::left(&watch.label);
Ok(())
}
@@ -2808,7 +2886,7 @@ impl Runner {
state.discovered.remove(&format!("{did}/{was}"));
}
state.discovered.insert(label.clone());
store::write(&self.state_path, &*state, false)?;
self.save(&state)?;
}
telemetry::found(&label);
report(&watch, self.recover(&watch).await);
@@ -2853,7 +2931,7 @@ impl Runner {
}
let mut state = self.state.lock().unwrap();
state.invited = page.cursor.or(cursor);
store::write(&self.state_path, &*state, false)?;
self.save(&state)?;
if page.invites.is_empty() {
return Ok(());
}
@@ -2903,7 +2981,7 @@ impl Runner {
let of = state.dropped.entry(watch.id()).or_default();
let at = of.entry(subject.keyed()).or_default();
at.entry(subject.at()).or_default().insert(wf.name.clone());
store::write(&self.state_path, &*state, false)?;
self.save(&state)?;
}
self.ask_again(watch, subject, wf)?;
self.never_started(watch, rkey, check, WAITED).await
@@ -2911,27 +2989,145 @@ impl Runner {
/// Waits until a run not started yet, at `of` (its repository, ref or
/// pull, and what that is at), is to be dropped: once the runner is
/// stopping, and no run going whose runs finish, a deploy, is at the
/// same place. A deploy may wait for its commit's checks, so those are
/// kept until it ends.
async fn dropping(&self, of: &str) {
/// stopping, unless it is a `batch`'s, or a run going whose runs
/// finish, a deploy, is at the same place, which may wait for its
/// commit's checks. Those are kept, `deciding` so, to run here: the
/// runner handed over to knows them as this one's.
async fn dropping(&self, of: &str, batch: bool, deciding: &mut Deciding) {
let mut stopping = self.stopping.subscribe();
let mut finishing = self.finishing.subscribe();
loop {
let kept = finishing.borrow_and_update().iter().any(|f| f == of);
if *stopping.borrow_and_update() && !kept {
return;
let _ = stopping.wait_for(|stopping| *stopping).await;
let kept = self.finishing.borrow().iter().any(|f| f == of);
if !batch && !kept {
return;
}
deciding.done();
std::future::pending::<()>().await
}
/// Writes `state`: to the state file, or once this runner has handed
/// over, into its note, as what it kept since.
fn save(&self, state: &State) -> Result<()> {
let handing = self.handing.lock().unwrap();
match handing.as_ref() {
None => store::write(&self.state_path, state, false),
Some(then) => self.note(then, state, false),
}
}
/// Writes this runner's note: the runs it has going, and what `state`
/// has kept since `then`, as it handed over; `done` once they have
/// all ended.
fn note(&self, then: &State, state: &State, done: bool) -> Result<()> {
let mut note = handover::Note {
beat: handover::now(),
done,
..Default::default()
};
{
let going = self.going.lock().unwrap();
note.going = going.values().map(|g| g.rkey.clone()).collect();
let batches = going.iter().filter(|(_, g)| g.batch);
let repos = batches.filter_map(|(id, _)| id.split('\n').next());
let repos: BTreeSet<&str> = repos.collect();
note.batches = repos.into_iter().map(str::to_string).collect();
}
handover::since(then, state, &mut note);
let path = handover::path(&self.state_path, &self.instance);
store::write(&path, &note, false)
}
/// Writes this runner's note again, once it has handed over, for the
/// runner beside it to know it goes on, or that it is `done`.
fn beat(&self, done: bool) {
let state = self.state.lock().unwrap();
let handing = self.handing.lock().unwrap();
if let Some(then) = handing.as_ref() {
quietly("writing its handover note", self.note(then, &state, done));
}
}
/// Hands over to the runner started next: writes the state file a last
/// time, and from then on its note in its place, making no cache's
/// base and removing none.
fn hand_over(&self) {
let state = self.state.lock().unwrap();
quietly(
"writing its state",
store::write(&self.state_path, &*state, false),
);
self.caches.hold_still();
let then = state.clone();
quietly("writing its handover note", self.note(&then, &state, false));
*self.handing.lock().unwrap() = Some(then);
telemetry::handed_over(self.going.lock().unwrap().len());
}
/// Reads the notes of the runners stopping beside this one: takes in
/// what each kept, and lends them the jobs their runs hold. Once the
/// last is gone, the caches' bases they may have read may go, and
/// every door is rung, for what they held up to go on.
fn heed(&self) {
let now = handover::now();
let notes = handover::read(&self.state_path, &self.instance);
let mut alive = BTreeMap::new();
let mut changed = false;
{
let before = self.before.lock().unwrap();
let mut state = self.state.lock().unwrap();
for (path, note) in notes {
let gone = note.gone(now);
changed |= handover::take(&mut state, &note, before.get(&path), gone);
if gone {
telemetry::taken_over(&path, note.done);
let _ = std::fs::remove_file(&path);
} else {
alive.insert(path, note);
}
}
tokio::select! {
_ = stopping.changed() => {}
_ = finishing.changed() => {}
if changed {
quietly("writing its state", self.save(&state));
}
}
// Lent a job for each run going, those waiting for one there too:
// they take it as the others end, and the two take no more than
// `--jobs` between them.
let held: usize = alive.values().map(|n| n.going.len()).sum();
let batches = alive.values().flat_map(|n| n.batches.iter().cloned());
*self.their_batches.lock().unwrap() = batches.collect();
self.turns.lend(held.min(self.opts.jobs));
let last = alive.is_empty();
let was = std::mem::replace(&mut *self.before.lock().unwrap(), alive);
if last {
self.caches.unpin();
}
// What one kept is run now, and once the last is gone, what it held
// up goes on.
if changed || (last && !was.is_empty()) {
for door in self.doors.lock().unwrap().values() {
door.notify_one();
}
}
}
/// The checks of the runs that the runners beside this one have going.
fn theirs(&self) -> HashSet<String> {
let before = self.before.lock().unwrap();
before.values().flat_map(|n| n.going.clone()).collect()
}
/// Whether a runner beside this one is checking a batch of `watch`'s.
fn checking_theirs(&self, watch: &Watch) -> bool {
self.their_batches.lock().unwrap().contains(&watch.id())
}
/// Removes the containers of this runner's no run of it has, as one
/// that leaked, or that a runner before it left, does.
/// that leaked, or that a runner before it left, does: not while a
/// runner beside it still has runs going, whose containers are its
/// account's too.
async fn remove_left(&self) {
if !self.before.lock().unwrap().is_empty() {
return;
}
let me = &self.reporter.identity.did;
let work = self.opts.work.as_deref();
self.opts
@@ -2960,22 +3156,43 @@ impl Runner {
cancel(&going.cancel, STOPPED);
}
}
if tokio::time::timeout(self.opts.stop_wait, self.idle())
// What is dropped is in the state file the next runner reads, and
// what is kept in this one's note.
let mut deciding = self.deciding.subscribe();
let _ = deciding.wait_for(|n| *n == 0).await;
self.hand_over();
if tokio::time::timeout(self.opts.stop_wait, self.beating(self.idle()))
.await
.is_err()
{
for going in self.going.lock().unwrap().values() {
cancel(&going.cancel, CUT_OFF);
}
self.idle().await;
self.beating(self.idle()).await;
}
self.beat(true);
}
/// Awaits `until`, writing this runner's note each [`handover::BEAT`]
/// meanwhile.
async fn beating(&self, until: impl Future<Output = ()>) {
let mut until = std::pin::pin!(until);
loop {
tokio::select! {
() = &mut until => return,
() = tokio::time::sleep(handover::BEAT) => self.beat(false),
}
}
}
/// Ends each check of `watch` this account left queued or running:
/// whatever ran it stopped before it finished.
/// whatever ran it stopped before it finished. Not those of the runs a
/// runner stopping beside this one has going: it finishes them.
async fn recover(&self, watch: &Watch) -> Result<()> {
let prefix = keys::check_prefix(watch.remote.did(), &watch.key);
for (rkey, mut check) in self.reporter.running(&watch.remote, &prefix).await? {
let theirs = self.theirs();
let running = self.reporter.running(&watch.remote, &prefix).await?;
for (rkey, mut check) in running.into_iter().filter(|(k, _)| !theirs.contains(k)) {
let ended = ["finished", "started"][usize::from(check.status == status::QUEUED)];
check.status = status::ERROR.into();
check.error = Some(format!("the runner stopped before this run {ended}"));
@@ -3714,14 +3931,17 @@ pub async fn watch(runner: Arc<Runner>, watches: Vec<Watch>) -> Result<()> {
for watch in watches {
runner.open_door(watch);
}
// Each goes until the runner stops: then it hands over, and the
// runner started next does what these do.
if runner.opts.discover {
tokio::spawn(discover(runner.clone()));
until_stopped(&runner, discover(runner.clone()));
}
tokio::spawn(sweep(runner.clone()));
tokio::spawn(tidy(runner.clone()));
tokio::spawn(schedules(runner.clone()));
until_stopped(&runner, sweep(runner.clone()));
until_stopped(&runner, tidy(runner.clone()));
until_stopped(&runner, schedules(runner.clone()));
until_stopped(&runner, heeding(runner.clone()));
if let Some(appview) = runner.appview.clone() {
tokio::spawn(wakes(runner.clone(), appview));
until_stopped(&runner, wakes(runner.clone(), appview));
}
let followed = tokio::select! {
() = follow(&runner) => Ok(()),
@@ -3732,6 +3952,27 @@ pub async fn watch(runner: Arc<Runner>, watches: Vec<Watch>) -> Result<()> {
followed
}
/// Spawns `work`, to go until `runner` stops.
fn until_stopped(runner: &Runner, work: impl Future<Output = ()> + Send + 'static) {
let mut stopping = runner.stopping.subscribe();
tokio::spawn(async move {
tokio::select! {
() = work => {}
_ = stopping.wait_for(|stopping| *stopping) => {}
}
});
}
/// Reads the notes of the runners stopping beside this one each
/// [`handover::BEAT`], for the jobs they give back and for when they are
/// gone.
async fn heeding(runner: Arc<Runner>) {
loop {
tokio::time::sleep(handover::BEAT).await;
runner.heed();
}
}
/// Reconciles `watch` each time its door rings. Rings while it runs are
/// one: the next read answers them all. One that fails, as when its PDS
/// does, rings again a while after, at most a minute: what it was to run
@@ -3935,6 +4176,11 @@ fn report(watch: &Watch, done: Result<()>) {
}
/// Says what went wrong with the work of `label`, if anything did.
/// Logs `done` only if it failed: for what is done every few seconds.
fn quietly(label: &str, done: Result<()>) {
let _ = done.inspect_err(|err| telemetry::failed(label, err));
}
fn noted(label: &str, done: Result<()>) {
match done {
Ok(()) => telemetry::succeeded(label),
@@ -4338,6 +4584,12 @@ fn vouched_for(subject: &Subject) -> bool {
mod tests {
use super::*;
#[test]
fn quietly_says_only_what_failed() {
quietly("a test", Ok(()));
quietly("a test", Err(anyhow!("it failed")));
}
// On paused time, so the 5ms tick always comes before the 30ms pass:
// on a busy host both could be due at once, and select! take either.
#[tokio::test(start_paused = true)]
crates/knotless-run/src/telemetry.rs+22−0
View file
@@ -269,6 +269,28 @@ pub(crate) fn pruned(label: &str, gone: usize) {
telemetry::note(Level::INFO, "checks pruned", &fields);
}
/// Logs that the runner handed over to the one started next, with
/// `going` runs left to finish.
pub(crate) fn handed_over(going: usize) {
let fields = Fields {
count: Some(going as u64),
..Fields::default()
};
telemetry::note(Level::INFO, "handed over", &fields);
}
/// Logs that the runner whose note is at `note`, stopping beside this
/// one, is gone: `done`, or not heard from.
pub(crate) fn taken_over(note: &std::path::Path, done: bool) {
let source = note.display().to_string();
let fields = Fields {
source: Some(&source),
reason: Some(["not heard from", "done"][usize::from(done)]),
..Fields::default()
};
telemetry::note(Level::INFO, "runner beside it gone", &fields);
}
#[cfg(test)]
mod tests {
use super::*;
crates/knotless-run/src/turns.rs+32−1
View file
@@ -20,6 +20,10 @@
//! run to start, so a burst of one owner's runs goes one by one between
//! others' rather than all before them. One owner's runs go in the order
//! they were started.
//!
//! Jobs can be lent, to a runner stopping beside this one that still has
//! runs going ([`Turns::lend`]): those are the machine's as much as this
//! one's, so the two never take more than `--jobs` between them.
use std::collections::{BTreeMap, HashMap, HashSet};
use std::sync::{Arc, Mutex};
@@ -68,6 +72,8 @@ pub struct Turns {
struct Board {
jobs: usize,
others: usize,
/// Of the jobs, those lent to a runner beside this one.
lent: usize,
held: HashSet<String>,
owners: HashMap<String, usize>,
waiting: BTreeMap<Place, Want>,
@@ -92,7 +98,8 @@ impl Board {
/// that can start counted as started, and the keys those that cannot
/// wait for kept for them.
fn starts(&self, place: &Place, want: &Want) -> bool {
let mut jobs = (self.jobs, self.others);
let lent = |n: usize| n.saturating_sub(self.lent);
let mut jobs = (lent(self.jobs), lent(self.others));
let mut ahead = HashMap::new();
let mut claimed = HashSet::new();
for (theirs, w) in self.waiting.range(..place) {
@@ -162,6 +169,7 @@ impl Turns {
board: Mutex::new(Board {
jobs,
others,
lent: 0,
held: HashSet::new(),
owners: HashMap::new(),
waiting: BTreeMap::new(),
@@ -171,6 +179,14 @@ impl Turns {
})
}
/// Lends `jobs` of them, in place of those lent before, to runs that
/// are not this runner's: none of its own starts in them until they
/// are given back, by lending fewer.
pub fn lend(&self, jobs: usize) {
self.board.lock().unwrap().lent = jobs;
self.changed.notify_waiters();
}
/// A job and `keys`, once the run ranked `rank`, `whose`, may have
/// them. Dropped while it waits, it leaves the line.
pub async fn take(self: Arc<Self>, rank: Rank, whose: Whose, mut keys: Vec<String>) -> Turn {
@@ -294,6 +310,21 @@ mod tests {
waiting.as_mut().now_or_never()
}
#[test]
fn lent_jobs_are_not_taken() {
let turns = Turns::new(2, 2);
turns.lend(1);
let first = ready(&mut wait(&turns, rank(Class::Main, 1), &[])).unwrap();
let mut second = wait(&turns, rank(Class::Main, 2), &[]);
assert!(ready(&mut second).is_none());
// More lent than there are takes none, and given back, they go.
turns.lend(5);
drop(first);
assert!(ready(&mut second).is_none());
turns.lend(0);
assert!(ready(&mut second).is_some());
}
#[test]
fn main_first_then_oldest() {
let turns = Turns::new(1, 1);
crates/knotless-run/tests/run.rs+132−34
View file
@@ -1432,8 +1432,8 @@ fn a_workflow_runs_after_others_holding_no_job() {
/// A stop waits only for the runs whose steps have started, a deploy's
/// among them: those waiting their turn are dropped, their checks filed
/// as cancelled, not deleted, and run when the runner starts again, and
/// only they. Those
/// of a going deploy's commit are kept until it ends.
/// only they. Those of a going deploy's commit are kept, and run once it
/// ends.
#[test]
fn a_stop_drops_what_has_not_started() {
let rt = tokio::runtime::Runtime::new().unwrap();
@@ -1555,9 +1555,9 @@ fn a_stop_drops_what_has_not_started() {
});
assert_eq!(of("refs/heads/main", "then")["status"], "queued");
// Stopped, it cancels the pull's run and reads no more, but waits
// for the deploy going, and keeps the push's run of its commit, which
// a deploy may wait for, until the deploy ends.
// Stopped, it drops the pull's run and reads no more, but waits for
// the deploy going, and keeps the push's run of its commit, which a
// deploy may wait for, to run here once the deploy ends.
let stopping = rt.spawn({
let run = run.clone();
async move { run.stop().await }
@@ -1568,45 +1568,143 @@ fn a_stop_drops_what_has_not_started() {
assert!(!stopping.is_finished());
assert_eq!(of("refs/heads/main", "then")["status"], "queued");
std::fs::write(go.join(&main), "").unwrap();
wait_until(|| ran("thens") == format!("{main} push\n"));
assert!(!stopping.is_finished());
std::fs::write(go.join("then"), "").unwrap();
promptly(&rt, stopping);
assert_eq!(of("refs/heads/main", "deploy")["status"], "success");
assert_eq!(of("refs/heads/main", "then")["status"], "cancelled");
assert_eq!(ran("thens"), "");
assert_eq!(of("refs/heads/main", "then")["status"], "success");
// Started again, it runs what it dropped, and not the deploy that
// finished: one of the two takes the job. Stopped while that one goes,
// it waits for it, an ordinary run, and drops the other again.
// Started again, it runs what it dropped, and not what ran.
let (run, watch) = start();
rt.block_on(run.reconcile(&watch)).unwrap();
let (push, pulled) = (format!("{main} push\n"), format!("{topic} {pull}\n"));
wait_until(|| [&push, &pulled].contains(&&ran("thens")));
let (first, second) = match ran("thens") == push {
true => ("refs/heads/main", pull.as_str()),
false => (pull.as_str(), "refs/heads/main"),
rt.block_on(run.idle());
assert_eq!(of(&pull, "then")["status"], "success");
assert_eq!(ran("thens"), format!("{main} push\n{topic} {pull}\n"));
assert_eq!(ran("deploys"), format!("{main}\n"));
}
/// A runner stopping hands over to the one started beside it: the new one
/// leaves the old one's run going, and its check, alone, lends it the
/// job that run holds, and runs what the old one dropped once that job
/// is given back. The old one's note goes once it is done.
#[test]
fn a_stop_hands_over_to_the_next_runner() {
let rt = tokio::runtime::Runtime::new().unwrap();
let mut owner = Sandbox::new();
let mut runner = Sandbox::new();
let base = appview(&rt, &Sandbox::new());
owner.set("KNOTLESS_APPVIEW", &base);
let mine = owner.account.handle.clone();
owner.set("KNOTLESS_IDENTIFIER", &mine);
runner.set("KNOTLESS_APPVIEW", &base);
let handle = runner.account.handle.clone();
runner.set("KNOTLESS_IDENTIFIER", &handle);
let work = owner.work("relay", 1);
owner.git(&work, &["remote", "add", "origin", &owner.url("relay")]);
owner.git(&work, &["push", "-q", "origin", "main"]);
let opts = knotless_run::Options {
jetstream: "ws://127.0.0.1:9".into(),
repos: vec![format!("{}/relay", owner.account.handle)],
pulls_only: Vec::new(),
discover: false,
executor: Executor::Shell,
secrets: None,
secrets_key: None,
state: Some(runner.path("state.json")),
work: Some(runner.path("work")),
step_timeout: Duration::from_secs(60),
run_timeout: None,
jobs: 1,
pull_jobs: 1,
owner_jobs: 0,
allowance: None,
private_allowance: None,
unmetered: Vec::new(),
trust: Vec::new(),
cache_max: 1 << 30,
disk_floor: 0,
caches: Some(runner.path("caches")),
sweep: Duration::from_secs(1),
stop_wait: Duration::from_secs(60),
pulls: false,
queue: false,
batch: 8,
depth: 1,
notice: Duration::from_secs(30),
review_wait: Duration::from_secs(30 * 60),
poll: Duration::from_secs(10),
paused: None,
keep_checks: None,
prune_grace: Duration::ZERO,
prune_every: Duration::ZERO,
watchdog: Duration::from_secs(300),
};
let start = || {
let started = knotless_run::Runner::start(opts.clone(), common::ctx(&runner));
rt.block_on(started).unwrap()
};
wait_until(|| of(second, "then")["status"] == "queued");
let (old, mut watches) = start();
let watch = watches.remove(0);
let key = watch.key.clone();
rt.block_on(old.reconcile(&watch)).unwrap();
// `block` holds the one job until `go/block` is there; main's run
// waits behind it, and notes that it ran.
let go = runner.path("go");
std::fs::create_dir_all(&go).unwrap();
let dir = work.join(".knotless/workflows");
std::fs::create_dir_all(&dir).unwrap();
let block = format!(
"on: {{push: {{refs: [refs/heads/block]}}}}\nsteps:\n - run: |\n touch {go}/started-block\n until [ -f {go}/block ]; do sleep 0.1; done\n",
go = go.display()
);
let then = format!(
"on: {{push: {{refs: [refs/heads/main]}}}}\nsteps:\n - run: echo $KNOTLESS_RUN_SHA >> {go}/thens\n",
go = go.display()
);
std::fs::write(dir.join("block.yml"), block).unwrap();
std::fs::write(dir.join("then.yml"), then).unwrap();
owner.git(&work, &["add", "."]);
owner.git(&work, &["commit", "-q", "-m", "workflows"]);
owner.git(&work, &["push", "-q", "origin", "HEAD:block"]);
rt.block_on(old.reconcile(&watch)).unwrap();
wait_until(|| go.join("started-block").exists());
owner.git(&work, &["push", "-q", "origin", "main"]);
let main = owner.git(&work, &["rev-parse", "HEAD"]);
rt.block_on(old.reconcile(&watch)).unwrap();
let of = |r: &str, wf: &str| check_on(&runner, &owner, &key, r, wf);
wait_until(|| of("refs/heads/main", "then")["status"] == "queued");
// Stopped, it drops main's run and hands over, its note beside the
// state file, while its own run goes on.
let stopping = rt.spawn({
let run = run.clone();
async move { run.stop().await }
let old = old.clone();
async move { old.stop().await }
});
wait_until(|| of(second, "then")["status"] == "cancelled");
wait_until(|| of("refs/heads/main", "then")["status"] == "cancelled");
let notes = runner.path("state.handover");
let noted = || std::fs::read_dir(&notes).map_or(0, |d| d.count());
wait_until(|| noted() == 1);
// The runner started beside it leaves its run and check alone, and
// runs nothing in the job lent to it.
let (new, watches) = start();
rt.spawn(knotless_run::watch(new.clone(), watches));
std::thread::sleep(Duration::from_secs(3));
assert_eq!(of("refs/heads/block", "block")["status"], "running");
let ran = |name: &str| std::fs::read_to_string(go.join(name)).unwrap_or_default();
assert_eq!(ran("thens"), "");
assert!(!stopping.is_finished());
std::fs::write(go.join("then"), "").unwrap();
promptly(&rt, stopping);
assert_eq!(of(first, "then")["status"], "success");
// And the other on the start after.
let (run, watch) = start();
rt.block_on(run.reconcile(&watch)).unwrap();
rt.block_on(run.idle());
let ran_second = of(second, "then");
assert_eq!(ran_second["status"], "success", "{ran_second}");
let mut thens: Vec<String> = ran("thens").lines().map(str::to_string).collect();
thens.sort();
let mut both = vec![push.trim().to_string(), pulled.trim().to_string()];
both.sort();
assert_eq!(thens, both);
assert_eq!(ran("deploys"), format!("{main}\n"));
// Its run done, the old one stops, and the new one runs what it
// dropped in the job given back.
std::fs::write(go.join("block"), "").unwrap();
promptly(&rt, stopping);
assert_eq!(of("refs/heads/block", "block")["status"], "success");
wait_until(|| ran("thens") == format!("{main}\n"));
wait_until(|| of("refs/heads/main", "then")["status"] == "success");
wait_until(|| noted() == 0);
}
/// A stand-in for the runner's PDS, in front of the reference one: it fails
deploy/deploy.sh+19−33
View file
@@ -76,14 +76,12 @@
# container beside the old, the old stopped once the new is healthy; a
# serve whose new build would rebuild the index is stopped and started.
#
# Run by knotless-run itself (KNOTLESS_RUN is set), as the deploy workflow
# runs it, it restarts every service but `run` at once, and `run`, if it
# is to be, after, from the host: stopping it waits for the runs whose
# steps have started, this one among them, which would wait for that stop forever if this
# waited too. A recreate cut short leaves its new container, named
# `<id>_knotless-run-1`, never started, and the next then fails on the
# name, having sent the runner its SIGTERM but never its SIGKILL: each is
# removed first, and `up` tried again a few times.
# `run`, if it is to be restarted, hands over (deploy/handover.sh): the
# one running is sent SIGTERM and, once it says it has handed over, the
# new one starts beside it, while it finishes the runs it has going. Run
# by knotless-run itself (KNOTLESS_RUN is set), as the deploy workflow
# runs it, that goes on from the host in the background, as this run is
# one of those the old runner finishes.
set -euo pipefail
cd "$(dirname "$0")/.."
@@ -255,7 +253,7 @@ fi
[ -z "${KNOTLESS_RUN_ARGS:-}" ] || echo "KNOTLESS_RUN_ARGS=$KNOTLESS_RUN_ARGS"
fi
} | $SSH 'cat > /etc/knotless/.env'
scp -q -o StrictHostKeyChecking=accept-new deploy/docker-compose.yml deploy/rollout.sh "$DEST:/etc/knotless/"
scp -q -o StrictHostKeyChecking=accept-new deploy/docker-compose.yml deploy/rollout.sh deploy/handover.sh "$DEST:/etc/knotless/"
# Whether `run` is restarted: only when what it runs changed, since a
# restart cancels the runs still waiting for a job (docs/design/runner.md,
@@ -283,7 +281,11 @@ STATUS=$($SSH sh -s <<'EOF'
cd /etc/knotless || exit 0
config=$(docker compose config run 2>/dev/null) || exit 0
{ printf '%s\n' "$config"; cat run.env 2>/dev/null; } | sha256sum | cut -d ' ' -f 1 | sed 's/^/settings /'
id=$(docker compose ps -q --status running run 2>/dev/null | head -n 1)
# The one leading: the newest docker still restarts, not one handing
# over (handover.sh).
id=$(docker compose ps -q --status running run 2>/dev/null |
xargs -r docker inspect -f '{{.Created}} {{.HostConfig.RestartPolicy.Name}} {{.Id}}' |
awk '$2 == "always"' | sort | tail -n 1 | awk '{ print $3 }')
[ -n "$id" ] || exit 0
echo "container $id"
echo "build $(docker exec "$id" cat /etc/knotless-run.fingerprint 2>/dev/null)"
@@ -327,7 +329,8 @@ else
fi
# pages and serve roll over with no gap (rollout.sh); anything else is
# restarted as compose does, `run` last, if it is to be.
# restarted as compose does; and `run`, if it is to be, last, the one
# running handing over to the new one (handover.sh).
say "starting"
if [ -z "${KNOTLESS_RUN:-}" ]; then
$SSH sh -s -- "$RUN_MODE" "${SETTINGS:-none}" <<'EOF'
@@ -336,14 +339,8 @@ if [ -z "${KNOTLESS_RUN:-}" ]; then
cd /etc/knotless
docker compose config --quiet
sh rollout.sh pages serve
if [ "$1" = restart ]; then
rm -f run-deployed
docker compose up -d --remove-orphans
id=$(docker compose ps -q run) || id=""
[ -z "$id" ] || [ "$2" = none ] || printf '%s %s\n' "$id" "$2" > run-deployed
else
docker compose up -d --remove-orphans $(docker compose config --services | grep -vx run)
fi
docker compose up -d --remove-orphans $(docker compose config --services | grep -vx run)
[ "$1" != restart ] || sh handover.sh "$2"
docker compose ps
} </dev/null
EOF
@@ -357,20 +354,9 @@ else
docker compose up -d --remove-orphans $(docker compose config --services | grep -vx run)
docker compose ps
[ "$1" = restart ] || exit 0
setsid nohup sh -c '
cd /etc/knotless
rm -f run-deployed
for try in 1 2 3; do
docker compose ps -aq --status created run | xargs -r docker rm
if docker compose up -d run; then
id=$(docker compose ps -q run) || id=""
[ -z "$id" ] || [ "$1" = none ] || printf "%s %s\n" "$id" "$1" > run-deployed
exit
fi
sleep 30
done
exit 1
' restart "$2" > /var/log/knotless-run-restart.log 2>&1 < /dev/null &
# In the background: the runner running this deploy hands over as it
# finishes it, and this run is one of its runs going.
setsid nohup sh handover.sh "$2" > /var/log/knotless-run-restart.log 2>&1 < /dev/null &
} </dev/null
EOF
fi
deploy/docker-compose.yml+5−4
View file
@@ -185,10 +185,11 @@ services:
# that only root can remove.
user: root
# Stopped, it drops each run still waiting its turn, for the next
# start to run, and waits only for the runs whose steps have started:
# most finish in a few minutes, the longest, test, in about 9, and a
# deploy that has started runs on to its end, its commit's checks with
# it, which it waits for. Past --stop-wait (15 minutes) it cancels
# start to run, and finishes only the runs whose steps have started,
# a batch's and those at a going deploy's commit: most finish in a few
# minutes, the longest, test, in about 9. A deploy does not wait for
# that: it starts the next runner beside it (handover.sh), and only a
# stop by hand does. Past --stop-wait (15 minutes) it cancels
# those still going itself, which a stop nothing comes to kill needs.
# 20 minutes is that with room to spare; past it, docker kills the
# runner, and the next ends what was cut off as an error and removes
deploy/handover.shnew+103−0
View file
@@ -0,0 +1,103 @@
#!/bin/sh
# Moves `run` onto what docker-compose.yml beside this script now says,
# the runner running handing over to the new one rather than stopping
# first:
#
# /etc/knotless/handover.sh <settings hash, or none>
#
# deploy.sh copies it to the host and runs it there when the runner is to
# be restarted. The runner leading, the newest still to be restarted by
# docker, is sent SIGTERM: it drops the runs waiting that it will not
# run, writes the state file a last time, and from then on a note of its
# own beside it (state.handover/), while it finishes the runs it has
# going (docs/design/runner.md, "A stop"). Once that note is there, the
# new runner starts beside it, reads the state file and takes up the
# rest, and the old one, no longer restarted, is removed once it exits,
# --stop-wait at most after its stop.
#
# A runner that writes no note within KNOTLESS_HANDOVER_WAIT seconds (60),
# one built before runners handed over, is stopped first instead, and the
# new one started once it has, as every deploy once did.
#
# Once the new runner is up, /etc/knotless/run-deployed says
# "<container id> <settings hash>", for the next deploy to compare.
set -eu
cd "$(dirname "$0")"
SETTINGS="${1:-none}"
WAIT="${KNOTLESS_HANDOVER_WAIT:-60}"
LOG="${KNOTLESS_DEPLOY_LOG:-/var/log/knotless-deploy.log}"
DATA=$(sed -n 's/^KNOTLESS_RUN_DATA=//p' .env 2>/dev/null | tail -n 1)
NOTES="${DATA:-/var/lib/knotless-run}/state.handover"
say() {
line="$(date -u +%Y-%m-%dT%H:%M:%SZ) → run: $*"
echo "$line"
echo "$line" >>"$LOG" 2>/dev/null || true
}
# The runners docker restarts, newest last: the one leading, and any whose
# handover a deploy cut short.
leading() {
docker compose ps -q --status running run |
xargs -r docker inspect -f '{{.Created}} {{.HostConfig.RestartPolicy.Name}} {{.Id}}' |
awk '$2 == "always"' | sort | awk '{ print $3 }'
}
# The notes beside the state file now.
notes() {
ls "$NOTES" 2>/dev/null | grep '\.json$' || true
}
# Notes the new runner as the one deployed with these settings.
record() {
id=$(leading | tail -n 1)
[ -z "$id" ] || [ "$SETTINGS" = none ] || printf '%s %s\n' "$id" "$SETTINGS" >run-deployed
}
rm -f run-deployed
# One a recreate cut short leaves never started, and those that handed
# over and have since exited.
docker compose ps -aq --status created run | xargs -r docker rm >/dev/null
docker compose ps -aq --status exited run | xargs -r docker rm >/dev/null
old=$(leading)
if [ -z "$old" ]; then
say "starting"
docker compose up -d --no-deps run
record
exit
fi
had=$(notes)
for id in $old; do
docker update --restart=no "$id" >/dev/null
docker kill -s TERM "$id" >/dev/null
done
waited=0
while :; do
new=$(notes | grep -vxF "$had" || true)
[ -z "$new" ] || break
if [ "$waited" -ge "$WAIT" ]; then
say "$(echo "$old" | cut -c1-12 | tr '\n' ' ')wrote no handover note in ${WAIT}s: stopping it first"
for id in $old; do docker update --restart=always "$id" >/dev/null; done
docker compose up -d --no-deps --force-recreate run
record
exit
fi
sleep 1
waited=$((waited + 1))
done
running=$(docker compose ps -q --status running run | wc -l)
say "handed over in ${waited}s; starting the new one beside $(echo "$old" | cut -c1-12 | tr '\n' ' ')"
docker compose up -d --no-deps --no-recreate --scale "run=$((running + 1))" run
record
say "started $(leading | tail -n 1 | cut -c1-12)"
# The old ones exit once their runs end; each is removed then, in the
# background, so a deploy by hand does not wait for them.
for id in $old; do
setsid nohup sh -c 'docker wait "$1" >/dev/null 2>&1; docker rm "$1" >/dev/null 2>&1' gone "$id" \
>/dev/null 2>&1 </dev/null &
done
deploy/README.md+34−3
View file
docs/design/merge-queue.md+9−8
View file
docs/design/runner.md+48−9
View file
AGENTS.md+3−3
View file