//! Runs a repository's workflows when its refs move, and reports each run //! as a `dev.knotless.check` record in the runner's own account. //! //! Jetstream is only the doorbell. An event naming a watched repository's //! ref sends the runner to read that repository's refs from its owner's //! PDS, and every ref whose generation has passed the one last run is run //! at the commit the PDS says it points at. Jetstream frames are unsigned, //! so nothing a frame claims is run; a frame missed while disconnected is //! caught by the read every reconnection makes; and a replayed one finds //! nothing new. Every repository is read again each `--sweep` besides, for //! the frames Jetstream drops while connected, and soon after a read that //! failed, as when the PDS does. //! //! A private repository is read in its space, as the runner's account, a //! member of it, and its checks are filed in the runner's repo there, so //! nothing of it is public. Nothing written in a space reaches Jetstream: //! its space is asked instead, every `--poll`, whether anything was //! written there since it was last asked. //! //! The owner asks for a ref or a pull to run again with a //! `dev.knotless.rerun` record in their own account, which rings as a ref //! event does and is read back from their PDS. One that names a workflow //! runs only that one, at the ref, with the variables it gives: one that //! runs on a push there, or that runs only when asked. A run asked for goes //! after every other waiting, and its steps may take as long as their own //! `timeout`s say, past the runner's limit; so as not to hold up a stop, as //! a deploy's, a stop cancels it, and it is asked for again on the next //! start. //! //! Runs go side by side, `--jobs` at once over every repository, so a //! quick workflow's check is never kept waiting behind a slow one's, and //! at most `--pull-jobs` of them (half) for runs other than a push to the //! default branch, so a merge that runs every open pull again starts only //! so many at once, and `--owner-jobs` (all) for the runs of one owner's //! repositories, between which runs waiting take turns. A newer commit's run of a workflow on a ref, or a //! pull, cancels the older one's still going, unless the workflow says its //! runs finish. Runs //! waiting for a job go a push to the default branch before the rest, then //! the oldest first. Each run sees a workflow's caches through overlays of //! its own (see [`caches`]), so runs of one workflow go side by side, and //! only a push to the default branch that passes keeps what it wrote. On //! SIGTERM the runner waits only for the runs whose steps have started, //! for `--stop-wait` at most, then cancels them: each still waiting its //! turn is dropped, its check filed as cancelled, and runs when the //! runner starts again. A check left running by a runner that stopped //! otherwise is ended as an error when it starts again. //! //! With `--pulls` it also runs each open pull into a watched repository, //! as the appview lists them, when its head or the tip of the branch it is //! into moves: at the head merged into that tip, as merging the pull would //! leave it, so a pull that would break the base with another merged since //! fails before it is merged. A pull that conflicts with its base fails //! without running, and a pull's run of a head or base it is no longer at //! is cancelled. One reopened runs as one opened does, unless its head and //! base have checks already. A pull's run is the code of whoever opened it: it gets //! no secrets, and only the docker executor runs it. //! //! A push to the default branch at the same tree as a trusted pull's run //! that passed, as a pull merged as it was checked leaves it, is not run //! again: each workflow that passed there is filed as passed at the push //! too, its log naming the pull's run, and what that run wrote to its //! caches is promoted as the push's own run's would be, where the bases //! it was written over are still current. The tree holds the workflows, //! so theirs were the same. One that asks for secrets, which the pull's run //! never got, or whose runs finish, as a deploy's do, runs as ever. //! //! A repository named with `--pulls-only` has its pushes run by another //! runner, so none here takes a pull's layers: a trusted pull's run into //! its default branch that passes makes its caches' bases instead, as a //! push's would, so its pulls start warm (see [`caches`]). //! //! A workflow with a `schedule` runs at the default branch's tip at the //! minutes its cron expressions name, in UTC, as a run asked for by name //! is, its steps seeing `KNOTLESS_SCHEDULE`, the expression that ran it. //! Each pass over a repository reads its schedules at the default //! branch's tip, when it moved, and keeps them with its state; each minute //! the runner asks for those due, but not one whose last run is still //! going, and rings their repositories. A minute the runner was down for //! is not made up. //! //! A workflow whose trigger names `paths` none of which a push or pull //! changes is not run: its check is filed at once as passed, its steps //! skipped, noting why, so it holds no cache volume and waits for nothing. pub mod caches; pub mod check; pub mod executor; pub mod live; mod prune; mod queue; mod telemetry; pub mod turns; pub use knotless::workflow; use anyhow::{Context, Result, anyhow, bail}; use check::{Check, Reporter, status}; use executor::{Executor, Session}; use futures::{FutureExt, StreamExt, TryStreamExt}; use knotless::checks; use knotless::env::Ctx; use knotless::fetch; use knotless::forge; use knotless::git::Git; use knotless::jetstream; use knotless::keys; use knotless::mirror::Mirror; use knotless::records::{ CheckImage, CheckOutput, Freeing, MinutesRecord, QUEUE_FIRST, RefRecord, RepoRecord, RepoRef, RerunRecord, RerunVar, now, nsid, }; use knotless::remote::{Remote, Snapshot, Stored}; use knotless::sealed; use knotless::session; use knotless::space; use knotless::store; use knotless::url::{RepoUrl, SCHEME}; use serde::{Deserialize, Serialize}; use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; use std::io::Write; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; use tokio::sync::{Notify, Semaphore}; use turns::{Class, Rank, Turns, Whose}; use workflow::{Superseded, Workflow}; const USAGE: &str = "usage: knotless-run [--jetstream ] [--executor docker|shell] [--image ] [--secrets ] [--secrets-key ] [--state ] [--work ] [--step-timeout ] [--run-timeout ] [--cpus ] [--memory ] [--pids ] [--network ] [--runtime ] [--trusted-runtime ] [--trust ] [--resolv ] [--jobs ] [--pull-jobs ] [--owner-jobs ] [--allowance ] [--private-allowance ] [--cache-max ] [--disk-floor ] [--caches ] [--paused ] [--keep-checks ] [--sweep ] [--stop-wait ] [--poll ] [--pulls] [--pulls-only /] [--unmetered ] [--queue [--batch ] [--depth 1|2] [--notice ] [--review-wait ]] [--discover] [/...]"; #[derive(Debug, Clone, PartialEq)] pub struct Options { pub jetstream: String, /// The repositories watched, as `/`. pub repos: Vec, /// Of `repos`, those whose open pulls alone this runner runs, with /// `--pulls`: never their pushes nor their merge queue, which another /// runner runs. pub pulls_only: Vec, /// Whether to watch, besides `repos`, every public repository whose /// record names this runner's account among its runners, as the /// appview lists them and Jetstream brings them. pub discover: bool, pub executor: Executor, /// Secrets granted per repository: `{"/": {"NAME": /// "value"}}`, or `{"value": "...", "refs": ["refs/heads/main"]}` for /// one granted only to pushes to those refs. pub secrets: Option, /// Where it keeps the key owners seal secrets to, which it publishes as /// its `dev.knotless.runnerKey` record. Defaults to `secrets-key.json` /// beside the state file. pub secrets_key: Option, /// The generation each ref was last run at. Defaults to one in the cache /// directory. pub state: Option, /// Where checkouts go while their steps run. Defaults to the system's /// temporary directory. pub work: Option, pub step_timeout: Duration, /// How long a run's steps may take together, if limited: a step going /// when it is up is killed, and those after it are skipped. A run its /// owner asked for by name is not held to it, as it is not to /// `step_timeout`. pub run_timeout: Option, /// Runs at once, of every repository together. pub jobs: usize, /// Of those, the most that runs other than a push to a default branch /// take, so that every pull run again at once when its base moves /// does not take the whole machine: 0 for half (see /// [`Options::pull_share`]). pub pull_jobs: usize, /// Of those, the most one owner's runs take, so that one owner's /// burst leaves the rest room: 0 for all of them (see /// [`Options::owner_share`]). pub owner_jobs: usize, /// The minutes of steps each owner's runs may take in 30 days, if /// limited, but the runner's own account's: past them, an owner's /// runs go one at a time. pub allowance: Option, /// The same for private repositories' runs, counted apart. pub private_allowance: Option, /// The owners, by DID, whose runs count against no allowance, as the /// runner's own account's do not, and may take every job, past /// `owner_jobs`. pub unmetered: Vec, /// The owners, by DID, whose own code, a push or a pull or batch every /// writer of which they vouch for, runs under `--trusted-runtime`. pub trust: Vec, /// The bytes a cache's base may grow to before it is dropped, to /// start afresh at the next run. pub cache_max: u64, /// The bytes kept free on the disk the caches are on: below them, a /// run sheds the caches no run reads before it starts, and waits for /// room if that is not enough, so that what else the disk holds, as /// the site on the same host, never runs out. 0 keeps none. pub disk_floor: u64, /// Where the caches' bases, and each run's layers over them, are kept: /// a path the docker daemon knows by the same name. Defaults to one in /// the cache directory. pub caches: Option, /// How often every watched repository is read again, for what /// Jetstream did not ring for, and how soon one whose read failed is, /// at most a minute. pub sweep: Duration, /// How long a stop waits for the runs going before it cancels them. pub stop_wait: Duration, /// Whether to run the heads of open pulls too. pub pulls: bool, /// Whether to land the merge queues of the repositories watched, as /// their owner or a collaborator. pub queue: bool, /// The most pulls a merge queue's batch takes. pub batch: usize, /// How many of a merge queue's batches run at once: 2, the default, /// runs the next on the one being checked, as if that one passed; 1, /// one at a time. Never more than one runs ahead. pub depth: usize, /// How often the merge queue's records are looked at, besides as /// Jetstream brings them. pub notice: Duration, /// How long a queued pull waits for a review of its head from its /// branch's gates ([`knotless::approvals::gated`]) before it is /// ejected, saying so. pub review_wait: Duration, /// How often a private repository's space is asked whether anything /// was written there: Jetstream carries nothing of a space. pub poll: Duration, /// A JSON object of owners whose runs this runner does not run, by /// DID, or `*` for every owner but the runner's own account, each to /// why: read again on every pass. pub paused: Option, /// How long a run on the default branch keeps its check, unless it is /// the newest of its workflow there: `None` deletes no checks at all /// (see [`prune`]). pub keep_checks: Option, /// How long any check that finished is kept, whatever it is. pub prune_grace: Duration, /// How often a repository's checks are pruned. pub prune_every: Duration, /// How long a pass over a repository runs before it is logged at /// warn, saying where it is, and again each time as long after. pub watchdog: Duration, } impl Options { /// What a runner does unless told otherwise, watching `repos`. pub fn new(repos: Vec) -> Options { Options { jetstream: "wss://jetstream2.us-east.bsky.network".into(), repos, pulls_only: Vec::new(), discover: false, executor: Executor::Shell, secrets: None, secrets_key: None, state: None, work: None, step_timeout: Duration::from_secs(30 * 60), run_timeout: None, jobs: 4, pull_jobs: 0, owner_jobs: 0, allowance: None, private_allowance: None, unmetered: Vec::new(), trust: Vec::new(), cache_max: 16 << 30, disk_floor: 20 << 30, caches: None, sweep: Duration::from_secs(600), stop_wait: Duration::from_secs(15 * 60), pulls: false, queue: false, batch: 8, depth: 2, notice: Duration::from_secs(30), review_wait: Duration::from_secs(30 * 60), poll: Duration::from_secs(10), paused: None, keep_checks: Some(Duration::from_secs(30 * DAY)), prune_grace: Duration::from_secs(DAY), prune_every: Duration::from_secs(60 * 60), watchdog: Duration::from_secs(5 * 60), } } /// The jobs runs other than a push to a default branch may hold: /// `pull_jobs`, at most all of them, or unset, half, at least one. pub fn pull_share(&self) -> usize { match self.pull_jobs { 0 => (self.jobs / 2).max(1), n => n.min(self.jobs), } } /// The jobs one owner's runs may hold: `owner_jobs`, at most all of /// them, or unset, all of them. pub fn owner_share(&self) -> usize { match self.owner_jobs { 0 => self.jobs, n => n.min(self.jobs), } } } /// How long a step may take, at most `step`, and within what is `left` of /// its run's time where that is limited; and whether what is left is what /// holds it. fn within(step: Duration, left: Option) -> (Duration, bool) { match left { Some(left) if left < step => (left, true), _ => (step, false), } } /// What a step killed for its time says of it: after `limit`, or as the /// whole `run`'s time was up. fn killed(limit: Duration, run: Option) -> String { match run { Some(run) => format!("killed as the run's {}s were up", run.as_secs()), None => format!("killed after {}s", limit.as_secs()), } } /// Why `owner`'s runs are paused, as `list`, the `--paused` file's text, /// says: never the runner's own account's, nor any if there is no list. fn paused_why(list: &str, owner: &str, runner: &str) -> Result> { if list.trim().is_empty() || owner == runner { return Ok(None); } let list: HashMap = serde_json::from_str(list).context("--paused takes a JSON object of DIDs to why")?; Ok(list.get(owner).or_else(|| list.get("*")).cloned()) } /// Seconds in a day. const DAY: u64 = 24 * 60 * 60; /// How long `--keep-checks ` keeps the default branch's checks: /// 0 keeps every check. fn kept_for(days: u64) -> Option { match days { 0 => None, days => Some(Duration::from_secs(days * DAY)), } } /// Whether a repository last pruned at `last`, if ever, is due again at /// `now`, `every` after. fn prune_due(last: Option, now: std::time::Instant, every: Duration) -> bool { last.is_none_or(|last| now.duration_since(last) >= every) } /// `value`, read as a number at least 1, or what `flag` takes. fn count(flag: &str, what: &str, value: String) -> Result { let n = value.parse().ok().filter(|n| *n > 0); n.ok_or(format!("{flag} takes a number of {what}, at least 1")) } /// The most pulls a batch takes: a check record's `batch` holds no more. const MAX_BATCH: usize = 64; /// `value`, read as CPUs, as docker's `--cpus` takes them: more than none. fn cpus(value: String) -> Result { let n: f64 = value.parse().unwrap_or(0.0); if !(n > 0.0 && n.is_finite()) { return Err(format!( "--cpus takes a number of CPUs, more than 0, not {value}" )); } Ok(value) } /// `value`, read as seconds, or what `flag` takes. fn secs(flag: &str, value: String) -> Result { let secs = value.parse().map_err(|_| format!("{flag} takes seconds"))?; Ok(Duration::from_secs(secs)) } pub fn parse_args(args: &[String]) -> Result { let mut opts = Options::new(Vec::new()); let mut executor = "docker".to_string(); let mut image = "debian:stable-slim".to_string(); let mut limits = executor::Limits::default(); let mut it = args.iter().skip(1); while let Some(flag) = it.next() { let mut value = || it.next().cloned().ok_or(format!("{flag} takes a value")); match flag.as_str() { "--jetstream" => opts.jetstream = value()?, "--executor" => executor = value()?, "--image" => image = value()?, "--secrets" => opts.secrets = Some(value()?.into()), "--secrets-key" => opts.secrets_key = Some(value()?.into()), "--state" => opts.state = Some(value()?.into()), "--work" => opts.work = Some(value()?.into()), "--caches" => opts.caches = Some(value()?.into()), "--pulls" => opts.pulls = true, "--pulls-only" => { let repo = value()?.trim_start_matches(SCHEME).to_string(); opts.pulls_only.push(repo.clone()); opts.repos.push(repo); } "--unmetered" => opts.unmetered.push(value()?), "--trust" => opts.trust.push(value()?), "--discover" => opts.discover = true, "--queue" => opts.queue = true, "--batch" => { opts.batch = count(flag, "pulls", value()?)?; if opts.batch > MAX_BATCH { return Err(format!("--batch is {MAX_BATCH} pulls at most")); } } "--depth" => { opts.depth = count(flag, "batches", value()?)?; if opts.depth > 2 { return Err("--depth is 1 or 2: one batch at most runs ahead".into()); } } "--notice" => { let secs = count(flag, "seconds", value()?)?; opts.notice = Duration::from_secs(secs as u64); } "--review-wait" => { let secs = count(flag, "seconds", value()?)?; opts.review_wait = Duration::from_secs(secs as u64); } "--poll" => { let secs = count(flag, "seconds", value()?)?; opts.poll = Duration::from_secs(secs as u64); } "--cache-max" => { let gib: u64 = value()? .parse() .map_err(|_| "--cache-max takes GiB".to_string())?; opts.cache_max = gib << 30; } "--disk-floor" => { let gib: u64 = value()? .parse() .map_err(|_| "--disk-floor takes GiB".to_string())?; opts.disk_floor = gib << 30; } "--jobs" => opts.jobs = count(flag, "runs", value()?)?, "--pull-jobs" => opts.pull_jobs = count(flag, "runs", value()?)?, "--owner-jobs" => opts.owner_jobs = count(flag, "runs", value()?)?, "--allowance" => { let minutes = count(flag, "minutes", value()?)?; opts.allowance = Some(minutes as u64); } "--private-allowance" => { let minutes = count(flag, "minutes", value()?)?; opts.private_allowance = Some(minutes as u64); } "--step-timeout" => opts.step_timeout = secs(flag, value()?)?, "--run-timeout" => { let secs = count(flag, "seconds", value()?)?; opts.run_timeout = Some(Duration::from_secs(secs as u64)); } "--cpus" => limits.cpus = Some(cpus(value()?)?), "--memory" => { let gib = count(flag, "GiB", value()?)?; limits.memory = Some((gib as u64) << 30); } "--pids" => limits.pids = Some(count(flag, "processes", value()?)? as u64), "--network" => limits.network = Some(value()?), "--runtime" => limits.runtime = Some(value()?), "--trusted-runtime" => limits.trusted_runtime = Some(value()?), "--resolv" => limits.resolv = Some(value()?), "--paused" => opts.paused = Some(value()?.into()), "--keep-checks" => { let days: u64 = value()? .parse() .map_err(|_| "--keep-checks takes days".to_string())?; opts.keep_checks = kept_for(days); } "--sweep" => { let every = count(flag, "seconds", value()?)?; opts.sweep = Duration::from_secs(every as u64); } "--stop-wait" => opts.stop_wait = secs(flag, value()?)?, other if other.starts_with('-') => return Err(format!("unexpected {other}\n{USAGE}")), repo => opts.repos.push(repo.trim_start_matches(SCHEME).to_string()), } } opts.executor = Executor::parse(&executor, &image) .ok_or(format!("--executor is docker or shell, not {executor}"))?; match &mut opts.executor { Executor::Docker { limits: held, .. } => *held = limits, Executor::Shell if limits != executor::Limits::default() => { return Err( "--cpus, --memory, --pids, --network, --runtime, --trusted-runtime and --resolv hold a run's container: they take --executor docker" .into(), ); } Executor::Shell => {} } if opts.repos.is_empty() && !opts.discover { return Err(format!( "name a repository to watch, or --discover\n{USAGE}" )); } if !opts.pulls_only.is_empty() && !opts.pulls { return Err("--pulls-only runs a repository's pulls: it takes --pulls".into()); } if (opts.pulls || opts.queue) && opts.executor == Executor::Shell { return Err( "--pulls and --queue run code anyone can open a pull with: they take --executor docker" .into(), ); } Ok(opts) } /// The generation of each ref last run, the head and base each open pull /// was last run or passed over at, the pulls the owner has asked to run and /// that have not run since, and the newest request to run again acted on, /// by `/`; and the runs dropped as a runner stopped, to run /// when it starts again. /// 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)] struct State { repos: BTreeMap>, #[serde(default)] pulls: BTreeMap>, #[serde(default)] reruns: BTreeMap, #[serde(default)] asked: BTreeMap>, /// Of the pulls asked for, those asked for only some workflows, by /// name: the rest of them run in full. #[serde(default)] asked_only: BTreeMap>>, /// The trusted pulls' runs that passed, oldest first, by repository. #[serde(default)] passed: BTreeMap>, #[serde(default)] dropped: BTreeMap, /// The workflows asked for by name and not run yet, by repository and /// ref. #[serde(default)] wanted: BTreeMap>, /// The milliseconds each owner's runs' steps took, by the day since /// the epoch they ended on, for the last 30 days. #[serde(default)] used: BTreeMap, /// The same of each owner's private repositories. #[serde(default)] used_private: BTreeMap, /// The repositories found naming this runner, by label, to watch /// again at a start, before the appview is asked. #[serde(default)] discovered: BTreeSet, /// Where the invites read so far end, to read on from. #[serde(default, skip_serializing_if = "Option::is_none")] invited: Option, /// Each repository's scheduled workflows, as its default branch had /// them when it was last read. #[serde(default)] schedules: BTreeMap, /// The minute, since the epoch, each scheduled workflow was last asked /// for, by repository and workflow: once a minute at most. #[serde(default)] fired: BTreeMap>, } /// A repository's scheduled workflows: the default branch they run at, /// the commit there they were read at, and each one's cron expressions, /// by its name. #[derive(Serialize, Deserialize, Default, Clone, Debug)] struct Scheduled { ref_name: String, #[serde(default)] commit: String, crons: BTreeMap>, } /// What a scheduled run's steps see as the cron expression that ran it. pub const SCHEDULE_VAR: &str = "KNOTLESS_SCHEDULE"; impl State { /// The usage of the public repositories, or the private ones. fn usage(&mut self, private: bool) -> &mut BTreeMap { if private { &mut self.used_private } else { &mut self.used } } } /// An owner's milliseconds of steps, by day since the epoch. type Used = BTreeMap; /// The days usage is counted over. const USAGE_DAYS: u64 = 30; /// `ms` used on `day`, with the days before the last [`USAGE_DAYS`] /// dropped. fn spend(used: &mut Used, day: u64, ms: u64) { *used.entry(day).or_default() += ms; used.retain(|d, _| d + USAGE_DAYS > day); } /// The milliseconds used in the last [`USAGE_DAYS`] up to `day`. fn spent(used: &Used, day: u64) -> u64 { used.range(day.saturating_sub(USAGE_DAYS - 1)..) .map(|(_, ms)| ms) .sum() } /// How often an owner's minutes record is written at most, unless its /// standing changes. const MINUTES_EVERY: Duration = Duration::from_secs(5 * 60); /// Where `used` milliseconds stand against an `allowance` of minutes: /// 0 under 80% of it or with none, 1 from there, 2 past it. fn standing(used: u64, allowance: Option) -> u8 { match allowance { Some(minutes) if used >= minutes * 60_000 => 2, Some(minutes) if used * 5 >= minutes * 60_000 * 4 => 1, _ => 0, } } /// Whether an owner's minutes record, last written `last` (when, and at /// what standing), is written again now that it stands at `now`. fn due(last: Option<(std::time::Instant, u8)>, now: u8) -> bool { match last { Some((at, was)) => was != now || at.elapsed() >= MINUTES_EVERY, None => true, } } /// `owner`'s minutes record, of `used` up to `day` and its `allowance`. fn minutes_record(owner: &str, used: &Used, allowance: Option, day: u64) -> MinutesRecord { let at = |d: u64| std::time::UNIX_EPOCH + Duration::from_secs((d + USAGE_DAYS) * 86_400); let freeing = used.iter().map(|(d, ms)| Freeing { at: humantime::format_rfc3339_seconds(at(*d)).to_string(), minutes: ms / 60_000, }); MinutesRecord { owner: Some(owner.to_string()), minutes: spent(used, day) / 60_000, allowance, freeing: freeing.filter(|f| f.minutes > 0).collect(), updated_at: now(), } } /// The `allowance` of minutes, if `used` milliseconds are past it. fn past(used: u64, allowance: Option) -> Option { match allowance { Some(minutes) if used >= minutes * 60_000 => Some(minutes), _ => None, } } /// The jobs an owner's runs may hold: one when `over` an allowance, or /// else `share`. fn most(over: Option, share: usize) -> usize { match over { Some(_) => 1, None => share, } } /// What a run's log says of its wait for a job, `took`, when its owner /// was `past` an allowance of minutes: nothing otherwise. fn waited(past: Option, took: Duration) -> String { match past { Some(minutes) => format!( "==> its owner's runs are past their allowance of {minutes} minutes of steps in 30 days, so they go one at a time: this one waited {}s for a job\n", took.as_secs() ), None => String::new(), } } /// How often a run waiting for room on the disk looks again. const ROOM_EVERY: Duration = Duration::from_secs(30); /// Waits until the disk the caches are on has `floor` bytes free, looking /// each `every`: below it, the caches no run reads are shed first, once, /// and then the runs going are waited for, as they end and free theirs. /// Says what the run's log says of it: nothing when there was room. async fn room(caches: &Arc, floor: u64, every: Duration) -> String { let (began, mut said, mut shed) = (tokio::time::Instant::now(), String::new(), false); loop { let free = match caches.free() { Ok(free) => free, Err(err) => { telemetry::failed("reading the disk's free space", &err.into()); return said; } }; if free >= floor { if began.elapsed() >= every { let took = began.elapsed().as_secs(); said += &format!("==> this one waited {took}s for room on the disk\n"); } return said; } let gib = |bytes: u64| bytes as f64 / (1u64 << 30) as f64; let short = format!( "{:.1} GiB free on the disk, under the runner's floor of {:.1} GiB", gib(free), gib(floor) ); if shed { telemetry::short_of_disk(&short, false); tokio::time::sleep(every).await; continue; } shed = true; telemetry::short_of_disk(&short, true); let caches = caches.clone(); let shedding = tokio::task::spawn_blocking(move || caches.shed()).await; let (bases, kept) = match shedding.map_err(anyhow::Error::from).and_then(|r| r) { Ok(gone) => gone, Err(err) => { telemetry::failed("shedding caches", &err); (0, 0) } }; said += &format!( "==> {short}: shed {bases} caches' bases no run read and {kept} kept runs' layers, so this one starts afresh\n" ); } } /// Today, as days since the epoch. fn today() -> u64 { let since = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH); since.unwrap_or_default().as_secs() / 86_400 } /// Workflows asked for by name, each with the variables it was given. type Asks = BTreeMap>; /// A trusted pull's run of `workflow` that passed at `tree`, into `base`, /// filed under `rkey`: a push to `base` at that tree passes it too. Its /// layers are kept under `rkey` while `pull` is at `at`, as /// [`Subject::at`] says. #[derive(Serialize, Deserialize, Clone, Debug, PartialEq)] struct Passed { tree: String, workflow: String, base: String, rkey: String, #[serde(default)] pull: String, #[serde(default)] at: String, } impl Passed { /// Where its run's layers are kept. fn rkey(&self) -> String { self.rkey.clone() } } /// What a merge queue's batch's runs that passed are kept under, before /// the branch it lands on, in place of a pull. const BATCH: &str = "queue "; /// The pull runs that passed kept for each repository, the newest: more /// than the workflows of the pulls open at once. const PASSED_MAX: usize = 256; /// The runs of a repository dropped before they started, as the runner /// stopped: by the ref or pull each is of, then what it was at /// ([`Subject::at`]), the workflows. One at a ref or a pull's head that /// has moved on since is not run: the newer commit's runs are. One of a /// pull whose base alone has moved runs on the base as it is then. type Dropped = BTreeMap>>; /// A pull due to run, as a pass finds it: its head `sha` on `branch` in /// the repository `remote`, keyed `key`, with its `packs`, and the tip of /// the branch `base` it is into. struct Due { remote: Remote, packs: knotless::remote::Page, key: String, sha: String, generation: u64, branch: String, base: String, base_sha: String, /// The generation of `base` at `base_sha`. base_generation: u64, uri: String, /// Whether its author and its head's owner may both push here. trusted: bool, /// Whether it goes after the merge queue's runs. later: bool, } /// What a run is of: a ref of the watched repository that moved, or the /// head of a pull into it. #[derive(Clone)] struct Subject { /// The ref, or the pull's head branch in the repository it comes from. ref_name: String, commit: String, generation: u64, /// Whether it is a push to the repository's default branch, whose runs /// go first. main: bool, /// The repository's default branch, in full, whose caches the others' /// start from until they have their own. default: String, pull: Option, /// The tree of what is checked out, for a push to the default branch /// and a trusted pull that merges: what runs that passed are found by. tree: Option, /// The files it changes, which a workflow's `paths` are matched /// against: from the commit a ref was at, or a pull's base's tip. /// `None` when not known, as for a new ref, which counts as all of them. changed: Option>, /// The merge queue's batch it is the tip of. batch: Option, /// Whether it moved, or was asked for again in full, rather than only /// some of its workflows being asked for, by name. pushed: bool, /// The workflows asked for by name. asked: Asks, /// The only workflows to run, if not all that run on it: a pull's /// pre-flights, which the merge queue's lander runs itself. only: Option>, } #[derive(Clone)] struct Pull { uri: String, /// The branch it would merge into, in full. base: String, /// That branch's tip, which the head is merged with. base_commit: String, /// That branch's generation at that tip: how many times it had moved, /// so which of two tips is the newer. None for a pull's pre-flight, /// which the merge queue runs. base_generation: Option, /// The commit the head merged into the base makes, or the paths that /// conflict. merged: Result>, /// Whether its author and its head's owner may both push here: then /// its runs that pass may be reused, and it reads its workflows' /// caches whatever they ask for. trusted: bool, /// Whether it runs after the runs of the merge queue it is not in, /// and the pushes'. later: bool, /// The accounts whose shared steps its lock adds that the repository /// does not vouch for, which make it untrusted. strangers: Vec, /// The other runners the repository names, by DID: a pre-flight one /// of them is running, or passed, at the same head and base is not /// run here too. others: Vec, } impl Subject { /// The commit whose workflows run, and which is checked out: a pull's /// merge with its base, or its head when the two conflict. fn checkout(&self) -> &str { match self.pull.as_ref().map(|p| &p.merged) { Some(Ok(merged)) => merged, _ => &self.commit, } } /// What a run is at: the commit, and for a pull or a batch the /// base's too. fn at(&self) -> String { at(&self.commit, self.base_commit()) } /// The base's tip a pull's head is merged with, or a batch is on. fn base_commit(&self) -> Option<&str> { let pull = self.pull.as_ref().map(|p| p.base_commit.as_str()); pull.or(self.batch.as_ref().map(|b| b.base_commit.as_str())) } /// Whether its runs go first, at the CPU share and before the jobs /// others wait for: a push to the default branch, or the merge /// queue's batch that lands next. One run ahead goes as a pull does. fn first(&self) -> bool { self.main && !self.batch.as_ref().is_some_and(|b| b.ahead) } /// The code of someone who may not push here, which gets no secrets. fn proposed(&self) -> bool { self.pull.is_some() || self.batch.is_some() } /// Whether runs that passed are found by its tree: a push to a /// branch, which may reuse those of the pulls into it, or a trusted /// pull that merges, whose runs that pass are kept for it. fn by_tree(&self) -> bool { match &self.pull { Some(pull) => pull.trusted && pull.merged.is_ok(), None => self.ref_name.starts_with("refs/heads/"), } } /// The branch it is of, in full: a pull's base, or the ref. fn branch(&self) -> &str { self.pull.as_ref().map_or(&self.ref_name, |p| &p.base) } /// What its checks are keyed by: its pull, its branch's queue for a /// batch, or its ref. fn keyed(&self) -> String { let queued = self.batch.as_ref().map(|b| keys::queue_name(&b.base)); let pull = self.pull.as_ref().map(|p| p.uri.clone()); (pull.or(queued)).unwrap_or_else(|| self.ref_name.clone()) } } /// What a run of `commit` is at, and for a pull of its base's `base`. fn at(commit: &str, base: Option<&str>) -> String { format!("{commit} {}", base.unwrap_or_default()) } /// The watched repositories, as `(did, key)`, with an open pull from each /// repository a pull comes from. type Heads = HashMap<(String, String), HashSet<(String, String)>>; /// A repository being watched. #[derive(Clone)] pub struct Watch { /// As the command line named it, which is also how the secrets file /// names it. pub label: String, pub remote: Remote, pub key: String, /// Found naming this runner rather than named on the command line: /// watched only while its record names it. pub discovered: bool, /// Named with `--pulls-only`: its pulls run here, its pushes and its /// queue elsewhere. pub pulls_only: bool, /// Set once it is no longer watched, as left or found again under /// another name: the loops that ring and answer it then end. gone: Arc, /// Where the pass over it going is, for [`watched`] to say. stage: Stage, } /// Where a pass over a repository is: what it is waiting on. type Stage = Arc>; impl Watch { fn id(&self) -> String { format!("{}/{}", self.remote.did(), self.key) } /// Says the pass over it is now at `stage`. fn at(&self, stage: &'static str) { *self.stage.lock().unwrap() = stage; } /// Whether it is still watched. fn watched(&self) -> bool { !self.gone.load(Ordering::SeqCst) } } pub struct Runner { ctx: Ctx, opts: Options, /// The keys it opens the secrets owners seal to it with. keys: sealed::RunnerKeys, pub reporter: Reporter, state_path: PathBuf, state: Mutex, heads: Mutex, /// One permit each run spawned holds, waiting or going, so /// [`Runner::idle`] can wait for them all. runs: Arc, /// The runs going, by repository, ref or pull, and workflow. going: Mutex>, /// The jobs, `--jobs` of them. turns: Arc, /// The caches' bases, and the runs' layers over them. caches: Arc, /// The turn of each workflow whose older runs finish /// ([`Superseded::Finish`]) on a ref or pull, under the same keys: /// taken before the rest, and kept from then on. refs: Arc, /// The runs started so far, for each run's place in line. started: AtomicU64, /// When each owner's minutes record was last written, and at what /// standing (see [`standing`]). published: Mutex>, /// When each watched repository's checks were last pruned, by /// [`Watch::id`]. pruned: Mutex>, /// Each merge queue's batches going, and what a failing one showed, /// by `/\n`. lines: Mutex>, /// Each watched repository's door, rung when one of its batches' /// runs ends, for its queue to go on. A repository no longer watched /// has none, and a loop holding a door that is not its repository's /// ends. doors: Mutex, /// Each repository found naming this runner, as it is watched, by /// `(did, key)`: a renamed one is found again under its new name. named: Mutex>, /// Held while a repository found is looked at, one at a time. considering: tokio::sync::Mutex<()>, /// Each repository whose merge queues this runner lands, by /// `/`, and its queue records as last looked at. noticing: Mutex>, /// Set once the runner is stopping: a run that has not started its /// steps by then is dropped. stopping: tokio::sync::watch::Sender, /// Held to read each repository while it reconciles, and to write by a /// stop, which so waits for the runs a reconcile files to be spawned. reconciling: tokio::sync::RwLock<()>, /// Held to land, as the runner stops, the batches whose runs passed, /// one pass at a time. landing: tokio::sync::Mutex<()>, /// The runs whose steps have a container, or are about to, by their /// checks' keys: any other of this runner's is left over, and removed. sessions: Mutex>, /// Where each run going of a workflow whose runs finish is at, as /// [`Runner::dropping`] names it. finishing: tokio::sync::watch::Sender>, /// The appview runs' output goes to as it is written, and writes in /// private repositories' spaces come from ([`live`]). appview: Option>, /// Set while the appview passes on writes in spaces, so that each /// space is asked itself less often. woken: AtomicBool, /// The door of each private repository watched, by its space. private: Mutex>>, } /// A run going of a workflow whose runs finish, noted in /// [`Runner::finishing`] until it is dropped. struct Finishing<'a> { finishing: &'a tokio::sync::watch::Sender>, of: String, } impl<'a> Finishing<'a> { fn new(runner: &'a Runner, of: String) -> Self { let finishing = &runner.finishing; finishing.send_modify(|all| all.push(of.clone())); Finishing { finishing, of } } } impl Drop for Finishing<'_> { fn drop(&mut self) { self.finishing.send_modify(|all| { let at = all.iter().position(|f| *f == self.of); all.swap_remove(at.unwrap_or_default()); }); } } /// A run going, or waiting to. struct Going { /// The commit it is of, and for a pull the base's too. at: String, /// The pull it is of. pull: Option, /// Whether it was asked for by name, so a stop cancels it. asked: bool, cancel: Cancel, } /// Set, with why, to cancel a run. type Cancel = tokio::sync::watch::Sender>; /// Sets `cancel` to cancel its run `why`, unless it is already cancelled /// as its pull moved on or its batch is dropped, which a run kept still /// heeds. fn cancel(cancel: &Cancel, why: &'static str) { cancel.send_modify(|was| *was = was.filter(|w| heeded(w)).or(Some(why))); } /// Whether `why` cancels even a run kept, of a workflow whose runs finish. fn heeded(why: &str) -> bool { why == MOVED || why == DROPPED || why == CUT_OFF } /// What says a run is cancelled, and why. struct Cancelled { why: tokio::sync::watch::Receiver>, /// Whether it has its turn on its ref or pull, as a run of a workflow /// whose runs finish: from then on a newer commit's run does not cancel /// it, and only its pull moving on does. kept: bool, } impl Cancelled { /// Waits until it is cancelled. async fn wait(&mut self) { let kept = self.kept; let cancels = |why: &Option<&str>| why.is_some_and(|w| !kept || heeded(w)); let _ = self.why.wait_for(cancels).await; } fn why(&self) -> &'static str { (*self.why.borrow()).unwrap_or(NEWER) } } /// A run's turn on its ref or pull, as one of a workflow whose runs finish: /// had at once, or waited for. type OnRef = std::pin::Pin + Send>>; /// Why a run is cancelled: a newer commit's run of its workflow on the same /// ref or pull. const NEWER: &str = "for a newer commit's run"; /// Why a pull's run is cancelled: its pull moved on, to another head or /// base, or closed. const MOVED: &str = "as the pull is no longer at that head and base"; /// Why a batch's run is cancelled: the merge queue dropped the batch, as /// the branch moved, or a pull in it moved or left, or a batch it was on /// failed. const DROPPED: &str = "as the merge queue dropped its batch"; /// Why a run not started yet is cancelled as the runner stops, for the /// next one to run. const WAITED: &str = "as the runner stopped, to run once it starts again,"; /// Why a run asked for by name is cancelled: the runner stopped. const STOPPED: &str = "as the runner stopped: asked for again when it starts"; /// Why any other run is cancelled as the runner stops: it ran on past /// what a stop waits for ([`Options::stop_wait`]). 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 { check: Check, rkey: String, rank: Rank, on_ref: Option, cancelled: Cancelled, } impl Runner { /// Logs in as the runner's account and finds each watched repository. pub async fn start(opts: Options, ctx: Ctx) -> Result<(Arc, Vec)> { if (opts.pulls || opts.discover) && ctx.env.appview.is_none() { bail!( "--pulls and --discover find pulls and repositories through an appview, and KNOTLESS_APPVIEW is set empty" ); } let reporter = Reporter::new(&ctx).await?; Runner::start_as(opts, ctx, reporter).await } /// The watch of `label`, a repository named on the command line. async fn named_watch(ctx: &Ctx, label: &str, pulls_only: bool) -> Result { let url = RepoUrl::parse(&format!("{SCHEME}{label}"))?; // A private one is read in its space, as the runner's account. let remote = Remote::open(ctx, &url).await?; let snap = remote.snapshot(ctx).await?; let key = snap .key() .with_context(|| format!("{label}: no repository by that name this runner can read"))?; Ok(Watch { label: label.to_string(), key: key.to_string(), remote, discovered: false, pulls_only, gone: Arc::default(), stage: Arc::default(), }) } /// [`Runner::start`], filing checks with `reporter`. pub async fn start_as( opts: Options, ctx: Ctx, reporter: Reporter, ) -> Result<(Arc, Vec)> { // One it cannot read is passed over, so as not to keep it from the // rest; with none to watch, and none to find, it stops. let mut watches = Vec::new(); let mut unread = None; for label in &opts.repos { let pulls_only = opts.pulls_only.contains(label); match Runner::named_watch(&ctx, label, pulls_only).await { Ok(watch) => watches.push(watch), Err(err) => { telemetry::failed(label, &err); unread.get_or_insert(err); } } } if let Some(err) = unread.filter(|_| watches.is_empty() && !opts.discover) { return Err(err); } let state_path = opts .state .clone() .unwrap_or_else(|| ctx.env.cache_dir.join("run").join("state.json")); let state: State = store::read(&state_path).unwrap_or_default(); let key_path = opts.secrets_key.clone(); let key_path = key_path.unwrap_or_else(|| state_path.with_file_name("secrets-key.json")); let keys = sealed::RunnerKeys::open_or_make(&key_path)?; reporter.put_key(&keys.public()).await?; // The runs whose layers may be kept: those a push may yet take. let kept: Vec = state.passed.values().flatten().map(Passed::rkey).collect(); let root = opts .caches .clone() .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. 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 runner = Runner { caches: caches::Caches::open(&root, &alive, &kept)?, appview: live::Appview::new(&ctx, &reporter.identity), woken: AtomicBool::new(false), private: Mutex::new(HashMap::new()), ctx, keys, turns: Turns::new(opts.jobs, opts.pull_share()), refs: Turns::new(usize::MAX, usize::MAX), runs: Arc::new(Semaphore::new(u32::MAX as usize)), opts, reporter, state_path, state: Mutex::new(state), heads: Mutex::new(HashMap::new()), going: Mutex::new(HashMap::new()), started: AtomicU64::new(0), published: Mutex::new(HashMap::new()), pruned: Mutex::new(HashMap::new()), lines: Mutex::new(HashMap::new()), doors: Mutex::new(HashMap::new()), named: Mutex::new(HashMap::new()), considering: tokio::sync::Mutex::new(()), noticing: Mutex::new(HashMap::new()), stopping: tokio::sync::watch::Sender::new(false), reconciling: tokio::sync::RwLock::new(()), landing: tokio::sync::Mutex::new(()), finishing: tokio::sync::watch::Sender::new(Vec::new()), sessions: Mutex::new(HashSet::new()), }; Ok((Arc::new(runner), watches)) } /// Runs every ref of `watch` that has moved since it last ran, and with /// `--pulls` every open pull into it whose head has. pub async fn reconcile(self: &Arc, watch: &Watch) -> Result<()> { watch.at("waiting for a stop"); let _reading = self.reconciling.read().await; if *self.stopping.borrow() { return Ok(()); } watch.at("reading the repository"); let snap = watch.remote.snapshot(&self.ctx).await?; let me = &self.reporter.identity.did; if snap.key() != Some(watch.key.as_str()) && self.deleted(watch).await { watch.at("clearing what it left"); self.clear(watch).await?; return self.forget(watch); } if watch.discovered && !names(snap.repo.as_ref(), &watch.key, me) { return self.leave(watch); } if snap.key() != Some(watch.key.as_str()) { bail!("{} no longer names the repository it did", watch.label); } let repo = snap .reference(watch.remote.did()) .context("a snapshot with a key has a repository")?; // A deleted branch's caches go with it. let live: Vec = snap.live_refs().into_iter().map(|(name, _)| name).collect(); self.caches.drop_branches(&watch.id(), &live)?; watch.at("reading reruns"); self.reruns(watch).await?; let moved = self.moved(watch, &snap)?; if !moved.is_empty() && !watch.pulls_only { watch.at("running refs"); self.refs(watch, &repo, &snap, moved).await?; } if !watch.pulls_only { watch.at("reading schedules"); self.learn(watch, &snap).await?; } if self.opts.pulls { watch.at("running pulls"); self.pulls(watch, &repo, &snap).await?; } if self.opts.queue && !watch.pulls_only { watch.at("reading the queue"); self.queues(watch, &repo, &snap).await?; } watch.at("pruning checks"); self.prune(watch, &snap).await } /// Whether the repository `watch` is of is gone from its owner's /// account: no repository record there has its key, as a renamed /// one's still does. Not when the account cannot be asked. async fn deleted(&self, watch: &Watch) -> bool { let repos = watch.remote.list_repos().await; repos.is_ok_and(|repos| repos.iter().all(|r| r.value.key != watch.key)) } /// Clears what this runner keeps of `watch`'s repository, deleted: the /// checks it filed of it, with their logs and outputs, the secrets the /// secrets file grants it by name, which a repository made under that /// name later is not to have, and its caches. async fn clear(&self, watch: &Watch) -> Result<()> { let prefix = keys::check_prefix(watch.remote.did(), &watch.key); let checks = self.reporter.checks(&watch.remote, &prefix).await?; let rkeys: Vec = checks.into_iter().map(|(rkey, _)| rkey).collect(); self.reporter.delete(&watch.remote, &rkeys).await?; let secrets = self.opts.secrets.as_deref(); secrets .map(|path| ungrant(path, &watch.label)) .transpose()?; self.caches.drop_repo(&watch.id())?; telemetry::pruned(&watch.label, rkeys.len()); Ok(()) } /// Deletes the checks of `watch` of no more use (see [`prune`]), at /// most once each `prune_every`, unless no checks are deleted. async fn prune(&self, watch: &Watch, snap: &Snapshot) -> Result<()> { let Some(keep) = self.opts.keep_checks else { return Ok(()); }; // A public repository's open pulls come from the appview: with none, // which pulls are done is not known, and nothing is pruned. if watch.remote.space.is_none() && self.ctx.env.appview.is_none() { return Ok(()); } let now = std::time::Instant::now(); { let mut pruned = self.pruned.lock().unwrap(); let last = pruned.get(&watch.id()).copied(); if !prune_due(last, now, self.opts.prune_every) { return Ok(()); } pruned.insert(watch.id(), now); } let target = forge::Target::open(&self.ctx, &watch.remote.url).await?; let entries = target.entries(&self.ctx).await?; let items = forge::assemble(&target.record, watch.remote.did(), &entries); let open = items .into_iter() .filter(|i| i.kind == "pull" && i.state == "open"); let open: HashSet = open.map(|i| i.uri).collect(); let refs: HashSet = snap.live_refs().into_iter().map(|(name, _)| name).collect(); let main = main_ref(snap); let prefix = keys::check_prefix(watch.remote.did(), &watch.key); let checks = self.reporter.checks(&watch.remote, &prefix).await?; let now = prune::Now { open: &open, refs: &refs, main: &main, at: std::time::SystemTime::now(), grace: self.opts.prune_grace, keep, }; let gone = prune::pruned(&checks, &now); self.reporter.delete(&watch.remote, &gone).await?; telemetry::pruned(&watch.label, gone.len()); Ok(()) } async fn refs( self: &Arc, watch: &Watch, repo: &RepoRef, snap: &Snapshot, moved: Vec<(RefRecord, bool)>, ) -> Result<()> { let mirror = Mirror::open(&self.ctx, watch.remote.did(), &watch.key).await?; // Where each ref was, as the last pass left the mirror, for what // its move changes. let was: HashMap = mirror.git.refs().await?.into_iter().collect(); mirror.sync(&watch.remote, snap).await?; let main = main_ref(snap); for (moved, pushed) in moved { let commit = moved.target.clone().unwrap_or_default(); // A ref the mirror had nowhere, or already where it moved to, // as when another pass synced it, changes what is not known. let from = was.get(&moved.name).filter(|from| **from != commit); let from = from.filter(|_| pushed); let changed = match from { Some(from) => changes(&mirror.git, from, &commit).await, None => None, }; let subject = Subject { ref_name: moved.name.clone(), commit, generation: moved.generation, main: moved.name == main, default: main.clone(), pull: None, tree: None, changed, batch: None, pushed, asked: self.wanted(watch, &moved.name), only: None, }; self.run_at(watch, repo, &mirror.git, subject).await?; let mut state = self.state.lock().unwrap(); state .repos .entry(watch.id()) .or_default() .insert(moved.name.clone(), moved.generation); // Stopping, its runs were dropped or cancelled as they were // filed, and asked for again: what is asked stays so. 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)?; } Ok(()) } /// Runs each open pull into `watch` whose head, or the tip of the /// branch it is into, is not where it was last run, as the appview /// lists them and their heads' owners' PDSes say. Each runs in the /// watched repository's mirror, at its head merged into that tip, as /// merging it would leave the branch: a fork's head is fetched there /// from the fork's packs. async fn pulls(self: &Arc, watch: &Watch, repo: &RepoRef, snap: &Snapshot) -> Result<()> { let target = forge::Target::read(&self.ctx, watch.remote.clone(), snap.clone()).await?; let entries = target.entries(&self.ctx).await?; let items = forge::assemble(&target.record, watch.remote.did(), &entries); // A pull queued runs with its batch, when this runner lands the // queue. A repository whose pulls alone run here has its queue // landed by another runner, so its pulls are run as if this one // landed it. let lands = watch.pulls_only || self.lands(&target.record, watch); let line = knotless::queue::queued(&items).into_iter(); let queued: HashSet = line.filter(|_| lands).map(|i| i.uri.clone()).collect(); let open = items .into_iter() .filter(|i| i.kind == "pull" && i.state == "open") .filter_map(|i| Some((i.uri, i.head?, i.base?))); let first = !self.state.lock().unwrap().pulls.contains_key(&watch.id()); let main = main_ref(snap); let owner = watch.remote.did(); let others = self.others(&target.record); let mut now = BTreeMap::new(); let mut due = Vec::new(); // Each head's repository read once a pass, by its owner and key: // the watched one's, as read already, and each fork's on its first // pull. let ours = (watch.remote.clone(), snap.clone()); let mut heads = HashMap::from([((owner.to_string(), watch.key.clone()), ours)]); for (uri, head, base) in open { let (did, _, name) = forge::parse_uri(&head.repo.uri)?; let watched = (watch.remote.did().to_string(), watch.key.clone()); self.heads .lock() .unwrap() .entry((did.to_string(), head.repo.key.clone())) .or_default() .insert(watched); let read = (did.to_string(), head.repo.key.clone()); if !heads.contains_key(&read) { let url = RepoUrl { authority: did.to_string(), name: name.to_string(), }; let remote = Remote::open(&self.ctx, &url).await?; let theirs = remote.snapshot(&self.ctx).await?; heads.insert(read.clone(), (remote, theirs)); } let (remote, theirs) = &heads[&read]; let branch = format!("refs/heads/{}", head.branch); let base = format!("refs/heads/{base}"); // A branch deleted, or a name that holds another repository now, // has no head to run; a base deleted, nothing to merge it into. let Some((sha, generation)) = theirs .find_ref(&branch) .filter(|_| theirs.key() == Some(head.repo.key.as_str())) .and_then(|r| Some((r.value.target.clone()?, r.value.generation))) else { continue; }; let tip = snap.find_ref(&base); let Some((base_sha, base_generation)) = tip.and_then(|r| Some((r.value.target.clone()?, r.value.generation))) else { continue; }; let at = format!("{sha} {base_sha}"); let last = self .state .lock() .unwrap() .pulls .get(&watch.id()) .and_then(|p| p.get(&uri).cloned()); // Whoever may push here runs their pulls by opening them, and // so does an agent the repository names, as code no one vouches // for; anyone else's runs only once the owner asks, and then // where it is. let author = forge::parse_uri(&uri)?.0; let (record, writers) = (&target.record, [author, did]); let trusted = writers.iter().all(|who| vouched(record, owner, who)); let itself = writers.iter().all(|who| runs_itself(record, owner, who)); // Forgotten only once its run is filed, below, so a read that // fails meanwhile leaves it asked for. let asked = (self.state.lock().unwrap().asked.get(&watch.id())) .is_some_and(|asked| asked.contains(&uri)); let moved = !first && last.as_deref() != Some(at.as_str()); let dropped = self.head_dropped(watch, &uri, &sha); // Landing the queue, a pull not in it runs again as its head // moves, not as the branch does: the queue checks it against // the branch once it is queued, and its runs go after the // queue's and the pushes'. let ran_head = last.as_deref().and_then(|l| l.split(' ').next()); let head_moved = ran_head != Some(sha.as_str()); let moves = moved && (!lands || head_moved); let mut runs = moves && itself && !queued.contains(&uri); // One not seen open since, as one closed and reopened, ran // already if its checks of this head and base are there. if runs && last.is_none() && !asked && !dropped { runs = !self.checked(watch, &uri, &sha, &base_sha).await?; } // Asked for only some of its workflows, and due for no other // reason, those run as a run dropped would, where it is. if asked && !runs && !dropped { self.ask_only(watch, &uri, &at); } let runs = runs || asked || dropped; let kept = last.filter(|_| lands && !head_moved && !runs); if runs { due.push(Due { remote: remote.clone(), packs: theirs.packs.clone(), key: head.repo.key.clone(), sha, generation, branch, base, base_sha, base_generation, uri: uri.clone(), trusted, later: lands, }); } now.insert(uri, kept.unwrap_or(at)); } self.out_of_date(watch, &now); // Those due run in the mirror, synced once for them all. if !due.is_empty() { let mirror = Mirror::open(&self.ctx, owner, &watch.key).await?; mirror.sync(&watch.remote, snap).await?; let git = &mirror.git; for d in due { let wants = std::slice::from_ref(&d.sha); fetch::fetch(&d.remote, &d.key, git, wants, Some(d.packs)).await?; let merged = forge::merge_result(git, &d.base_sha, &d.sha).await?; // One that conflicts fails whatever it changes. let changed = match &merged { Ok(merged) => changes(git, &d.base_sha, merged).await, Err(_) => None, }; let strangers = match &merged { Ok(merged) => { self.strangers(git, &target.record, owner, &d.base_sha, merged) .await } Err(_) => Vec::new(), }; let subject = Subject { ref_name: d.branch, commit: d.sha, generation: d.generation, main: false, default: main.clone(), pull: Some(Pull { uri: d.uri, base: d.base, base_commit: d.base_sha, base_generation: Some(d.base_generation), merged, trusted: d.trusted && strangers.is_empty(), later: d.later, strangers, others: others.clone(), }), tree: None, changed, batch: None, pushed: true, asked: Asks::new(), only: None, }; self.run_at(watch, repo, git, subject).await?; } } // Pulls closed since, or whose branch is gone, drop out, and their // runs dropped with them. let mut state = self.state.lock().unwrap(); let before = state.pulls.insert(watch.id(), now.clone()); let before = before.unwrap_or_default(); // Each asked for has run, or is closed or gone. state.asked.remove(&watch.id()); state.asked_only.remove(&watch.id()); let dropped = state.dropped.entry(watch.id()).or_default(); for uri in before.keys() { if !now.contains_key(uri) { dropped.remove(uri); } } // The layers a pull's run kept go once the pull is at another head // or base, or a pass after it closed: by then the push a merge // made, read before the pulls are, has taken them. // A batch's go as the next batch on its branch passes instead. let passed = state.passed.get(&watch.id()).into_iter().flatten(); let gone = |p: &&Passed| match now.get(&p.pull) { _ if p.pull.starts_with(BATCH) => false, Some(at) => *at != p.at, None => !before.contains_key(&p.pull), }; let stale: Vec = passed.filter(gone).map(Passed::rkey).collect(); store::write(&self.state_path, &*state, false)?; drop(state); self.caches.forget(&stale, self.opts.cache_max) } /// Whether this runner has filed checks of the pull `uri` into `watch` /// at its head `sha` merged with its base at `base`. async fn checked(&self, watch: &Watch, uri: &str, sha: &str, base: &str) -> Result { let of = keys::checks_of(watch.remote.did(), &watch.key, uri); let at = format!("{of}.{sha}.{base}"); self.reporter.any(&watch.remote, &at).await } /// Cancels the runs of pulls into `watch` that are out of date: of a /// head or base other than the one `now` says the pull is at, or of a /// pull no longer open, even of a workflow whose runs finish. fn out_of_date(&self, watch: &Watch, now: &BTreeMap) { let repo = format!("{}\n", watch.id()); for (id, going) in self.going.lock().unwrap().iter() { let moved = (going.pull.as_ref()).is_some_and(|uri| now.get(uri) != Some(&going.at)); if moved && id.starts_with(&repo) { cancel(&going.cancel, MOVED); } } } /// Forgets where each ref or pull the owner has asked to run again since /// the last request acted on was last run, so the passes after run it. async fn reruns(&self, watch: &Watch) -> Result<()> { let found = watch .remote .scan_all::(nsid::RERUN, &watch.key) .await?; // Keyed by a TID, so the newest acted on sorts before every request // after it. let asked: Vec<_> = found .into_iter() .filter(|r| { keys::suffix(&watch.key, &r.rkey) .and_then(keys::tid_micros) .is_some() }) .collect(); let newest = asked.iter().map(|r| r.rkey.clone()).max(); let id = watch.id(); let mut state = self.state.lock().unwrap(); if let Some(since) = state.reruns.get(&id).cloned() { let wanted = asked .into_iter() .filter(|r| r.rkey > since) .filter_map(|r| serde_json::from_value::(r.value).ok()); for rerun in wanted { let State { repos, pulls, asked, asked_only, dropped, wanted, .. } = &mut *state; // Asked for again, it all runs, not only what was dropped. let dropped = dropped.entry(id.clone()).or_default(); let env = rerun.env.into_iter().map(|v| (v.name, v.value)); let asked = asked.entry(id.clone()).or_default(); let only = asked_only.entry(id.clone()).or_default(); match (rerun.ref_name, &rerun.workflow) { (Some(name), Some(workflow)) => { let refs = wanted.entry(id.clone()).or_default(); let asks = refs.entry(name).or_default(); asks.insert(workflow.clone(), env.collect()); } (Some(name), None) => { repos.get_mut(&id).map(|refs| refs.remove(&name)); dropped.remove(&name); } (None, _) => {} } match (rerun.pull, rerun.workflow) { // One workflow of a pull runs where the pull is, unless // all of it is asked for already. (Some(uri), Some(workflow)) => { if asked.insert(uri.clone()) || only.contains_key(&uri) { only.entry(uri).or_default().insert(workflow); } } (Some(uri), None) => { pulls.get_mut(&id).map(|heads| heads.remove(&uri)); dropped.remove(&uri); only.remove(&uri); asked.insert(uri); } (None, _) => {} } } } let since = state.reruns.get(&id).cloned().unwrap_or_default(); state .reruns .insert(id, newest.unwrap_or_default().max(since)); store::write(&self.state_path, &*state, false) } /// Where the writes in the space of `watch`, a private repository, /// are now, if any were made after `since`. async fn space_moved(&self, watch: &Watch, since: Option<&str>) -> Result> { let uri = &watch .remote .space .as_ref() .expect("a private repository") .uri; let joined = space::joined(&self.ctx, uri).await?; let (host, owner) = (&watch.remote.xrpc, watch.remote.did()); joined.moved_since(host, owner, since).await } /// Asks the space `watch`'s private repository is in to tell the /// appview of each write there, as a member may. async fn register(&self, watch: &Watch) -> Result<()> { let appview = self.appview.as_ref().expect("an appview to tell"); let space = watch.remote.space.as_ref().expect("a private repository"); let joined = space::joined(&self.ctx, &space.uri).await?; let service = appview.service().await?; let (host, owner) = (&watch.remote.xrpc, watch.remote.did()); joined.register_notify(host, owner, &service).await?; Ok(()) } /// Whether a run of the pull `uri` at its head `sha` was dropped, for /// `watch`, as the runner stopped, on whatever base: one whose base has /// moved since runs on the base as it is now, as a runner landing the /// queue runs a pull again only as its head moves. fn head_dropped(&self, watch: &Watch, uri: &str, sha: &str) -> bool { let state = self.state.lock().unwrap(); let dropped = state.dropped.get(&watch.id()).and_then(|d| d.get(uri)); let head = |at: &String| at.split(' ').next() == Some(sha); dropped.is_some_and(|d| d.keys().any(head)) } /// Has the pull `uri`'s run at `at` run only the workflows it was /// asked for by name, if it was, as [`Runner::resumed`] does a run /// dropped. fn ask_only(&self, watch: &Watch, uri: &str, at: &str) { let mut state = self.state.lock().unwrap(); let only = state.asked_only.get(&watch.id()); let Some(only) = only.and_then(|o| o.get(uri)).cloned() else { return; }; let dropped = state.dropped.entry(watch.id()).or_default(); let ats = dropped.entry(uri.to_string()).or_default(); ats.entry(at.to_string()).or_default().extend(only); } /// Whether a run of the ref or pull `of` at `at` was dropped, for /// `watch`, as the runner stopped. fn was_dropped(&self, watch: &Watch, of: &str, at: &str) -> bool { let state = self.state.lock().unwrap(); let dropped = state.dropped.get(&watch.id()).and_then(|d| d.get(of)); dropped.is_some_and(|d| d.contains_key(at)) } /// Reads the schedules of the workflows at the tip of the default /// branch of `watch`, when it is not where they were last read: when /// the repository is first watched, after a push there, and at a start /// once its tip moved meanwhile, whichever runner ran the push. async fn learn(&self, watch: &Watch, snap: &Snapshot) -> Result<()> { let main = main_ref(snap); let tip = snap.refs.iter().find(|r| r.value.name == main); let tip = tip.and_then(|r| r.value.target.clone()); let known = self.known_schedule(watch); // An empty repository has nothing to read. let Some(tip) = tip.filter(|tip| known.as_ref() != Some(tip)) else { return Ok(()); }; let mirror = Mirror::open(&self.ctx, watch.remote.did(), &watch.key).await?; mirror.sync(&watch.remote, snap).await?; let loaded = workflow::load(&mirror.git, &tip).await?; let scheduled = Scheduled { ref_name: main, commit: tip, crons: workflow::scheduled(&loaded), }; let mut state = self.state.lock().unwrap(); // What a schedule no longer there last fired at goes with it. 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) } /// The commit `watch`'s schedules were last read at. fn known_schedule(&self, watch: &Watch) -> Option { let state = self.state.lock().unwrap(); state.schedules.get(&watch.id()).map(|s| s.commit.clone()) } /// Asks for each scheduled workflow due at `at`, in minutes since the /// epoch, at its repository's default branch, once, and rings each /// repository that has one. pub fn on_schedule(&self, at: u64) -> Result<()> { let minute = knotless::cron::Minute::at(at * 60); let due = |cron: &&String| { let cron = knotless::cron::Cron::parse(cron); cron.is_ok_and(|c| c.runs_at(minute)) }; // A run at the default branch still going finishes, rather than // being cancelled for this one: this time is passed over. let going: HashSet = self.going.lock().unwrap().keys().cloned().collect(); let mut rung = Vec::new(); let mut state = self.state.lock().unwrap(); let State { schedules, fired, wanted, .. } = &mut *state; for (id, scheduled) in schedules.iter() { let last = fired.entry(id.clone()).or_default(); for (name, crons) in &scheduled.crons { let cron = crons.iter().find(due); let run = format!("{id}\n{}\n{name}", scheduled.ref_name); let fresh = |_: &&String| last.get(name) != Some(&at) && !going.contains(&run); let Some(cron) = cron.filter(fresh) else { continue; }; last.insert(name.clone(), at); let refs = wanted.entry(id.clone()).or_default(); let asks = refs.entry(scheduled.ref_name.clone()).or_default(); asks.insert(name.clone(), vec![(SCHEDULE_VAR.into(), cron.clone())]); rung.push(id.clone()); } } if !rung.is_empty() { store::write(&self.state_path, &*state, false)?; } drop(state); let doors = self.doors.lock().unwrap(); for (did, key) in rung.iter().filter_map(|id| id.split_once('/')) { let door = doors.get(&(did.to_string(), key.to_string())); door.inspect(|d| d.notify_one()); } Ok(()) } /// The workflows asked for by name at the ref `name` of `watch`. fn wanted(&self, watch: &Watch, name: &str) -> Asks { let state = self.state.lock().unwrap(); let refs = state.wanted.get(&watch.id()); refs.and_then(|r| r.get(name)).cloned().unwrap_or_default() } /// Asks again, on the next start, for the run of `wf` at `subject` asked /// for by name, which the runner stopped before it ended. fn ask_again(&self, watch: &Watch, subject: &Subject, wf: &Workflow) -> Result<()> { let Some(env) = subject.asked.get(&wf.name) else { return Ok(()); }; let mut state = self.state.lock().unwrap(); 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) } /// The refs in `snap` that point past the generation last run, or at /// a commit whose runs were dropped as the runner stopped, each with /// whether it did; and those with workflows asked for by name, which /// are forgotten for a ref that is no more. fn moved(&self, watch: &Watch, snap: &Snapshot) -> Result> { let mut state = self.state.lock().unwrap(); let seen = match state.repos.get(&watch.id()) { Some(seen) => seen.clone(), None => { // Seen first, its refs are where it starts from, but one // found naming this runner runs its default branch's tip // once, so the push that made it shows a check. let main = main_ref(snap); let first: BTreeMap = (snap.refs.iter()) .filter(|r| !(watch.discovered && r.value.name == main)) .map(|r| (r.value.name.clone(), r.value.generation)) .collect(); state.repos.insert(watch.id(), first.clone()); store::write(&self.state_path, &*state, false)?; if !watch.discovered { return Ok(Vec::new()); } first } }; let live: Vec<&RefRecord> = (snap.refs.iter().map(|r| &r.value)) .filter(|r| r.target.is_some()) .collect(); 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)?; drop(state); let moved = |r: &RefRecord| { let past = seen.get(&r.name).is_none_or(|g| *g < r.generation); let target = r.target.as_deref().unwrap_or_default(); past || self.was_dropped(watch, &r.name, &at(target, None)) }; let due = live.into_iter().filter_map(|r| { let pushed = moved(r); (pushed || wanted.contains_key(&r.name)).then(|| (r.clone(), pushed)) }); Ok(due.collect()) } /// The workflows to run at `subject`, if not all: those dropped as the /// runner stopped, when it is where they were dropped. What was dropped /// of it is forgotten either way. fn resumed(&self, watch: &Watch, subject: &Subject) -> Option> { let mut state = self.state.lock().unwrap(); let mut ats = state .dropped .get_mut(&watch.id())? .remove(&subject.keyed())?; ats.remove(&subject.at()) } /// Starts the workflows at `subject`'s commit that run on it, each once /// a permit is free, and files a check for each, and for each workflow /// file that is not one. Every check is filed, as queued or as how it /// ended, before any run starts, so a reader never sees some of /// a commit's checks passed while another is not there yet. A run goes /// on after this returns. async fn run_at( self: &Arc, watch: &Watch, repo: &RepoRef, git: &Git, mut subject: Subject, ) -> Result<()> { if subject.by_tree() { let rev = format!("{}^{{tree}}", subject.checkout()); let tree = git.rev_parse(&rev).await?; subject.tree = Some(tree); } let passed = self.passed_at(watch, &subject); // A pull's checks are keyed by the pull, as a ref's by the ref, and // a batch's by its branch's queue. let keyed = subject.keyed(); let keyed = keyed.as_str(); let base = subject.base_commit(); let batch = subject.batch.as_ref(); let resumed = self.resumed(watch, &subject); let only = subject.only.clone().or(resumed); let mut filed = Vec::new(); let blank = |workflow: &str| Check { repo: repo.clone(), ref_name: subject.ref_name.clone(), commit: subject.commit.clone(), base_commit: base.map(str::to_string), generation: Some(subject.generation), pull: subject.pull.as_ref().map(|p| p.uri.clone()), queue: None, batch: batch.map(|b| b.pulls.clone()).unwrap_or_default(), workflow: workflow.to_string(), status: status::QUEUED.into(), steps: Vec::new(), error: None, skipped: None, log: None, outputs: Vec::new(), started_at: now(), finished_at: None, }; let loaded = workflow::load(git, subject.checkout()).await?; let paused = self.paused(watch)?; // One asked for that is not there says so, under its name. for name in subject.asked.keys() { if !workflow::valid_name(name) || loaded.iter().any(|l| l.name == *name) { continue; } let mut check = blank(name); check.status = status::ERROR.into(); check.error = Some(format!( "asked for, but there is no workflow named {name} at {}", short(&subject.commit) )); check.finished_at = Some(now()); let rkey = run_rkey(watch, &subject, name); self.reporter.put(&watch.remote, &rkey, &check).await?; } for loaded in loaded { let asked = subject.asked.contains_key(&loaded.name); if only .as_ref() .is_some_and(|only| !only.contains(&loaded.name)) || !(subject.pushed || asked) { continue; } let mut check = blank(&loaded.name); let rkey = run_rkey(watch, &subject, &loaded.name); // A paused owner's runs end as a workflow that does not load. let workflow = match &paused { Some(why) => Err(anyhow!("paused by this runner's operator: {why}")), None => loaded.workflow, }; let wf = match workflow { Ok(wf) if asked && !wf.runs_when_asked(&subject.ref_name) && !(subject.main && !wf.on.schedule.is_empty()) => { check.status = status::ERROR.into(); check.error = Some(format!( "asked for, but it runs neither on a push to {} nor when asked there", subject.ref_name )); check.finished_at = Some(now()); self.reporter.put(&watch.remote, &rkey, &check).await?; continue; } Ok(wf) if (asked || runs(&wf, &subject)) && self.named(&wf).await => wf, Ok(_) => continue, Err(err) => { check.status = status::ERROR.into(); check.error = Some(format!("{err:#}")); check.finished_at = Some(now()); self.reporter.put(&watch.remote, &rkey, &check).await?; batch.inspect(|b| b.ended(&loaded.name, status::ERROR)); continue; } }; // A pull's run of a pre-flight says so, for the queue's readers. if subject .pull .as_ref() .is_some_and(|p| wf.first_in_queue(&p.base)) { check.queue = Some(QUEUE_FIRST.to_string()); } if !asked && self.reused(watch, &passed, &wf, &mut check).await? { self.reporter.put(&watch.remote, &rkey, &check).await?; continue; } // A pull that does not merge has nothing to run: merging it // would fail too. if let Some(Pull { base, base_commit, merged: Err(files), .. }) = &subject.pull { check.status = status::FAILURE.into(); check.error = Some(conflict(&subject.commit, base, base_commit, files)); check.finished_at = Some(now()); self.reporter.put(&watch.remote, &rkey, &check).await?; continue; } // One the change leaves out has passed, saying so, for whatever // waits for it. A batch is pulls landing, so it is left out by // the pull trigger's paths, as they are. let pull = subject.pull.is_some() || subject.batch.is_some(); let skipped = wf.skipped(pull, subject.changed.as_deref()); if let Some(why) = skipped.filter(|_| !asked) { check.status = status::SUCCESS.into(); check.steps = wf.steps.iter().map(skipped_step).collect(); check.skipped = Some(why); check.finished_at = Some(now()); self.reporter.put(&watch.remote, &rkey, &check).await?; // A push at the same tree takes it, once its base passed. self.keep_passed(watch, &subject, &wf, &rkey)?; continue; } self.reporter.put(&watch.remote, &rkey, &check).await?; telemetry::run("run queued", watch, &subject, &wf.name, None, None); filed.push((wf, check, rkey)); } for (wf, check, rkey) in filed { // One run of a workflow on a ref or pull at a time: an older // commit's still going is cancelled, or, if the workflow says // its runs finish, only while it waits for its turn on the ref. // A queue's batches go side by side, each its own. let tip = batch.map(|_| format!(" {}", subject.commit)); let id = format!( "{}\n{keyed}{}\n{}", watch.id(), tip.unwrap_or_default(), wf.name ); batch.inspect(|b| b.started(&wf.name)); let (sender, why) = tokio::sync::watch::channel(None); let asked = subject.asked.contains_key(&wf.name); let going = Going { at: subject.at(), pull: subject.pull.as_ref().map(|p| p.uri.clone()), asked, cancel: sender.clone(), }; // One asked for cancels the last asked for at the same commit, // as both would file the one check. let older = self.going.lock().unwrap().insert(id.clone(), going); if let Some(older) = older && (older.at != subject.at() || asked) { cancel(&older.cancel, NEWER); } let later = subject.pull.as_ref().is_some_and(|p| p.later); let class = match (asked, subject.first(), later) { (true, _, _) => Class::Asked, (false, true, _) => Class::Main, (false, false, true) => Class::Later, (false, false, false) if subject.main => Class::Later, (false, false, false) => Class::Other, }; let seq = self.started.fetch_add(1, Ordering::Relaxed); let rank = Rank { class, seq }; let mut cancelled = Cancelled { why, kept: false }; // A run of a workflow whose runs finish takes its turn on the // ref here, if no run has it, before a newer commit's run can be // filed, and is kept from then on; or waits for it. let mut on_ref = None; if wf.superseded == Superseded::Finish { let whose = Whose { owner: watch.remote.did().to_string(), most: usize::MAX, }; let mut turn: OnRef = Box::pin(self.refs.clone().take(rank, whose, vec![id.clone()])); if let Some(had) = (&mut turn).now_or_never() { (turn, cancelled.kept) = (Box::pin(std::future::ready(had)), true); } on_ref = Some(turn); } let run = self.runs.clone().acquire_owned().await?; let (runner, watch, git) = (self.clone(), watch.clone(), git.clone()); let subject = subject.clone(); let job = Job { check, rkey, rank, on_ref, cancelled, }; tokio::spawn(async move { let ran = runner.finish(&watch, &git, &subject, &wf, job).await; // Stopping, a batch's run dropped is not one that failed, // and one that failed is checked again on the next start. // One that passed counts: the last of a batch to, the batch // lands before this run lets the runner stop, rather than // be checked again for a landing of a few seconds. let stopping = *runner.stopping.borrow(); let ended = ran.as_deref().unwrap_or(status::ERROR); let counts = !stopping || ended == status::SUCCESS; if let Some(batch) = subject.batch.as_ref().filter(|_| counts) { batch.ended(&wf.name, ended); if !stopping { runner.ring(&watch); } if stopping && batch.outcome().is_some() { report(&watch, runner.land_passed(&watch).await); } } report(&watch, ran.map(drop)); let mut going = runner.going.lock().unwrap(); if going .get(&id) .is_some_and(|g| g.cancel.same_channel(&sender)) { going.remove(&id); } drop(run); }); } Ok(()) } /// Runs `wf` into the job's check until it ends or is cancelled, and /// files how it ended, which it returns. async fn finish( &self, watch: &Watch, git: &Git, subject: &Subject, wf: &Workflow, job: Job, ) -> Result { let Job { mut check, rkey, rank, on_ref, mut cancelled, } = job; let rkey = rkey.as_str(); let turns = self.take_turns(rank, watch, on_ref, &mut cancelled); // Stopping, one not started yet is dropped, to run on the next start. let of = format!("{}\n{}\n{}", watch.id(), subject.keyed(), subject.at()); let held = tokio::select! { biased; _ = self.dropping(&of) => { 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; } held = turns => held, }; let Some(Held { on_ref, job, said }) = held else { let why = cancelled.why(); let said = format!("run cancelled before it started: {why}"); telemetry::run( &said, watch, subject, &wf.name, Some(status::CANCELLED), None, ); return self.never_started(watch, rkey, check, why).await; }; // 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; // A pre-flight another runner took up meanwhile is left to it. if let Some(theirs) = self.run_elsewhere(watch, subject, &check).await { let said = format!("run left to another runner's: {}", theirs.uri); telemetry::run(&said, watch, subject, &wf.name, None, None); self.reporter .delete(&watch.remote, &[rkey.to_string()]) .await?; return Ok(theirs.record.status); } check.status = status::RUNNING.into(); check.started_at = now(); self.reporter.put(&watch.remote, rkey, &check).await?; let began = std::time::Instant::now(); telemetry::run("run started", watch, subject, &wf.name, None, None); let started = Started { finishing: (wf.superseded == Superseded::Finish).then(|| Finishing::new(self, of)), job, cancelled, said, }; let ran = self.steps(watch, git, subject, wf, &mut check, started); if let Err(err) = ran.await { check.status = status::ERROR.into(); check.error = Some(format!("{err:#}")); } check.finished_at = Some(now()); if let Some(minutes) = self.use_steps(watch, &check)? { self.reporter .put_minutes(watch.remote.did(), &minutes) .await?; } self.reporter.put(&watch.remote, rkey, &check).await?; let (ended, took) = (Some(check.status.as_str()), Some(began.elapsed())); telemetry::run("run ended", watch, subject, &wf.name, ended, took); if check.status == status::SUCCESS && !gated(&check) { self.keep_passed(watch, subject, wf, rkey)?; } if check.status == status::CANCELLED && *self.stopping.borrow() { self.ask_again(watch, subject, wf)?; } Ok(check.status) } /// The accounts the lock at `tip` names shared steps in that the lock /// at `base` does not, and the repository does not vouch for: a pull /// or batch bringing one is code no one has vouched for, as an agent's /// is, until it lands. One whose name does not resolve is among them. pub(crate) async fn strangers( &self, git: &Git, record: &RepoRecord, owner: &str, base: &str, tip: &str, ) -> Vec { let mut strangers = Vec::new(); for who in knotless::uses::added(git, base, tip).await { let resolved = knotless::identity::resolve(&self.ctx, &who).await; if !resolved.is_ok_and(|id| vouched(record, owner, &id.did)) { strangers.push(who); } } strangers } /// The other runners `record` names, by DID. fn others(&self, record: &RepoRecord) -> Vec { let me = &self.reporter.identity.did; let others = record.runners.iter().filter(|r| r.did != *me); others.map(|r| r.did.clone()).collect() } /// Another runner's run of a pull's pre-flight, `check`, at the same /// head and base, that is running or has passed: then it is not run /// here too. Its check here is taken back rather than filed as /// cancelled, which would be the newest run of it the merge queue /// reads, and run again. One that cannot be read is run here. async fn run_elsewhere( &self, watch: &Watch, subject: &Subject, check: &Check, ) -> Option { let first = check.queue.as_deref() == Some(QUEUE_FIRST); let pull = (subject.pull.as_ref()).filter(|p| first && !p.others.is_empty())?; let filter = checks::Filter { pull: Some(&pull.uri), commit: Some(&subject.commit), workflow: Some(&check.workflow), ..checks::Filter::default() }; let (remote, key) = (&watch.remote, &watch.key); let read = checks::by(&self.ctx, remote, key, &pull.others, filter, false); let base = Some(pull.base_commit.as_str()); let ran = |c: &checks::Check| { [status::RUNNING, status::SUCCESS].contains(&c.record.status.as_str()) }; let theirs = read.await.unwrap_or_default().into_iter(); theirs .filter(ran) .find(|c| c.record.base_commit.as_deref() == base) } /// Keeps a trusted pull's run of `wf` that passed, filed under `rkey`, /// for a push at its tree to reuse; and a trusted batch's, for the push /// that lands it. fn keep_passed( &self, watch: &Watch, subject: &Subject, wf: &Workflow, rkey: &str, ) -> Result<()> { let base = match (&subject.pull, &subject.batch) { (Some(pull), _) => &pull.base, (None, Some(batch)) if batch.trusted => &subject.ref_name, _ => return Ok(()), }; let Some(tree) = &subject.tree else { return Ok(()); }; let pull = match &subject.pull { Some(pull) => pull.uri.clone(), None => format!("{BATCH}{base}"), }; let mut state = self.state.lock().unwrap(); let kept = state.passed.entry(watch.id()).or_default(); // The layers of the batches before on this branch: the push that // landed each has taken them. Not the one this batch runs ahead // on, which may not have landed yet. let batch = subject.pull.is_none(); let on = format!("{} ", subject.base_commit().unwrap_or_default()); let older = |p: &Passed| batch && p.pull == pull && p.workflow == wf.name && !p.at.starts_with(&on); let superseded: Vec = kept.iter().filter(|p| older(p)).map(Passed::rkey).collect(); kept.retain(|p| !older(p) && (&p.tree, &p.workflow, &p.base) != (tree, &wf.name, base)); kept.push(Passed { tree: tree.clone(), workflow: wf.name.clone(), base: base.clone(), rkey: rkey.to_string(), pull, at: subject.at(), }); let over = kept.len().saturating_sub(PASSED_MAX); kept.drain(..over); store::write(&self.state_path, &*state, false)?; drop(state); self.caches.forget(&superseded, self.opts.cache_max) } /// The pull runs that passed at `subject`'s tree, into its ref, for a /// push. fn passed_at(&self, watch: &Watch, subject: &Subject) -> Vec { let Some(tree) = subject.tree.as_ref().filter(|_| subject.pull.is_none()) else { return Vec::new(); }; let state = self.state.lock().unwrap(); let kept = state.passed.get(&watch.id()).into_iter().flatten(); kept.filter(|p| &p.tree == tree && p.base == subject.ref_name) .cloned() .collect() } /// Whether `check`, of `wf` at a push, is passed as one of the pull runs /// `passed` at its tree did, as that run's check still says: filled in /// from it if so, its log saying whose it is, and what that run wrote /// to its caches taken as the push's own run's would be, where it can /// be: a push that runs `wf`, as this one does, keeps what its run /// writes ([`promotes`]), to the caches of the branch the pulls it /// takes from ran into. Never one that asks for secrets, which the /// pull's run did not get, nor one whose runs finish, which a push /// runs for what it does, as a deploy. async fn reused( &self, watch: &Watch, passed: &[Passed], wf: &Workflow, check: &mut Check, ) -> Result { let reuses = reusable(wf); let Some(p) = passed.iter().find(|p| p.workflow == wf.name && reuses) else { return Ok(false); }; let earlier = self.reporter.get(&watch.remote, &p.rkey).await?; let Some(earlier) = earlier.filter(|c| c.status == status::SUCCESS) else { return Ok(false); }; // One its paths left out read nothing that changed from its base: // it stands for the push only once the base's own run passed. if earlier.skipped.is_some() && !self.passed_on(watch, p, &earlier, wf).await? { return Ok(false); } let uri = format!( "at://{}/{}/{}", self.reporter.identity.did, check::NSID, p.rkey ); let (tree, wf_name) = (&p.tree, &wf.name); let base = earlier.base_commit.unwrap_or_default(); let mut line = match earlier.pull { Some(pull) => format!( "==> passed: the same tree, {tree}, as pull {pull} at {head} merged with {base}, whose run of {wf_name} passed ({uri}); not run again\n", head = earlier.commit, ), None => format!( "==> passed: the same tree, {tree}, as the merge queue's batch at {tip} on {base}, whose run of {wf_name} passed ({uri}); not run again\n", tip = earlier.commit, ), }; let taken = self.caches.take(&p.rkey); line += &taken_lines(&taken.kept(), &taken.moved, !wf.cache.is_empty()); taken.lease.promote(self.opts.cache_max).await?; check.status = status::SUCCESS.into(); check.steps = earlier.steps; check.outputs = earlier.outputs; // Its log copies the earlier one's, which may be pruned; one its // paths left out has none. let before = match &earlier.log { Some(log) => self.reporter.log_of(&watch.remote, log).await, None => Ok(earlier .skipped .as_ref() .map(|why| format!("==> {why}\n").into_bytes()) .unwrap_or_default()), }; let log = check::copied(&line, &before.unwrap_or_default()); check.log = Some(self.reporter.upload_log(&log).await?); check.finished_at = Some(now()); Ok(true) } /// Whether `wf` passed on `p`'s base branch at the commit `earlier`, a /// run there that `p` keeps, was checked on: a push's own run there. async fn passed_on( &self, watch: &Watch, p: &Passed, earlier: &Check, wf: &Workflow, ) -> Result { let base = earlier.base_commit.as_deref().unwrap_or_default(); let (owner, key) = (watch.remote.did(), &watch.key); let rkey = check::rkey(owner, key, &p.base, base, None, &wf.name); let there = self.reporter.get(&watch.remote, &rkey).await?; Ok(there.is_some_and(|c| c.status == status::SUCCESS)) } /// Waits for what a run ranked `rank` holds while its steps go, or /// `None` if `cancelled` first. One of a workflow whose runs finish /// first has its turn on its ref or pull, `on_ref`, and is kept from /// then on; then every run waits for a job. Its caches it shares, each /// run through overlays of its own. The turn on a ref is waited for /// only by runs that hold nothing, so no two runs wait on each other. async fn take_turns( &self, rank: Rank, watch: &Watch, on_ref: Option, cancelled: &mut Cancelled, ) -> Option { let mut kept = None; if let Some(turn) = on_ref { kept = Some(until_cancelled(cancelled, turn).await?); cancelled.kept = true; } let (whose, over) = self.whose(watch); let began = std::time::Instant::now(); let turn = until_cancelled(cancelled, self.turns.clone().take(rank, whose, Vec::new())); let turn = turn.await?; let mut said = waited(over, began.elapsed()); // Holding its job, so that no other run starts meanwhile. let room = room(&self.caches, self.opts.disk_floor, ROOM_EVERY); said += &until_cancelled(cancelled, room).await?; Some(Held { on_ref: kept, job: turn, said, }) } /// Whose a run of `watch` is, and the allowance it is past, if it is: /// one past it holds one job at a time, the rest `--owner-jobs`, or /// one `--unmetered` all of them. fn whose(&self, watch: &Watch) -> (Whose, Option) { let (owner, private) = (watch.remote.did(), watch.remote.space.is_some()); let mut state = self.state.lock().unwrap(); let used = state .usage(private) .get(owner) .map_or(0, |u| spent(u, today())); let over = past(used, self.allowance(owner, private)); let unmetered = self.opts.unmetered.iter().any(|d| d == owner); let share = if unmetered { self.opts.jobs } else { self.opts.owner_share() }; let most = most(over, share); let owner = owner.to_string(); (Whose { owner, most }, over) } /// Why the operator paused `watch`'s owner's runs, if they did: a /// `--paused` file that cannot be read pauses none. fn paused(&self, watch: &Watch) -> Result> { let path = self.opts.paused.clone().unwrap_or_default(); let list = std::fs::read_to_string(path).unwrap_or_default(); paused_why(&list, watch.remote.did(), &self.reporter.identity.did) } /// The allowance of `owner`'s public or `private` repositories' runs: /// none for the runner's own account, or one `--unmetered`. fn allowance(&self, owner: &str, private: bool) -> Option { let allowance = if private { self.opts.private_allowance } else { self.opts.allowance }; let mine = owner == self.reporter.identity.did; let free = mine || self.opts.unmetered.iter().any(|d| d == owner); if free { None } else { allowance } } /// Counts the time `check`'s steps took against the owner of `watch`, /// and gives the owner's minutes record when it is to be written: of /// public repositories only, as private ones' minutes are not shown /// to everyone. fn use_steps(&self, watch: &Watch, check: &Check) -> Result> { let (owner, private) = (watch.remote.did(), watch.remote.space.is_some()); let ms = check.steps.iter().map(|s| s.duration_ms).sum(); let mut state = self.state.lock().unwrap(); let used = state.usage(private).entry(owner.to_string()).or_default(); spend(used, today(), ms); 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)?; let mut published = self.published.lock().unwrap(); let last = published.get(owner).copied(); if private || !due(last, now) { return Ok(None); } published.insert(owner.to_string(), (std::time::Instant::now(), now)); Ok(Some(record)) } /// Files `check` as cancelled, `why`, while it waited its turn. async fn never_started( &self, watch: &Watch, rkey: &str, mut check: Check, why: &str, ) -> Result { let line = format!("==> cancelled {why} before it started\n"); check.status = status::CANCELLED.into(); check.log = Some(self.reporter.upload_log(line.as_bytes()).await?); check.finished_at = Some(now()); self.reporter.put(&watch.remote, rkey, &check).await?; Ok(check.status) } /// Whether this runner lands the merge queues of `watch`, whose /// record is `record`: asked to, and the owner or a collaborator, or /// the runner it names to land them ([`RepoRecord::lander`]), which /// lands through the gateway. fn lands(&self, record: &RepoRecord, watch: &Watch) -> bool { let (me, owner) = (&self.reporter.identity.did, watch.remote.did()); let named = record.lander(owner) == Some(me.as_str()); self.opts.queue && (writes(record, owner, me) || named) } /// Watches `watch` from now on: its door, and the loops that ring it /// and answer it, which end once it is left. fn open_door(self: &Arc, watch: Watch) -> Arc { let door = Arc::new(Notify::new()); let id = (watch.remote.did().to_string(), watch.key.clone()); self.doors.lock().unwrap().insert(id, door.clone()); tokio::spawn(notice(self.clone(), watch.clone(), door.clone())); if let Some(space) = &watch.remote.space { let mut private = self.private.lock().unwrap(); private.insert(space.uri.clone(), door.clone()); tokio::spawn(poll(self.clone(), watch.clone(), door.clone())); } tokio::spawn(queue(self.clone(), watch, door.clone())); door } /// Stops watching `watch`, found naming this runner, whose record no /// longer does, unless it is watched by another name now. What it /// runs meanwhile finishes. fn leave(&self, watch: &Watch) -> Result<()> { let id = (watch.remote.did().to_string(), watch.key.clone()); // One watched by another name now is that one's to leave. let same = |w: &&Watch| w.remote.url.name == watch.remote.url.name; let left = self.named.lock().unwrap().get(&id).filter(same).cloned(); let Some(left) = left else { return Ok(()) }; self.forget(&left) } /// Stops watching `watch`, however it came to be watched: its loops /// end, and what it runs meanwhile finishes. fn forget(&self, watch: &Watch) -> Result<()> { watch.gone.store(true, Ordering::SeqCst); let id = (watch.remote.did().to_string(), watch.key.clone()); self.named.lock().unwrap().remove(&id); if let Some(door) = self.doors.lock().unwrap().remove(&id) { door.notify_one(); } if let Some(space) = &watch.remote.space { self.private.lock().unwrap().remove(&space.uri); } let mut state = self.state.lock().unwrap(); state.discovered.remove(&watch.label); state.schedules.remove(&watch.id()); store::write(&self.state_path, &*state, false)?; telemetry::left(&watch.label); Ok(()) } /// Watches `did`'s repository `name` if its record names this runner /// and it is not watched already by that name: by another, it was /// renamed, and is watched by this one instead. async fn consider(self: &Arc, did: &str, name: &str) -> Result<()> { let _one = self.considering.lock().await; let url = RepoUrl { authority: did.to_string(), name: name.to_string(), }; let remote = Remote::open(&self.ctx, &url).await?; let snap = remote.snapshot(&self.ctx).await?; let me = &self.reporter.identity.did; // One with no record, or whose record does not name this runner, // is passed over. let key = snap.key().map(str::to_string); let named = key.filter(|key| names(snap.repo.as_ref(), key, me)); let Some(key) = named else { return Ok(()) }; let id = (did.to_string(), key.clone()); let was = self .named .lock() .unwrap() .get(&id) .map(|w| w.remote.url.name.clone()); let watched = self.doors.lock().unwrap().contains_key(&id); // One on the command line is watched as it is. if watched && was.as_deref().is_none_or(|was| was == name) { return Ok(()); } let label = format!("{did}/{name}"); let watch = Watch { label: label.clone(), remote, key, discovered: true, pulls_only: false, gone: Arc::default(), stage: Arc::default(), }; // Watched by its old name until now, that one's loops end. let old = self.named.lock().unwrap().insert(id, watch.clone()); old.inspect(|w| w.gone.store(true, Ordering::SeqCst)); { let mut state = self.state.lock().unwrap(); if let Some(was) = was { state.discovered.remove(&format!("{did}/{was}")); } state.discovered.insert(label.clone()); store::write(&self.state_path, &*state, false)?; } telemetry::found(&label); report(&watch, self.recover(&watch).await); self.open_door(watch).notify_one(); Ok(()) } /// Considers each repository the appview lists as naming this runner, /// and each found before. async fn rediscover(self: &Arc) -> Result<()> { let mut labels = self.state.lock().unwrap().discovered.clone(); let appview = self.ctx.env.appview.clone().unwrap_or_default(); let xrpc = knotless::xrpc::Xrpc::new(self.ctx.http.clone(), &appview); let me = self.reporter.identity.did.as_str(); // The page to ask for next, if any: the first, from no cursor. let mut next: Option> = Some(None); while let Some(cursor) = next { let mut params = vec![("runner", me), ("limit", "500")]; params.extend(std::iter::repeat("cursor").zip(cursor.as_deref())); let page: Listing = xrpc.query("dev.knotless.listRepos", ¶ms, None).await?; labels.extend(page.repos.iter().map(|r| format!("{}/{}", r.did, r.name))); next = page.cursor.map(Some); } for label in labels { let (did, name) = label.split_once('/').unwrap_or_default(); noted(&label, self.consider(did, name).await); } Ok(()) } /// Considers each private repository that invited this runner since /// the invites last read: nothing written in a space reaches Jetstream /// or the appview's index ([`knotless::invites`]). async fn invited(self: &Arc) -> Result<()> { let me = session::ensure(&self.ctx, &self.reporter.identity).await?; loop { let cursor = self.state.lock().unwrap().invited.clone(); let page = knotless::invites::list(&self.ctx, &me, cursor.as_deref()).await?; for invite in &page.invites { let label = format!("{}/{}", invite.did, invite.name); noted(&label, self.consider(&invite.did, &invite.name).await); } let mut state = self.state.lock().unwrap(); state.invited = page.cursor.or(cursor); store::write(&self.state_path, &*state, false)?; if page.invites.is_empty() { return Ok(()); } } } /// With `--discover`, considers each repository whose record /// `event` writes, which may name this runner now. fn found(self: &Arc, event: &jetstream::Event) { let Some(did) = event.did.clone().filter(|_| self.opts.discover) else { return; }; let written = event.ops.iter().filter(|op| op.record.is_some()); for op in written.filter(|op| op.collection == nsid::REPO) { let (runner, did, name) = (self.clone(), did.clone(), op.rkey.clone()); tokio::spawn(async move { let label = format!("{did}/{name}"); noted(&label, runner.consider(&did, &name).await); }); } } /// Rings `watch`'s door, when it is being watched. fn ring(&self, watch: &Watch) { let doors = self.doors.lock().unwrap(); let door = doors.get(&(watch.remote.did().to_string(), watch.key.clone())); door.inspect(|d| d.notify_one()); } /// Notes a run of `wf` at `subject` that had not started as dropped, /// for the runner started next to run it, and files its check as /// cancelled, which a merge takes as not run: run, it is filed again /// under the same key. Not deleted: a deletion's commit can need MST /// blocks a PDS leaves out of it, and every relay and indexer that /// cannot check one fetches the whole repository again, all at once, /// which held pds.js's account for a minute or two after each deploy. async fn drop_run( &self, watch: &Watch, subject: &Subject, wf: &Workflow, rkey: &str, check: Check, ) -> Result { { let mut state = self.state.lock().unwrap(); 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.ask_again(watch, subject, wf)?; self.never_started(watch, rkey, check, WAITED).await } /// 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) { 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; } tokio::select! { _ = stopping.changed() => {} _ = finishing.changed() => {} } } } /// Removes the containers of this runner's no run of it has, as one /// that leaked, or that a runner before it left, does. async fn remove_left(&self) { let me = &self.reporter.identity.did; let work = self.opts.work.as_deref(); self.opts .executor .remove_left(me, &self.sessions, work) .await } /// Waits until no run is going. pub async fn idle(&self) { let _ = self.runs.acquire_many(u32::MAX).await; } /// Stops: reads no more, drops each run that has not started its /// steps, to run when the runner starts again, and waits for those /// that have, but for those asked for by name: they are cancelled, to /// be asked for again. It waits for [`Options::stop_wait`] at most: /// then each still going is cancelled, and ends. A stop never waits on /// a run that hangs, as one did for good when nothing came to kill the /// runner. pub async fn stop(&self) { self.stopping.send_replace(true); drop(self.reconciling.write().await); for going in self.going.lock().unwrap().values() { if going.asked { cancel(&going.cancel, STOPPED); } } if tokio::time::timeout(self.opts.stop_wait, self.idle()) .await .is_err() { for going in self.going.lock().unwrap().values() { cancel(&going.cancel, CUT_OFF); } self.idle().await; } } /// Ends each check of `watch` this account left queued or running: /// whatever ran it stopped before it finished. 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 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}")); check.finished_at = Some(now()); self.reporter.put(&watch.remote, &rkey, &check).await?; } Ok(()) } /// Checks out the commit and runs the steps, stopping at the first that /// fails, into `check`. async fn steps( &self, watch: &Watch, git: &Git, subject: &Subject, wf: &Workflow, check: &mut Check, started: Started<'_>, ) -> Result<()> { let Started { finishing, job, cancelled, said, } = started; // Each step that uses a shared step is put in place of by its // steps, as the repository's lock pins them, before anything runs. let expanded = knotless::uses::expanded(&self.ctx, git, subject.checkout(), wf).await?; let wf = &expanded.workflow; let (vars, granted, secrets, ungranted) = self.vars(watch, subject, wf).await?; let dir = match &self.opts.work { Some(work) => { std::fs::create_dir_all(work)?; tempfile::tempdir_in(work)? } None => tempfile::tempdir()?, }; let src = dir.path().join("src"); checkout(git, &src, subject.checkout()).await?; // Each shared step's repository, at the commit the lock pins, beside // the checkout, for its steps to find what it keeps beside them. let shared = dir.path().join("steps"); let mut step_dirs = Vec::new(); for (i, source) in expanded.sources.iter().enumerate() { let at = shared.join(i.to_string()); checkout(&source.git, &at, &source.commit).await?; step_dirs.push(self.opts.executor.step_dir(i, &at)); } let log_path = dir.path().join("log"); let mut log = std::fs::File::create(&log_path)?; log.write_all(said.as_bytes())?; if subject.asked.contains_key(&wf.name) { let scheduled = given(subject, wf).any(|(k, _)| k == SCHEDULE_VAR); let why = if scheduled { "on its schedule" } else { "asked for by name" }; let given = given(subject, wf).map(|(k, v)| format!(" {k}={v}")); writeln!(log, "==> {why}{}", given.collect::())?; } log.write_all(not_granted(&ungranted, watch, subject).as_bytes())?; let pull = subject.pull.as_ref().map(|p| &p.strangers); let strangers = pull.or(subject.batch.as_ref().map(|b| &b.strangers)); if let Some(who) = strangers.filter(|s| !s.is_empty()) { let who = who.join(", "); let said = format!( "==> runs as code no one vouches for, as its lock adds shared steps from {who}\n" ); log.write_all(said.as_bytes())?; } // What it writes goes to the appview as it is written, ending // once all of it is, before it is filed. let rkey = run_rkey(watch, subject, &wf.name); let uri = self.reporter.uri_of(&watch.remote, &rkey); let (progress, progressed) = tokio::sync::watch::channel(live::Progress::default()); let (ending, ended) = tokio::sync::watch::channel(false); let sending = self.appview.clone().map(|appview| { let (log, secrets) = (log_path.clone(), secrets.clone()); let started = check.started_at.clone(); tokio::spawn(live::send( appview, log, uri, started, secrets, progressed, ended, )) }); let outputs = dir.path().join("outputs"); std::fs::create_dir(&outputs)?; let branch = own_branch(subject, wf); let wanted = caches(&watch.id(), kept_apart(subject, wf), wf, branch); let lease = self.caches.lease(&wanted)?; let caches = lease.mounts(); let trusts = self.opts.trust.contains(&watch.remote.did().to_string()); let setup = executor::Setup { dir: &src, vars: &vars, image: wf.image.as_deref(), caches: &caches, outputs: &outputs, steps: (!step_dirs.is_empty()).then_some(shared.as_path()), owner: executor::Owner { runner: &self.reporter.identity.did, run: &run_rkey(watch, subject, &wf.name), }, // What deploys wait for goes first: pushes to the default // branch, the merge queue's next batch among them. behind: !subject.first(), vouched: trusts && vouched_for(subject), }; let timeout = self.opts.step_timeout; let _tracked = Tracked::new(self, setup.owner.run); // A start that fails says so in the log its error points at, and // the log is kept before the check is filed as an error. let session = match self.opts.executor.start(&setup, &log, timeout).await { Ok(session) => session, Err(err) => { // The start's error is what the check says, whatever // keeping the log comes to. let _ = log.write_all(format!("==> {err:#}\n").as_bytes()); let text = std::fs::read(&log_path).unwrap_or_default(); let text = check::redact(&text, &secrets); check.log = self.reporter.upload_log(&text).await.ok(); return Err(err); } }; // A run on its schedule is held to the runner's limits, as one // that recurs on its own: only its owner's own ask goes past them. let scheduled = given(subject, wf).any(|(k, _)| k == SCHEDULE_VAR); let asked = subject.asked.contains_key(&wf.name) && !scheduled; let steps = Steps { log, progress: &progress, granted: &granted, step_dirs: &step_dirs, at: ref_at(subject), }; let ran = (self.run_steps(&session, wf, check, steps, cancelled, asked)).await; session.stop().await; // Its job is free once its container is gone: what is left is // writing it down, which a slow PDS may keep waiting. The note // that it goes is let go before its turn is, so the runs it kept // see it end first. drop(finishing); drop(job); ran?; let ended = |status: &str| check.steps.iter().any(|s| s.status == status); check.status = if ended(status::CANCELLED) { status::CANCELLED } else if ended(status::FAILURE) { status::FAILURE } else { status::SUCCESS } .into(); // What a push that passed wrote, to the default branch or to one // whose caches are its own, is what the runs after it start from; // a trusted pull's or batch's that passed is kept, for the push // that takes its check to take, or, where no push runs, is what the // runs after it start from, as a push's would be; any other run's // goes with it. A batch, not landed yet, never writes there. let passed = check.status == status::SUCCESS; let mut warmed = String::new(); if passed && promotes(subject, wf) { lease.promote(self.opts.cache_max).await?; } else if let Some(at) = warms(watch.pulls_only, subject, wf).filter(|_| passed) { let (kept, moved) = lease.advance(self.opts.cache_max, at).await?; warmed = warmed_lines(&kept, &moved); } else if passed && keeps(subject, wf) && !gated(check) && !watch.pulls_only { let rkey = run_rkey(watch, subject, &wf.name); lease.keep(&rkey, self.opts.cache_max).await?; } let mut noted = std::fs::OpenOptions::new().append(true).open(&log_path)?; noted.write_all(warmed.as_bytes())?; // One asked for by name that passed writes back what its workflow // says it may; one that cannot fails, saying why. if passed && asked && !wf.commits().is_empty() { match self.write_back(watch, git, &src, subject, wf).await { Ok(said) => writeln!(noted, "==> {said}")?, Err(err) => { let why = format!("not written back: {err:#}"); writeln!(noted, "==> {why}")?; check.status = status::FAILURE.into(); check.error = Some(why); } } } check.outputs = self.outputs(&outputs, &mut noted).await?; let _ = ending.send(true); if let Some(sending) = sending { let _ = tokio::time::timeout(live::LAST, sending).await; } let log = check::redact(&std::fs::read(&log_path)?, &secrets); check.log = Some(self.reporter.upload_log(&log).await?); Ok(()) } /// Commits what the run of `wf` at `subject`, asked for by name, /// changed in the checkout `src` among the files its workflow names /// ([`Workflow::commits`]) onto the branch it ran at, and pushes it as /// this runner's account. Says what it wrote. async fn write_back( &self, watch: &Watch, git: &Git, src: &Path, subject: &Subject, wf: &Workflow, ) -> Result { let branch = subject.ref_name.strip_prefix("refs/heads/"); let branch = branch.context("only a run at a branch writes back")?; let globs = wf.commits(); let tree = src.to_string_lossy(); let mut work = Git::new(src.join(".git")); // Each path as it is, whatever it holds. work.envs.push(("GIT_LITERAL_PATHSPECS".into(), "1".into())); let args = [ "status", "--porcelain", "-z", "--untracked-files=all", "--no-renames", ]; let args = [&["--work-tree", &tree][..], &args].concat(); let status = work.run(&args, None).await?; let changed: Vec = changed(&status) .into_iter() .filter(|f| globs.iter().any(|g| workflow::glob(g, f))) .collect(); if changed.is_empty() { return Ok(format!("nothing to write back in {}", globs.join(", "))); } let mut add = vec!["--work-tree", &tree, "add", "-A", "--"]; add.extend(changed.iter().map(String::as_str)); work.run(&add, None).await?; let email = knotless::emails::own(&self.reporter.identity.did); for var in ["AUTHOR", "COMMITTER"] { work.envs .push((format!("GIT_{var}_NAME"), "knotless-run".into())); work.envs.push((format!("GIT_{var}_EMAIL"), email.clone())); } let written = work.text(&["write-tree"], None).await?; let message = format!( "{}: what it wrote, as asked\n\nRun at {} when asked for by name, it changed {} under {}.", wf.name, short(&subject.commit), plural(changed.len()), globs.join(", ") ); let commit = [ "commit-tree", &written, "-p", &subject.commit, "-m", &message, ]; let sha = work.text(&commit, None).await?; // Into the mirror, which pushes what it has. work.run(&["update-ref", "refs/heads/written", &sha], None) .await?; let from = src.join(".git"); let fetch = ["fetch", "-q", &from.to_string_lossy(), "refs/heads/written"]; git.run(&fetch, None).await?; let url = &watch.remote.url; forge::push_back(&self.ctx, url, git, branch, &subject.commit, &sha).await?; Ok(format!( "wrote back {} to {branch} as {}", plural(changed.len()), short(&sha) )) } /// What the steps left in `dir` to be looked at, as outputs: each /// image, in the order of their names, up to [`MOST_OUTPUTS`] of at /// most [`OUTPUT_SIZE`], named for what they show by their files' /// names. Each file left out is a line in `log`, saying why. /// /// A PDS can take seconds over each blob, so each image is uploaded /// once however many files hold it, and [`UPLOADS`] at a time. async fn outputs(&self, dir: &Path, log: &mut impl Write) -> Result> { let mut paths: Vec = std::fs::read_dir(dir)? .map(|e| e.map(|e| e.path())) .collect::>()?; paths.sort(); let mut kept = Vec::new(); for path in paths { let name = path.file_name().unwrap_or_default().to_string_lossy(); let Some(mime) = image_type(&path).filter(|_| path.is_file()) else { writeln!(log, "==> {name}: not kept, as only images are outputs")?; continue; }; let bytes = std::fs::read(&path)?; if bytes.len() > OUTPUT_SIZE { writeln!(log, "==> {name}: not kept, being over {OUTPUT_SIZE} bytes")?; continue; } if kept.len() == MOST_OUTPUTS { writeln!(log, "==> {name}: not kept, past {MOST_OUTPUTS} outputs")?; continue; } let stem = path.file_stem().unwrap_or_default().to_string_lossy(); let alt: String = stem.chars().take(300).collect(); kept.push((bytes, mime, alt)); } // The distinct images, and which of them each kept file is. let mut images: Vec<(&[u8], &str)> = Vec::new(); let mut which = Vec::new(); for (bytes, mime, _) in &kept { let image = (bytes.as_slice(), *mime); let seen = images.iter().position(|i| *i == image); which.push(seen.unwrap_or(images.len())); if seen.is_none() { images.push(image); } } // Made before they are awaited: a closure over borrowed images, // held across an await, keeps the spawned run from being `Send`. let uploads: Vec<_> = images .into_iter() .map(|(bytes, mime)| self.reporter.upload(bytes.to_vec(), mime)) .collect(); let blobs: Vec<_> = futures::stream::iter(uploads) .buffered(UPLOADS) .try_collect() .await?; let outputs = kept.into_iter().zip(which).map(|((_, _, alt), i)| { let image = blobs[i].clone(); CheckOutput::Image(CheckImage { image, alt }) }); Ok(outputs.collect()) } /// Runs `wf`'s steps in `session`, writing to `log`, and records each /// in `check`: once one fails or is cancelled, the rest are skipped. async fn run_steps( &self, session: &Session, wf: &Workflow, check: &mut Check, steps: Steps<'_>, mut cancelled: Cancelled, asked: bool, ) -> Result<()> { let Steps { mut log, progress, granted, step_dirs, at, } = steps; let mut failed = false; let begun = std::time::Instant::now(); let whole = self.opts.run_timeout.filter(|_| !asked); let shared = wf.steps.iter().any(|s| s.from.is_some()); for step in &wf.steps { let left = whole.map(|run| run.saturating_sub(begun.elapsed())); let (limit, last) = within(self.limit(step, asked), left); if failed { check.steps.push(skipped_step(step)); continue; } writeln!(log, "==> {}", step.name)?; // One whose `on` the run's ref does not match passes unrun. if let Some(refs) = step.unmatched(at.ok()) { let why = at.map_or_else(str::to_string, |r| { format!("{r} is not {}", refs.join(" or ")) }); writeln!(log, "==> skipped, as {why}")?; check.steps.push(skipped_step(step)); let steps = check.steps.clone(); progress.send_replace(live::Progress { steps, going: None }); continue; } progress.send_modify(|p| p.going = Some(step.name.clone())); telemetry::step(&check.commit, &wf.name, &step.name, None, None, None); let own = step.env.iter().map(|(k, v)| (k.clone(), v.clone())); // Each by the name it sees it by, a shared step's or the // workflow's, from the name the run is granted it by. let seen = step.seen(granted); let from = step .from .map(|i| ("KNOTLESS_STEP_DIR".into(), step_dirs[i].clone())); let given = !seen.is_empty(); let own: Vars = own.chain(seen).chain(from).collect(); // Beside shared steps, what a step given secrets leaves running // is ended once it ends, so none holds them for the next. let before = match shared && given { true => session.running().await, false => None, }; let cmd = session.command(&step.run, &own); // The run's task holds the sender to its end, so this ends only // when it is set. let cancel = cancelled.wait(); let exit = executor::run(cmd, &log, limit, async { let _ = cancel.await; }) .await?; if exit.cancelled { writeln!(log, "==> cancelled {}", cancelled.why())?; } else if exit.code.is_none() { let said = killed(limit, last.then_some(whole).flatten()); writeln!(log, "==> {said}")?; } session.end_left(before.as_ref(), &step.name, &log).await?; failed = !exit.success(); let ended: &str = match (exit.cancelled, failed) { (true, _) => status::CANCELLED, (false, true) => status::FAILURE, (false, false) => status::SUCCESS, }; let took = Some(exit.elapsed); telemetry::step( &check.commit, &wf.name, &step.name, Some(ended), exit.code, took, ); check.steps.push(check::Step { name: step.name.clone(), status: ended.into(), exit_code: exit.code, duration_ms: u64::try_from(exit.elapsed.as_millis()).unwrap_or(u64::MAX), }); let steps = check.steps.clone(); progress.send_replace(live::Progress { steps, going: None }); } Ok(()) } /// How long `step` may take: its own `timeout`, within the runner's /// limit or, in a run its owner `asked` for, past it. fn limit(&self, step: &workflow::Step, asked: bool) -> Duration { let max = self.opts.step_timeout; let own = step .timeout .map(|m| Duration::from_secs(m.saturating_mul(60))); own.filter(|own| asked || *own < max).unwrap_or(max) } /// What a step sees beyond the runner's tools: the workflow's own /// variables, those a run asked for by name was given, and the run's; /// apart from them, the secrets both the workflow asks for and the /// runner grants this repository, in its secrets file or sealed to it /// by the owner, never to a pull's run, each given only to the steps /// that may see it; those secrets' values, to keep out of the log; and /// the names of those it asks for and is not given. async fn vars( &self, watch: &Watch, subject: &Subject, wf: &Workflow, ) -> Result<(Vars, Vars, Vec, Vec)> { let mut env = wf.env.clone(); env.extend(given(subject, wf).cloned()); let mut vars: Vars = env.into_iter().collect(); let (mut secrets, mut given, mut granted_vars) = (Vec::new(), Vec::new(), Vars::new()); let pull = subject.pull.as_ref().map(|p| p.uri.as_str()); for (k, v) in [ ("KNOTLESS_RUN", Some("1")), ("KNOTLESS_RUN_REPO", Some(watch.label.as_str())), ("KNOTLESS_RUN_REF", Some(subject.ref_name.as_str())), ("KNOTLESS_RUN_SHA", Some(subject.commit.as_str())), ("KNOTLESS_RUN_PULL", pull), ("KNOTLESS_RUN_BASE", subject.base_commit()), ] { vars.extend(v.map(|v| (k.to_string(), v.to_string()))); } // A workflow that asks for none runs whatever state the file is in. if let Some(path) = self .opts .secrets .as_ref() .filter(|_| !wf.secrets.is_empty() && !subject.proposed()) { let mut all: BTreeMap> = store::read(path) .with_context(|| format!("reading the runner's secrets, {}", path.display()))?; let mut granted = all.remove(&watch.label).unwrap_or_default(); for name in &wf.secrets { if let Some(value) = granted.remove(name).and_then(|s| s.to(&subject.ref_name)) { secrets.push(value.clone()); granted_vars.push((name.clone(), value)); given.push(name); } } } // Those the file does not grant, the owner may have sealed to it. let missing: Vec<&String> = wf.secrets.iter().filter(|n| !given.contains(n)).collect(); let sealed = !missing.is_empty() && !subject.proposed() && self.ctx.env.appview.is_some(); if sealed { for (name, value) in self.sealed(watch, subject, &missing).await? { secrets.push(value.clone()); granted_vars.push((name.clone(), value)); given.push(name); } } let ungranted = wf.secrets.iter().filter(|n| !given.contains(n)); Ok((vars, granted_vars, secrets, ungranted.cloned().collect())) } /// Whether `wf` is this runner's to run: it names no runners, or this /// one, by DID or by a handle that resolves to it. async fn named(&self, wf: &Workflow) -> bool { let me = &self.reporter.identity.did; for runner in &wf.runners { let did = match runner.starts_with("did:") { true => Some(runner.clone()), false => knotless::identity::resolve(&self.ctx, runner) .await .ok() .map(|i| i.did), }; if did.as_ref() == Some(me) { return true; } } wf.runners.is_empty() } /// Those of `names` the repository's owner sealed to this runner and /// left at the appview, opened, where they reach `subject`'s ref. One /// that does not open, sealed for somewhere else or to a key it no /// longer holds, is not given. async fn sealed<'a>( &self, watch: &Watch, subject: &Subject, names: &[&'a String], ) -> Result> { let me = session::ensure(&self.ctx, &self.reporter.identity).await?; let owner = watch.remote.did(); let left = knotless::secrets::list(&self.ctx, &me, owner, &watch.key).await; let left = left.context("reading the secrets its owner sealed to this runner")?; let opened = left.into_iter().filter_map(|s| { let name = names.iter().find(|n| ***n == s.name)?; let refs = &s.refs; let reaches = refs.is_empty() || refs.iter().any(|p| workflow::matches(p, &subject.ref_name)); let bound = sealed::Bound { owner, repo: &watch.key, runner: &self.reporter.identity.did, name: &s.name, refs, }; let value = self.keys.open(&s.key, &bound, s.sealed.as_deref()?).ok()?; reaches.then(|| (*name, String::from_utf8_lossy(&value).into_owned())) }); Ok(opened.collect()) } } /// What a run's log says of each secret its workflow asks for and it is /// not given, which would otherwise be simply absent: a pull's or a /// batch's run gets none, and a push gets only those the runner grants /// its repository at its ref. fn not_granted(names: &[String], watch: &Watch, subject: &Subject) -> String { let to = if subject.proposed() { "a pull's or a merge-queue batch's run, which gets no secrets".to_string() } else { format!("{} at {}", watch.label, subject.ref_name) }; let lines = names .iter() .map(|n| format!("==> {n} is not granted to {to}\n")); lines.collect() } /// Takes the grants to `label` out of the secrets file at `path`, if it /// has any, keeping the rest as they are. fn ungrant(path: &std::path::Path, label: &str) -> Result<()> { let all: Option> = store::read(path); let mut all = all.unwrap_or_default(); if all.remove(label).is_some() { store::write(path, &all, true)?; } Ok(()) } /// A secret the runner grants a repository: to every push, or only to /// pushes to the refs named. #[derive(Deserialize)] #[serde(untagged)] enum Secret { Any(String), To { value: String, refs: Vec }, } impl Secret { /// Its value, if a push to `ref_name` is granted it. fn to(self, ref_name: &str) -> Option { match self { Secret::Any(value) => Some(value), Secret::To { value, refs } => refs .iter() .any(|p| workflow::matches(p, ref_name)) .then_some(value), } } } /// Why a pull whose head is at `head` failed against `base` at `tip`: it /// conflicts in `files`. fn conflict(head: &str, base: &str, tip: &str, files: &[String]) -> String { let into = base.trim_start_matches("refs/heads/"); let (head, tip) = (short(head), short(tip)); let files = files.join(", "); format!( "the head, {head}, conflicts with {into} at {tip} in {files}: merge {into} into it, or rebase it onto {into}" ) } /// The key the check of a run of `workflow` at `subject` is filed by. fn run_rkey(watch: &Watch, subject: &Subject, workflow: &str) -> String { let base = subject.base_commit(); let (owner, keyed) = (watch.remote.did(), subject.keyed()); let keyed = keyed.as_str(); check::rkey(owner, &watch.key, keyed, &subject.commit, base, workflow) } /// Whether `did` may push to the repository `record` is of, whose owner is /// `owner`: its owner, a collaborator it names, or a member of the /// organization the owner is, as its readers see them. fn writes(record: &RepoRecord, owner: &str, did: &str) -> bool { did == owner || record.is_collaborator(did) } /// Whether what `did` wrote is vouched for: it may push, and is not named /// as an agent, which it is even if it is a collaborator too. A pull or /// batch is trusted only if each who wrote it is. pub(crate) fn vouched(record: &RepoRecord, owner: &str, did: &str) -> bool { writes(record, owner, did) && !record.is_agent(did) } /// Whether a pull `did` wrote runs as it moves, without the owner asking: /// it may push, or it is an agent the repository names, whose pulls run as /// a collaborator's do but as code no one vouches for. fn runs_itself(record: &RepoRecord, owner: &str, did: &str) -> bool { writes(record, owner, did) || record.is_agent(did) } /// The first 12 characters of a commit's ID. fn short(sha: &str) -> &str { sha.get(..12).unwrap_or(sha) } /// The paths `git status --porcelain -z --no-renames` says differ. fn changed(status: &[u8]) -> Vec { let entries = status.split(|b| *b == 0).filter(|e| e.len() > 3); entries .map(|e| String::from_utf8_lossy(&e[3..]).into_owned()) .collect() } /// `n` files, in words. fn plural(n: usize) -> String { match n { 1 => "1 file".into(), n => format!("{n} files"), } } /// The files that differ between the commits `from` and `to`, renames as a /// deletion and an addition, or `None` if `git` cannot say. async fn changes(git: &Git, from: &str, to: &str) -> Option> { let args = ["diff", "--name-only", "--no-renames", "-z", from, to]; let out = git.text(&args, None).await.ok()?; Some( out.split('\0') .filter(|f| !f.is_empty()) .map(str::to_string) .collect(), ) } /// The ref a step's `on` is matched against in `subject`'s run: the one a /// push moved, or a run was asked for at. A pull's run and a batch's are at /// none of the repository's yet, whatever `KNOTLESS_RUN_REF` says. fn ref_at(subject: &Subject) -> Result<&str, &'static str> { match (&subject.pull, &subject.batch) { (Some(_), _) => Err("a pull's run is at no ref of the repository's"), (None, Some(_)) => Err("a merge queue batch's run is at no ref until it lands"), (None, None) => Ok(&subject.ref_name), } } /// Whether a step of `check`'s run was left out for the ref it ran at: /// then a push, at another, would not run as it did, and may not take it. fn gated(check: &Check) -> bool { check.steps.iter().any(|s| s.status == status::SKIPPED) } /// `step`, as not run. fn skipped_step(step: &workflow::Step) -> check::Step { check::Step { name: step.name.clone(), status: status::SKIPPED.into(), exit_code: None, duration_ms: 0, } } /// The variables `subject`'s run of `wf`, asked for by name, was given and /// its steps see: none of the runner's own, nor any named as a secret. fn given<'a>(subject: &'a Subject, wf: &Workflow) -> impl Iterator { let env = subject.asked.get(&wf.name).into_iter().flatten(); env.filter(|(k, _)| RerunVar::valid_name(k) && !wf.secrets.contains(k)) } /// Whether `wf` runs on `subject`: a pull into its base, or a push to its /// ref. fn runs(wf: &Workflow, subject: &Subject) -> bool { let base = subject.pull.as_ref().map(|p| &p.base); match base.or(subject.batch.as_ref().map(|b| &b.base)) { Some(base) => wf.runs_on_pull(base), None => wf.runs_on_push(&subject.ref_name), } } /// Watches until SIGTERM or SIGINT, then stops ([`Runner::stop`]): one /// queue per repository reads what moved, and its runs go side by side. pub async fn run(opts: Options, ctx: Ctx) -> Result<()> { let (runner, watches) = Runner::start(opts, ctx).await?; watch(runner, watches).await } /// [`run`], with a runner started. pub async fn watch(runner: Arc, watches: Vec) -> Result<()> { // Before anything else, for a stop asked for meanwhile to be heard. let mut term = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?; telemetry::watching(&runner.opts.repos, &runner.reporter.identity.did); for watch in &watches { report(watch, runner.recover(watch).await); } for watch in watches { runner.open_door(watch); } if runner.opts.discover { tokio::spawn(discover(runner.clone())); } tokio::spawn(sweep(runner.clone())); tokio::spawn(tidy(runner.clone())); tokio::spawn(schedules(runner.clone())); if let Some(appview) = runner.appview.clone() { tokio::spawn(wakes(runner.clone(), appview)); } let followed = tokio::select! { () = follow(&runner) => Ok(()), _ = term.recv() => Ok(()), _ = tokio::signal::ctrl_c() => Ok(()), }; runner.stop().await; followed } /// 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 /// is run then, not left for whatever rings next. async fn queue(runner: Arc, watch: Watch, door: Arc) { loop { door.notified().await; if !watch.watched() { return; } let began = std::time::Instant::now(); let every = runner.opts.watchdog; let done = watched(&watch.label, &watch.stage, every, runner.reconcile(&watch)).await; telemetry::pass(&watch.label, began.elapsed(), done.is_ok()); if done.is_err() { let (door, after) = (door.clone(), runner.opts.sweep.min(RETRY_MAX)); tokio::spawn(async move { tokio::time::sleep(after).await; door.notify_one(); }); } report(&watch, done); } } /// Awaits `pass` over `label`, saying each `every` it goes on how long it /// has and at what `stage` it is: one hung for a quarter of an hour once, /// with nothing to say where. async fn watched( label: &str, stage: &Stage, every: Duration, pass: impl Future, ) -> T { let began = std::time::Instant::now(); let mut pass = std::pin::pin!(pass); loop { tokio::select! { done = &mut pass => return done, () = tokio::time::sleep(every) => { let at = *stage.lock().unwrap(); telemetry::long_pass(label, at, began.elapsed()); } } } } /// Rings the door of each private repository whose space the appview /// says was written to, as soon as it says so; every one's, when it may /// have missed some. One that cannot be asked is asked again a while after, /// each space asked itself meanwhile as often as ever. async fn wakes(runner: Arc, appview: Arc) { let mut cursor: Option = None; loop { let heard = appview.await_writes(cursor.as_deref()).await; runner.woken.store(heard.is_ok(), Ordering::Relaxed); match heard { Ok(heard) => { let private = runner.private.lock().unwrap().clone(); let written = |space: &String| heard.missed || heard.writes.iter().any(|w| w.space == *space); for (_, door) in private.iter().filter(|(space, _)| written(space)) { door.notify_one(); } cursor = Some(heard.cursor); } // Nothing follows the wait but the loop: a line after it would // count only for a test that outlasts it. Err(_) => tokio::time::sleep(live::RETRY).await, } } } /// Removes, each `--sweep`, the containers of runs this runner has /// none of: one left running for seven hours once, its run long gone. async fn tidy(runner: Arc) { loop { tokio::time::sleep(runner.opts.sweep).await; runner.remove_left().await; } } /// Rings every door each `--sweep`, for what Jetstream never rang for: /// every instance of it dropped a whole account's events for a quarter /// of an hour once. Each read runs only what has not run, as any does. async fn sweep(runner: Arc) { loop { tokio::time::sleep(runner.opts.sweep).await; for door in runner.doors.lock().unwrap().values() { door.notify_one(); } } } /// Asks for the scheduled workflows due as each minute begins. async fn schedules(runner: Arc) { loop { let since = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH); let now = since.unwrap_or_default(); let next = now.as_secs() / 60 + 1; tokio::time::sleep(Duration::from_secs(next * 60) - now).await; noted("schedules", runner.on_schedule(next)); } } /// Watches each repository that names this runner, as the appview lists /// them and invites to private ones say, at the start and every /// `--sweep`: Jetstream brings public ones named meanwhile /// ([`Runner::found`]), but not always. async fn discover(runner: Arc) { loop { noted("discovery", runner.rediscover().await); noted("invites", runner.invited().await); tokio::time::sleep(runner.opts.sweep).await; } } /// Whether `repo`, a repository's record, is still the one keyed `key` /// and names `runner` among its runners, and is not archived: an archived /// one is left, and watched again once it is unarchived. fn names(repo: Option<&Stored>, key: &str, runner: &str) -> bool { repo.is_some_and(|r| { let named = r.value.runners.iter().any(|r| r.did == runner); r.value.key == key && named && r.value.archived_at.is_none() }) } /// A page of the appview's `dev.knotless.listRepos`, as discovery reads it. #[derive(Deserialize)] struct Listing { repos: Vec, cursor: Option, } #[derive(Deserialize)] struct Listed { did: String, name: String, } /// Rings `door` whenever a look at `watch`'s merge queue records, every /// `--notice`, finds them changed: a pull queued is landed within that, /// whatever Jetstream brings. pub async fn notice(runner: Arc, watch: Watch, door: Arc) { while watch.watched() { tokio::time::sleep(runner.opts.notice).await; let began = std::time::Instant::now(); let noticed = runner.noticed(&watch).await; let changed = noticed.as_ref().is_ok_and(|changed| *changed); telemetry::noticed(&watch.label, began.elapsed(), changed); if changed { door.notify_one(); } report(&watch, noticed.map(drop)); } } /// Rings `door` whenever the space `watch`'s private repository is in /// says something was written there since the last look, every `--poll`: /// a push, a rerun asked for, a pull opened. The first look rings too. /// While the appview passes on writes there ([`wakes`]), it looks only /// every [`live::WOKEN_POLL`], for one the appview missed; and it asks the /// space to tell the appview of them, every [`live::REGISTER`]. pub async fn poll(runner: Arc, watch: Watch, door: Arc) { // A private repository is never found, so it is left only as it // stops naming this runner or is deleted. let mut since = None; let (mut looked, mut registered) = (None, None); while watch.watched() { tokio::time::sleep(runner.opts.poll).await; let lapsed = |at: Option, every| at.is_none_or(|at| at.elapsed() >= every); if runner.appview.is_some() && lapsed(registered, live::REGISTER) { registered = Some(std::time::Instant::now()); report(&watch, runner.register(&watch).await); } let woken = runner.woken.load(Ordering::Relaxed); if woken && !lapsed(looked, live::WOKEN_POLL) { continue; } looked = Some(std::time::Instant::now()); let moved = runner.space_moved(&watch, since.as_deref()).await; if let Ok(Some(now)) = &moved { since = Some(now.clone()); door.notify_one(); } report(&watch, moved.map(drop)); } } /// The ref of `snap`'s default branch. fn main_ref(snap: &Snapshot) -> String { let default = snap.repo.as_ref(); let default = default.and_then(|r| r.value.default_branch.as_deref()); format!("refs/heads/{}", default.unwrap_or("main")) } /// Says what went wrong with `watch`'s work, if anything did. fn report(watch: &Watch, done: Result<()>) { noted(&watch.label, done); } /// Says what went wrong with the work of `label`, if anything did. fn noted(label: &str, done: Result<()>) { match done { Ok(()) => telemetry::succeeded(label), Err(err) => telemetry::failed(label, &err), } } /// Variables a step sees, by name. type Vars = Vec<(String, String)>; type Doors = HashMap<(String, String), Arc>; /// The most outputs a check keeps, as its lexicon allows. const MOST_OUTPUTS: usize = 32; /// How many outputs a run uploads at once. const UPLOADS: usize = 4; /// The largest output kept, as its lexicon allows. const OUTPUT_SIZE: usize = 1_000_000; /// The type of the image `path` is, by its extension, if it is one. fn image_type(path: &Path) -> Option<&'static str> { let ext = path.extension()?.to_string_lossy().to_ascii_lowercase(); let [png, jpeg, webp, gif] = knotless::records::OUTPUT_MIMES; match ext.as_str() { "png" => Some(png), "jpg" | "jpeg" => Some(jpeg), "webp" => Some(webp), "gif" => Some(gif), _ => None, } } /// The longest a pass that failed waits before it reads again. const RETRY_MAX: Duration = Duration::from_secs(60); /// Rings a repository's door for every ref event Jetstream names it in, or /// a repository an open pull into it comes from, and every pull or queue /// event about it; and every door on each attempt to connect, for what passed while /// there was no connection. fn follow(runner: &Arc) -> impl Future { let base = &runner.opts.jetstream; let doorbell = Doorbell { runner, base }; jetstream::follow(base.clone(), jetstream::IDLE, doorbell) } /// Jetstream as the runner hears it: a doorbell, never a source of what /// to run, so each connection starts at the stream's present. struct Doorbell<'a> { runner: &'a Arc, base: &'a str, } impl jetstream::Follower for Doorbell<'_> { fn connecting(&mut self) { for door in self.runner.doors.lock().unwrap().values() { door.notify_one(); } } fn connected(&mut self, up: bool) { telemetry::feed(self.base, up); } fn event(&mut self, event: jetstream::Event, _frame: &str) { let doors = self.runner.doors.lock().unwrap().clone(); ring(&doors, &self.runner.heads.lock().unwrap(), &event); self.runner.found(&event); } } fn ring(doors: &Doors, heads: &Heads, event: &jetstream::Event) { let Some(did) = &event.did else { return; }; for op in &event.ops { let rung = match op.collection.as_str() { // A request to run again is keyed by the repository, as a ref is. nsid::REF | nsid::RERUN => { let key = op.rkey.split('.').next().unwrap_or_default(); let repo = (did.clone(), key.to_string()); let mut rung: Vec<_> = heads.get(&repo).into_iter().flatten().cloned().collect(); rung.push(repo); rung } // The record is only a doorbell: what the pull or queue is // about is read back from the appview and its writers' repos. // A pull's state rings as the pull does, for one reopened or // retargeted to be run. nsid::PULL | nsid::STATE | nsid::QUEUE => { op.record.as_ref().and_then(pull_into).into_iter().collect() } // A repository's record rings each of its owner's watched, // for one that no longer names this runner, or is gone. nsid::REPO => doors.keys().filter(|(d, _)| d == did).cloned().collect(), _ => Vec::new(), }; for repo in rung { if let Some(door) = doors.get(&repo) { door.notify_one(); } } } } /// The repository a pull, or a state record, says it is of, as `(did, /// key)`. fn pull_into(record: &serde_json::Value) -> Option<(String, String)> { let repo: RepoRef = serde_json::from_value(record["repo"].clone()).ok()?; let (did, _, _) = forge::parse_uri(&repo.uri).ok()?; Some((did.to_string(), repo.key)) } /// What a run holds while its steps go: its turn on its ref, for a /// workflow whose runs finish, and its job. struct Held { on_ref: Option, job: turns::Turn, /// What its log says of its wait for the job. said: String, } /// A run's steps' container, noted in [`Runner::sessions`] from before it /// starts until it is gone. struct Tracked<'a> { sessions: &'a Mutex>, run: String, } impl<'a> Tracked<'a> { fn new(runner: &'a Runner, run: &str) -> Self { let sessions = &runner.sessions; sessions.lock().unwrap().insert(run.to_string()); Tracked { sessions, run: run.to_string(), } } } impl Drop for Tracked<'_> { fn drop(&mut self) { self.sessions.lock().unwrap().remove(&self.run); } } /// A run whose steps may start: a note that it goes, for a workflow whose /// runs finish, its job, and what says it is cancelled. /// Where a run's steps write, and tell of each as it ends. struct Steps<'a> { log: std::fs::File, progress: &'a tokio::sync::watch::Sender, /// The secrets the run is granted, by name, for each step to be given /// those it may see. granted: &'a Vars, /// Where each shared step's repository is checked out, as its steps /// see it. step_dirs: &'a [String], /// The ref a step's `on` is matched against, or why there is none. at: Result<&'a str, &'static str>, } struct Started<'a> { finishing: Option>, job: turns::Turn, cancelled: Cancelled, /// What its log says of its wait for the job. said: String, } /// `f`'s output, or `None` if `cancelled` first. async fn until_cancelled(cancelled: &mut Cancelled, f: impl Future) -> Option { tokio::select! { biased; _ = cancelled.wait() => None, t = f => Some(t), } } /// The caches `wf` keeps, as `(name, path)`: one for each path it names, /// for the repository `repo`, which every run of it reads and runs of /// pushes alone write: the default branch's, or `branch`'s own; or, /// `apart`, ones no run writes, and so always empty. fn caches(repo: &str, apart: bool, wf: &Workflow, branch: Option<&str>) -> Vec<(String, String)> { let scope = if apart { "\napart" } else { "" }; wf.cache .iter() .filter_map(|path| { let path = workflow::mount_path(path)?; let what = format!("{}\n{path}{scope}", wf.name); let name = caches::name(repo, &what); let name = match branch.filter(|_| !apart) { Some(branch) => caches::on_branch(&name, branch), None => name, }; Some((name, path)) }) .collect() } /// The branch whose own caches `subject`'s run of `wf` reads: its branch /// ([`Subject::branch`]) when that is not the default branch, and both a /// push to it and a pull into it run `wf`, so that its pushes leave what /// the pulls into it start from, as the default branch's do. `None` for /// the default branch's caches, which every other branch reads: one only /// pushed to, as a pull's head, is not worth a base of its own. fn own_branch<'a>(subject: &'a Subject, wf: &Workflow) -> Option<&'a str> { let branch = subject.branch(); let own = branch != subject.default && wf.runs_on_push(branch) && wf.runs_on_pull(branch); own.then_some(branch) } /// Whether `subject`'s run of `wf`, once it passes, makes what it wrote /// the base the runs after it start from: a push's, not a pull's or a /// batch's, to the default branch or to a branch whose caches are its own /// ([`own_branch`]). Not a push's to any other branch, which reads the /// default branch's caches, or an asked-for run's at a ref no push runs /// `wf` at. fn promotes(subject: &Subject, wf: &Workflow) -> bool { let push = subject.pull.is_none() && subject.batch.is_none(); push && (subject.main || own_branch(subject, wf).is_some()) } /// Whether a push's run of `wf` may be taken from a pull's that passed at /// the same tree: not if it asks for secrets, which the pull's run did not /// get, nor if its runs finish, as a deploy's, which a push runs for what /// it does. fn reusable(wf: &Workflow) -> bool { wf.secrets.is_empty() && wf.superseded != Superseded::Finish } /// Whether `subject`'s run of `wf`, once it passes, keeps the layers it /// wrote for a push to take: a trusted pull's that merges, or a trusted /// batch's, of a workflow a push may take. Never an untrusted pull's, or a /// batch's with one in it, whose writes would reach the base. fn keeps(subject: &Subject, wf: &Workflow) -> bool { let pull = subject.pull.as_ref().is_some_and(|p| p.trusted); let batch = subject.batch.as_ref().is_some_and(|b| b.trusted); (pull || batch) && subject.by_tree() && reusable(wf) } /// Whether `subject`'s run of `wf`, once it passes, makes what it wrote /// the base the runs after it start from, as a push's run's would, on a /// repository that is `pulls_only`, whose pushes run elsewhere and so never /// take kept layers: a trusted pull's into the default branch, or into a /// branch whose caches are its own ([`own_branch`]), of a workflow a push /// may take, as [`keeps`] says. Never an untrusted pull's, nor one into a /// branch that reads the default branch's caches, whose tips are not the /// default branch's to put in order. If so, the base branch's generation /// at the tip it merged with. fn warms(pulls_only: bool, subject: &Subject, wf: &Workflow) -> Option { let pull = subject.pull.as_ref()?; let ours = pull.base == subject.default || own_branch(subject, wf).is_some(); let tip = pull.base_generation.filter(|_| ours); tip.filter(|_| pulls_only && keeps(subject, wf)) } /// What a pull's log says of the caches its run made the base, as /// [`warms`] says: the paths `kept`, and those `moved`, over a base that /// has moved on since or is of a newer tip of the branch, which are not. fn warmed_lines(kept: &[String], moved: &[String]) -> String { let mut lines = String::new(); if !kept.is_empty() { let paths = kept.join(", "); lines += &format!( "==> caches: what this run wrote to {paths} is what the runs here start from now\n" ); } if !moved.is_empty() { let paths = moved.join(", "); lines += &format!( "==> caches: what this run wrote to {paths} is not kept: the base it was written over has moved on since, or is of a newer tip of the base\n" ); } lines } /// What a push's log says of the caches of the pull run it takes: of the /// paths `kept`, taken into its bases, and those `moved`, over a base /// that has moved on, which are not; or, for a workflow that is `cached`, /// that none were kept. fn taken_lines(kept: &[String], moved: &[String], cached: bool) -> String { let mut lines = String::new(); if !kept.is_empty() { let paths = kept.join(", "); lines += &format!( "==> caches: what that run wrote to {paths} is kept, as this run's would be\n" ); } if !moved.is_empty() { let paths = moved.join(", "); lines += &format!( "==> caches: what that run wrote to {paths} is not kept: the base it was written over has moved on since, and the next run here warms it\n" ); } if lines.is_empty() && cached { lines += "==> caches: none kept from that run\n"; } lines } /// Whether `subject`'s run of `wf` starts from nothing rather than what /// the default branch's runs left: an untrusted pull's, or a batch's with /// one in it, of a workflow that asks for secrets, which those runs had /// and might have left there. fn kept_apart(subject: &Subject, wf: &Workflow) -> bool { let pull = subject.pull.as_ref().is_some_and(|p| !p.trusted); let batch = subject.batch.as_ref().is_some_and(|b| !b.trusted); (pull || batch) && !wf.secrets.is_empty() } /// Checks out `commit` of the mirror `git` at `at`, its objects hard links /// to the mirror's rather than borrowed from it, so git works in a /// container that sees only the checkout. async fn checkout(git: &Git, at: &Path, commit: &str) -> Result<()> { let text = at.to_string_lossy(); let mirror = git.dir.to_string_lossy(); git.run(&["clone", "-q", "--no-checkout", &mirror, &text], None) .await?; let cloned = Git::new(at.join(".git")); let args = ["--work-tree", &text, "checkout", "-q", "--detach", commit]; cloned.run(&args, None).await?; Ok(()) } /// Whether `subject` is code its repository's owner vouches for: a push, /// or a pull or batch every writer of which the owner vouches for. fn vouched_for(subject: &Subject) -> bool { match (&subject.pull, &subject.batch) { (Some(pull), _) => pull.trusted, (None, Some(batch)) => batch.trusted, (None, None) => true, } } #[cfg(test)] mod tests { use super::*; // 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)] async fn a_run_waits_for_room_on_the_disk() { use std::sync::atomic::AtomicI64; let root = tempfile::tempdir().unwrap(); // The bytes free, as the test sets them, under 0 unreadable; and // how many looks find 5 first, as runs going hold the rest. let (free, short) = (Arc::new(AtomicI64::new(10)), Arc::new(AtomicI64::new(0))); let (told, shorted) = (free.clone(), short.clone()); let space: caches::Space = Box::new(move |_| { if shorted.fetch_sub(1, Ordering::SeqCst) > 0 { return Ok(5); } match told.load(Ordering::SeqCst) { n if n < 0 => Err(std::io::Error::other("no disk")), n => Ok(n as u64), } }); let caches = caches::Caches::open_with(root.path(), &[], &[], space).unwrap(); let every = Duration::from_secs(30); // Room enough, or no floor: nothing to say. assert_eq!(room(&caches, 10, every).await, ""); assert_eq!(room(&caches, 0, every).await, ""); // A disk it cannot read holds no run back. free.store(-1, Ordering::SeqCst); assert_eq!(room(&caches, 10, every).await, ""); free.store(20, Ordering::SeqCst); // Short, it sheds the caches, and room made so, says so alone. short.store(1, Ordering::SeqCst); let said = room(&caches, 10, every).await; assert!(said.contains("under the runner's floor"), "{said}"); assert!(said.contains("shed 0 caches' bases"), "{said}"); assert!(!said.contains("waited"), "{said}"); // Short after shedding, it waits for the runs going to free some: // a look before the shed, one after, and one each 30s. short.store(4, Ordering::SeqCst); let said = room(&caches, 10, every).await; assert!( said.ends_with("waited 90s for room on the disk\n"), "{said}" ); // Caches it cannot shed are logged, and waited past. std::fs::remove_dir_all(root.path().join("kept")).unwrap(); short.store(2, Ordering::SeqCst); let said = room(&caches, 10, every).await; assert!(said.contains("shed 0 caches' bases"), "{said}"); } #[tokio::test(start_paused = true)] async fn a_long_pass_is_watched() { let stage = Stage::default(); *stage.lock().unwrap() = "landing a batch"; let pass = tokio::time::sleep(Duration::from_millis(30)); watched("a/b", &stage, Duration::from_millis(5), pass).await; let quick = async { 7 }; assert_eq!( watched("a/b", &stage, Duration::from_secs(60), quick).await, 7 ); } #[test] fn archived_repositories_are_left() { let mut value = RepoRecord { key: "k".into(), runners: vec![knotless::records::Runner { did: "did:r".into(), }], ..Default::default() }; let stored = |value: &RepoRecord| Stored { rkey: "r".into(), cid: "c".into(), value: value.clone(), }; assert!(names(Some(&stored(&value)), "k", "did:r")); assert!(!names(Some(&stored(&value)), "other", "did:r")); assert!(!names(Some(&stored(&value)), "k", "did:x")); value.archived_at = Some("2026-10-29T00:00:00Z".into()); assert!(!names(Some(&stored(&value)), "k", "did:r")); assert!(!names(None, "k", "did:r")); } #[test] fn cancels_once_moved() { // A newer commit's cancel replaces one for a newer commit, but never // one for the pull having moved, which a run kept still heeds. let (sender, why) = tokio::sync::watch::channel(None); cancel(&sender, NEWER); cancel(&sender, NEWER); assert_eq!(*why.borrow(), Some(NEWER)); cancel(&sender, MOVED); cancel(&sender, NEWER); assert_eq!(*why.borrow(), Some(MOVED)); } #[test] fn what_changed() { let status = b" M shots/a.png\0?? shots/new b.png\0D gone.txt\0"; assert_eq!( changed(status), ["shots/a.png", "shots/new b.png", "gone.txt"] ); assert!(changed(b"").is_empty()); assert_eq!((plural(1), plural(3)), ("1 file".into(), "3 files".into())); } #[test] fn whose_caches() { let pull = |trusted| Pull { uri: "at://p".into(), base: "refs/heads/main".into(), base_commit: "b".into(), base_generation: Some(1), merged: Ok("m".into()), strangers: Vec::new(), trusted, later: false, others: Vec::new(), }; let subject = |pull| Subject { ref_name: "refs/heads/x".into(), commit: "h".into(), generation: 1, main: false, default: "refs/heads/main".into(), pull, tree: None, changed: None, batch: None, pushed: true, asked: Asks::new(), only: None, }; let plain = workflow::parse("ci.yml", b"steps: []\n").unwrap(); let secret = workflow::parse("ci.yml", b"secrets: [S]\nsteps: []\n").unwrap(); assert!(vouched_for(&subject(None))); assert!(vouched_for(&subject(Some(pull(true))))); assert!(!vouched_for(&subject(Some(pull(false))))); assert!(!kept_apart(&subject(None), &secret)); assert!(!kept_apart(&subject(Some(pull(true))), &secret)); assert!(!kept_apart(&subject(Some(pull(false))), &plain)); assert!(kept_apart(&subject(Some(pull(false))), &secret)); let batch = |trusted| Subject { batch: Some(queue::Batch::new( "refs/heads/main", "b", Vec::new(), trusted, )), ..subject(None) }; assert!(vouched_for(&batch(true))); assert!(!vouched_for(&batch(false))); assert!(!kept_apart(&batch(true), &secret)); assert!(!kept_apart(&batch(false), &plain)); assert!(kept_apart(&batch(false), &secret)); } #[test] fn whose_layers_are_kept() { let pull = |trusted, merged: Result>| Pull { uri: "at://p".into(), base: "refs/heads/main".into(), base_commit: "b".into(), base_generation: Some(1), merged, strangers: Vec::new(), trusted, later: false, others: Vec::new(), }; let subject = |pull, main| Subject { ref_name: "refs/heads/x".into(), commit: "h".into(), generation: 1, main, default: "refs/heads/main".into(), pull, tree: None, changed: None, batch: None, pushed: true, asked: Asks::new(), only: None, }; let parse = |text: &str| workflow::parse("ci.yml", text.as_bytes()).unwrap(); let plain = parse("cache: [target]\nsteps: []\n"); let secret = parse("secrets: [S]\ncache: [target]\nsteps: []\n"); let deploy = parse("superseded: finish\ncache: [target]\nsteps: []\n"); let merges = || Ok("m".to_string()); // A trusted pull's that merges, of a workflow a push may take. assert!(keeps(&subject(Some(pull(true, merges())), false), &plain)); // Never an untrusted pull's, whatever it runs: its writes would // reach the base. assert!(!keeps(&subject(Some(pull(false, merges())), false), &plain)); assert!(!keeps( &subject(Some(pull(false, merges())), false), &secret )); // Nor one that conflicts, which no push is at, nor a push's own. let conflicts = pull(true, Err(vec!["f".into()])); assert!(!keeps(&subject(Some(conflicts), false), &plain)); assert!(!keeps(&subject(None, true), &plain)); // Nor of a workflow no push takes from a pull. assert!(!keeps(&subject(Some(pull(true, merges())), false), &secret)); assert!(!keeps(&subject(Some(pull(true, merges())), false), &deploy)); // A trusted batch's is kept for the push that lands it; never one // with an untrusted pull in it. let batch = |trusted| Subject { batch: Some(queue::Batch::new("refs/heads/x", "b", Vec::new(), trusted)), ..subject(None, true) }; assert!(keeps(&batch(true), &plain)); assert!(!keeps(&batch(false), &plain)); // A step's `on` is matched against a push's ref; a pull's run and // a batch's are at none, whatever their ref names. assert_eq!(ref_at(&subject(None, false)), Ok("refs/heads/x")); let pulled = subject(Some(pull(true, merges())), false); assert!(ref_at(&pulled).unwrap_err().contains("a pull's run")); assert!(ref_at(&batch(true)).unwrap_err().contains("batch")); // Where pushes run elsewhere, a trusted pull's that passes makes // the base, in place of a push's; never an untrusted pull's, one // of a workflow no push takes from a pull, nor anywhere else. let trusted = subject(Some(pull(true, merges())), false); assert_eq!(warms(true, &trusted, &plain), Some(1)); assert_eq!(warms(false, &trusted, &plain), None); assert_eq!(warms(true, &trusted, &secret), None); assert_eq!(warms(true, &trusted, &deploy), None); let untrusted = subject(Some(pull(false, merges())), false); assert_eq!(warms(true, &untrusted, &plain), None); assert_eq!(warms(true, &batch(true), &plain), None); // A pull into another branch whose pushes run the workflow warms // that branch's own caches, its tips put in order as the default // branch's are; one into a branch that reads the default branch's // caches does not, nor a pre-flight, with no tip to order by. let side = Pull { base: "refs/heads/feature/x".into(), ..pull(true, merges()) }; let into_side = subject(Some(side.clone()), false); let both = parse( "on: {push: {refs: [refs/heads/*]}, pull: {base: [refs/heads/*]}}\ncache: [target]\nsteps: []\n", ); assert_eq!(warms(true, &into_side, &both), Some(1)); assert_eq!(warms(true, &into_side, &plain), None); let main_only = parse( "on: {push: {refs: [refs/heads/main]}, pull: {base: [refs/heads/*]}}\ncache: [target]\nsteps: []\n", ); assert_eq!(warms(true, &into_side, &main_only), None); let preflight = Pull { base_generation: None, ..pull(true, merges()) }; assert_eq!(warms(true, &subject(Some(preflight), false), &plain), None); } #[test] fn whose_branch_the_caches_are() { let parse = |text: &str| workflow::parse("ci.yml", text.as_bytes()).unwrap(); let feature = parse( "on: {push: {refs: [refs/heads/main, refs/heads/feature/*]}, pull: {base: [refs/heads/*]}}\ncache: [target]\nsteps: []\n", ); let main_only = parse( "on: {push: {refs: [refs/heads/main]}, pull: {base: [refs/heads/*]}}\ncache: [target]\nsteps: []\n", ); let push = |ref_name: &str| Subject { ref_name: ref_name.into(), commit: "h".into(), generation: 1, main: ref_name == "refs/heads/main", default: "refs/heads/main".into(), pull: None, tree: None, changed: None, batch: None, pushed: true, asked: Asks::new(), only: None, }; let pull = |base: &str| Subject { pull: Some(Pull { uri: "at://p".into(), base: base.into(), base_commit: "b".into(), base_generation: Some(1), merged: Ok("m".into()), strangers: Vec::new(), trusted: true, later: false, others: Vec::new(), }), ..push("refs/heads/topic") }; let batch = |base: &str| Subject { batch: Some(queue::Batch::new(base, "b", Vec::new(), true)), main: true, ..push(base) }; let f = "refs/heads/feature/x"; // A push to a branch its pushes run on, a pull into it and a // batch on it read that branch's own caches; the default branch's // runs, and those of a branch only asked for, read the default's. assert_eq!(own_branch(&push(f), &feature), Some(f)); assert_eq!(own_branch(&pull(f), &feature), Some(f)); assert_eq!(own_branch(&batch(f), &feature), Some(f)); assert_eq!(own_branch(&push("refs/heads/main"), &feature), None); assert_eq!(own_branch(&pull("refs/heads/main"), &feature), None); assert_eq!(own_branch(&push(f), &main_only), None); assert_eq!(own_branch(&pull(f), &main_only), None); // Nor does a branch only pushed to, which no pull goes into. let pushes = parse("cache: [target]\nsteps: []\n"); assert_eq!(own_branch(&push(f), &pushes), None); assert!(!promotes(&push(f), &pushes)); // Only a push writes them: to the default branch, or one whose // caches are its own. assert!(promotes(&push(f), &feature)); assert!(promotes(&push("refs/heads/main"), &main_only)); assert!(!promotes(&push(f), &main_only)); assert!(!promotes(&pull(f), &feature)); assert!(!promotes(&batch(f), &feature)); assert!(!promotes(&batch("refs/heads/main"), &feature)); // A branch's caches are the default's named for the branch too; a // cache read apart is never a branch's. let main = caches("did:a/k", false, &feature, None); let on_f = caches("did:a/k", false, &feature, Some(f)); assert_eq!(on_f[0].0, caches::on_branch(&main[0].0, f)); assert_eq!(on_f[0].1, main[0].1); let apart = caches("did:a/k", true, &feature, Some(f)); assert_eq!(apart, caches("did:a/k", true, &feature, None)); assert_ne!(apart[0].0, main[0].0); } #[test] fn says_what_was_taken() { let paths = |all: &[&str]| all.iter().map(|p| p.to_string()).collect::>(); let (target, cargo) = (paths(&["/work/target"]), paths(&["/a", "/b"])); assert_eq!( warmed_lines(&target, &cargo), "==> caches: what this run wrote to /work/target is what the runs here start from now\n\ ==> caches: what this run wrote to /a, /b is not kept: the base it was written over has moved on since, or is of a newer tip of the base\n" ); assert!(warmed_lines(&[], &[]).is_empty()); assert_eq!( taken_lines(&target, &[], true), "==> caches: what that run wrote to /work/target is kept, as this run's would be\n" ); let both = taken_lines(&target, &cargo, true); assert!(both.starts_with("==> caches: what that run wrote to /work/target is kept")); assert!(both.ends_with("==> caches: what that run wrote to /a, /b is not kept: the base it was written over has moved on since, and the next run here warms it\n"), "{both}"); assert_eq!( taken_lines(&[], &[], true), "==> caches: none kept from that run\n" ); assert_eq!(taken_lines(&[], &[], false), ""); } #[test] fn caches_apart() { let wf = workflow::parse("ci.yml", b"cache: [target, /usr/local/cargo/]\nsteps: []\n").unwrap(); let all = caches("did:plc:a/k", false, &wf, None); let paths: Vec<&str> = all.iter().map(|(_, p)| p.as_str()).collect(); assert_eq!(paths, ["/work/target", "/usr/local/cargo"]); assert!(all[0].0.starts_with("knotless-cache-")); assert_ne!(all[0].0, all[1].0); // The same for every run, and apart for each repository. assert_eq!(caches("did:plc:a/k", false, &wf, None), all); assert_ne!(caches("did:plc:b/k", false, &wf, None)[0], all[0]); // Kept apart, others, whose paths are the same. let apart = caches("did:plc:a/k", true, &wf, None); assert_ne!(apart[0].0, all[0].0); assert_eq!(apart[0].1, all[0].1); } fn args(list: &str) -> Result { let mut all = vec!["knotless-run".to_string()]; all.extend(list.split_whitespace().map(str::to_string)); parse_args(&all) } #[test] fn arguments() { let opts = args("alice.test/proj knotless://did:plc:b/other").unwrap(); assert_eq!(opts.repos, ["alice.test/proj", "did:plc:b/other"]); assert_eq!( opts.executor, Executor::Docker { image: "debian:stable-slim".into(), limits: executor::Limits::default(), } ); let opts = args( "--jetstream ws://h --executor shell --secrets s.json --state st.json --work w --step-timeout 5 a/b", ) .unwrap(); assert_eq!(opts.jetstream, "ws://h"); assert_eq!(opts.executor, Executor::Shell); assert_eq!(opts.secrets, Some("s.json".into())); assert_eq!(opts.state, Some("st.json".into())); assert_eq!(opts.work, Some("w".into())); assert_eq!(opts.step_timeout, Duration::from_secs(5)); assert_eq!(opts.jobs, 4); assert_eq!(opts.cache_max, 16 << 30); assert_eq!(opts.caches, None); assert_eq!(opts.sweep, Duration::from_secs(600)); assert_eq!(opts.stop_wait, Duration::from_secs(900)); let opts = args("--sweep 5 --stop-wait 0 a/b").unwrap(); assert_eq!(opts.sweep, Duration::from_secs(5)); assert_eq!(opts.stop_wait, Duration::ZERO); assert_eq!( args("--sweep 0 a/b").unwrap_err(), "--sweep takes a number of seconds, at least 1" ); assert_eq!( args("--stop-wait x a/b").unwrap_err(), "--stop-wait takes seconds" ); // Checks are pruned unless told not to, keeping main's for 30 days. let day = Duration::from_secs(DAY); assert_eq!(opts.keep_checks, Some(30 * day)); let opts = args("--keep-checks 7 a/b").unwrap(); assert_eq!(opts.keep_checks, Some(7 * day)); assert_eq!(args("--keep-checks 0 a/b").unwrap().keep_checks, None); assert_eq!( args("--keep-checks x a/b").unwrap_err(), "--keep-checks takes days" ); assert_eq!( args("--paused p.json a/b").unwrap().paused, Some("p.json".into()) ); // An owner listed, or every one under `*`, is paused, but never // the runner's own account. let list = r#"{"did:plc:o": "mining", "*": "closed"}"#; assert_eq!( paused_why(list, "did:plc:o", "did:plc:r") .unwrap() .as_deref(), Some("mining") ); assert_eq!( paused_why(list, "did:plc:x", "did:plc:r") .unwrap() .as_deref(), Some("closed") ); assert_eq!(paused_why(list, "did:plc:r", "did:plc:r").unwrap(), None); assert_eq!( paused_why(r#"{"did:plc:o": "m"}"#, "did:plc:x", "r").unwrap(), None ); assert_eq!(paused_why(" \n", "did:plc:o", "did:plc:r").unwrap(), None); let err = paused_why("[]", "did:plc:o", "did:plc:r").unwrap_err(); assert_eq!( err.to_string(), "--paused takes a JSON object of DIDs to why" ); // Each repository is pruned when first read, then each hour. let now = std::time::Instant::now(); let hour = Duration::from_secs(3600); assert!(prune_due(None, now, hour)); assert!(!prune_due(Some(now), now, hour)); assert!(prune_due(Some(now), now + hour, hour)); assert_eq!(args("--caches c a/b").unwrap().caches, Some("c".into())); assert_eq!(args("--cache-max 2 a/b").unwrap().cache_max, 2 << 30); assert_eq!( args("--cache-max x a/b").unwrap_err(), "--cache-max takes GiB" ); assert_eq!(Options::new(Vec::new()).disk_floor, 20 << 30); assert_eq!(args("--disk-floor 0 a/b").unwrap().disk_floor, 0); assert_eq!( args("--disk-floor x a/b").unwrap_err(), "--disk-floor takes GiB" ); assert_eq!(args("--jobs 2 a/b").unwrap().jobs, 2); // Pulls take half the jobs, or as many as asked, up to all of them. assert_eq!(opts.pull_share(), 2); assert_eq!(args("--jobs 1 a/b").unwrap().pull_share(), 1); assert_eq!(args("--pull-jobs 3 a/b").unwrap().pull_share(), 3); assert_eq!(args("--pull-jobs 9 a/b").unwrap().pull_share(), 4); // Made in code, unset, it is half too. assert_eq!(Options::new(Vec::new()).pull_share(), 2); // An owner takes all the jobs, or as many as asked, up to all. assert_eq!(opts.owner_share(), 4); assert_eq!(args("--owner-jobs 2 a/b").unwrap().owner_share(), 2); assert_eq!(args("--owner-jobs 9 a/b").unwrap().owner_share(), 4); assert_eq!( args("--owner-jobs 0 a/b").unwrap_err(), "--owner-jobs takes a number of runs, at least 1" ); assert_eq!(opts.allowance, None); assert_eq!(args("--allowance 2000 a/b").unwrap().allowance, Some(2000)); let private = args("--private-allowance 500 a/b").unwrap(); assert_eq!(private.private_allowance, Some(500)); assert_eq!( args("--allowance 0 a/b").unwrap_err(), "--allowance takes a number of minutes, at least 1" ); for bad in ["--pull-jobs 0 a/b", "--pull-jobs x a/b"] { assert_eq!( args(bad).unwrap_err(), "--pull-jobs takes a number of runs, at least 1" ); } for bad in ["--jobs 0 a/b", "--jobs x a/b"] { assert_eq!( args(bad).unwrap_err(), "--jobs takes a number of runs, at least 1" ); } let opts = args("--queue --batch 3 --depth 2 a/b").unwrap(); assert_eq!((opts.queue, opts.batch, opts.depth), (true, 3, 2)); let opts = args("a/b").unwrap(); assert_eq!((opts.queue, opts.batch, opts.depth), (false, 8, 2)); assert_eq!(args("--batch 64 a/b").unwrap().batch, 64); assert_eq!( args("--batch 65 a/b").unwrap_err(), "--batch is 64 pulls at most" ); assert_eq!(args("--depth 1 a/b").unwrap().depth, 1); assert_eq!( args("--depth 0 a/b").unwrap_err(), "--depth takes a number of batches, at least 1" ); assert_eq!( args("--depth 3 a/b").unwrap_err(), "--depth is 1 or 2: one batch at most runs ahead" ); assert_eq!(opts.notice, Duration::from_secs(30)); assert_eq!(args("--notice 5 a/b").unwrap().notice.as_secs(), 5); let wait = args("--review-wait 60 a/b").unwrap().review_wait; assert_eq!(wait.as_secs(), 60); assert_eq!(args("a/b").unwrap().poll.as_secs(), 10); assert_eq!(args("--poll 2 a/b").unwrap().poll.as_secs(), 2); assert_eq!( args("--notice 0 a/b").unwrap_err(), "--notice takes a number of seconds, at least 1" ); let opts = args("--image rust:1 a/b").unwrap(); assert_eq!( opts.executor, Executor::Docker { image: "rust:1".into(), limits: executor::Limits::default(), } ); // A run's container held to limits, and its steps together to a // time. let opts = args( "--cpus 1.5 --memory 4 --pids 512 --network runs --runtime runsc --trusted-runtime runc --trust did:plc:c --resolv /r --run-timeout 90 a/b", ); let opts = opts.unwrap(); assert_eq!( opts.executor, Executor::Docker { image: "debian:stable-slim".into(), limits: executor::Limits { cpus: Some("1.5".into()), memory: Some(4 << 30), pids: Some(512), network: Some("runs".into()), runtime: Some("runsc".into()), trusted_runtime: Some("runc".into()), resolv: Some("/r".into()), }, } ); assert_eq!(opts.trust, ["did:plc:c"]); assert_eq!(opts.run_timeout, Some(Duration::from_secs(90))); assert_eq!(args("a/b").unwrap().run_timeout, None); for bad in ["0", "-1", "x", "inf", "NaN"] { assert_eq!( args(&format!("--cpus {bad} a/b")).unwrap_err(), format!("--cpus takes a number of CPUs, more than 0, not {bad}") ); } for (flag, what) in [ ("--memory", "GiB"), ("--pids", "processes"), ("--run-timeout", "seconds"), ] { assert_eq!( args(&format!("{flag} 0 a/b")).unwrap_err(), format!("{flag} takes a number of {what}, at least 1") ); } // The shell holds nothing to them. assert_eq!( args("--executor shell --pids 9 a/b").unwrap_err(), "--cpus, --memory, --pids, --network, --runtime, --trusted-runtime and --resolv hold a run's container: they take --executor docker" ); assert_eq!( args("--executor shell --run-timeout 9 a/b") .unwrap() .run_timeout, Some(Duration::from_secs(9)) ); assert!(args("").unwrap_err().starts_with("name a repository")); // Or none, finding those that name it. let found = args("--discover").unwrap(); assert!(found.discover && found.repos.is_empty()); assert!( args("--nope a/b") .unwrap_err() .starts_with("unexpected --nope") ); assert_eq!(args("a/b --state").unwrap_err(), "--state takes a value"); assert_eq!( args("--step-timeout x a/b").unwrap_err(), "--step-timeout takes seconds" ); assert_eq!( args("--executor vm a/b").unwrap_err(), "--executor is docker or shell, not vm" ); assert!(args("--pulls a/b").unwrap().pulls); // One named --pulls-only is watched, for its pulls. let only = args("--pulls --pulls-only knotless://c/d --unmetered did:plc:c a/b").unwrap(); assert_eq!( (only.repos, only.pulls_only, only.unmetered), ( vec!["c/d".to_string(), "a/b".to_string()], vec!["c/d".to_string()], vec!["did:plc:c".to_string()] ) ); assert_eq!( args("--pulls-only c/d").unwrap_err(), "--pulls-only runs a repository's pulls: it takes --pulls" ); for shell in ["--pulls", "--queue"] { let err = args(&format!("{shell} --executor shell a/b")).unwrap_err(); assert!(err.ends_with("they take --executor docker"), "{err}"); } } #[test] fn rings() { use futures::FutureExt; let door = Arc::new(Notify::new()); let a = ("did:plc:a".to_string(), "k1".to_string()); let doors: Doors = [(a.clone(), door.clone())].into(); // An open pull into it comes from did:plc:f's repository keyed kf. let heads: Heads = [(("did:plc:f".to_string(), "kf".to_string()), [a].into())].into(); let rung = || door.notified().now_or_never().is_some(); let op = |collection: &str, rkey: &str, record: Option| jetstream::Op { action: "update".into(), collection: collection.into(), rkey: rkey.into(), record, }; let event = |did: Option<&str>, ops| jetstream::Event { seq: None, did: did.map(str::to_string), ops, ..Default::default() }; let pull = |uri: &str, key: &str| serde_json::json!({"repo": {"uri": uri, "key": key}}); ring(&doors, &heads, &event(None, vec![])); for quiet in [ event(Some("did:plc:a"), vec![op(nsid::PACK, "k1.x", None)]), event(Some("did:plc:a"), vec![op(nsid::REF, "k2.x", None)]), event(Some("did:plc:b"), vec![op(nsid::REF, "k1.x", None)]), event(Some("did:plc:f"), vec![op(nsid::REF, "k2.x", None)]), event(Some("did:plc:b"), vec![op(nsid::PULL, "p", None)]), event( Some("did:plc:b"), vec![op(nsid::PULL, "p", Some(serde_json::json!({})))], ), event( Some("did:plc:b"), vec![op(nsid::PULL, "p", Some(pull("nope", "k1")))], ), event( Some("did:plc:b"), vec![op( nsid::PULL, "p", Some(pull("at://did:plc:a/dev.knotless.repo/r", "k2")), )], ), ] { ring(&doors, &heads, &quiet); } assert!(!rung()); ring( &doors, &heads, &event( Some("did:plc:a"), vec![op(nsid::REF, "k1.x", None), op(nsid::REF, "k1.y", None)], ), ); assert!(rung()); assert!(!rung()); ring( &doors, &heads, &event(Some("did:plc:f"), vec![op(nsid::REF, "kf.x", None)]), ); assert!(rung()); let into = pull("at://did:plc:a/dev.knotless.repo/r", "k1"); ring( &doors, &heads, &event( Some("did:plc:b"), vec![op(nsid::PULL, "p", Some(into.clone()))], ), ); assert!(rung()); // A pull reopened, or retargeted, rings as the pull does. ring( &doors, &heads, &event(Some("did:plc:a"), vec![op(nsid::STATE, "s", Some(into))]), ); assert!(rung()); // A pull queued, taken out or turned away, as its record is about. let queued = pull("at://did:plc:a/dev.knotless.repo/r", "k1"); let queue = vec![op(nsid::QUEUE, "q", Some(queued))]; ring(&doors, &heads, &event(Some("did:plc:b"), queue)); assert!(rung()); // A repository record rings its owner's, and no one else's. let repo = |did| event(Some(did), vec![op(nsid::REPO, "r", None)]); ring(&doors, &heads, &repo("did:plc:b")); assert!(!rung()); ring(&doors, &heads, &repo("did:plc:a")); assert!(rung()); } #[test] fn usage_over_30_days() { let mut used = Used::new(); spend(&mut used, 100, 5); spend(&mut used, 100, 7); spend(&mut used, 120, 1); assert_eq!(spent(&used, 120), 13); // Day 100 is the 30th day back from 129, and gone from 130. assert_eq!(spent(&used, 129), 13); assert_eq!(spent(&used, 130), 1); spend(&mut used, 130, 2); assert_eq!(used.keys().copied().collect::>(), [120, 130]); assert_eq!(spent(&Used::new(), 3), 0); // Past an allowance of minutes, one job at a time. assert_eq!(past(59_999, Some(1)), None); assert_eq!(past(60_000, Some(1)), Some(1)); assert_eq!(past(u64::MAX, None), None); assert_eq!((most(Some(1), 4), most(None, 4)), (1, 4)); // A run's log says why it waited, when its owner is past one. assert_eq!( waited(Some(2000), Duration::from_secs(75)), "==> its owner's runs are past their allowance of 2000 minutes of steps in 30 days, so they go one at a time: this one waited 75s for a job\n" ); assert_eq!(waited(None, Duration::from_secs(75)), ""); // Where an owner stands, and when its record is written again. assert_eq!(standing(79_999 * 60, Some(100)), 0); assert_eq!(standing(80 * 60_000, Some(100)), 1); assert_eq!(standing(100 * 60_000, Some(100)), 2); assert_eq!(standing(u64::MAX / 8, None), 0); let now = std::time::Instant::now(); assert!(due(None, 0)); assert!(!due(Some((now, 1)), 1)); assert!(due(Some((now, 0)), 1)); let long_ago = now.checked_sub(MINUTES_EVERY).unwrap(); assert!(due(Some((long_ago, 1)), 1)); // A record says what frees up when, of the days with a minute. let mut used = Used::new(); spend(&mut used, 100, 90_000); spend(&mut used, 101, 59_999); let record = minutes_record("did:plc:o", &used, Some(2000), 101); assert_eq!((record.minutes, record.allowance), (2, Some(2000))); assert_eq!(record.owner.as_deref(), Some("did:plc:o")); let freeing = [Freeing { at: "1970-05-11T00:00:00Z".into(), minutes: 1, }]; assert_eq!(record.freeing, freeing); } }