chadtmiller.com/knotless
git that lives in your Atmosphere account
7
//! 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 <wss://host>] [--executor docker|shell] [--image <image>] [--secrets <file>] [--secrets-key <file>] [--state <file>] [--work <dir>] [--step-timeout <secs>] [--run-timeout <secs>] [--cpus <n>] [--memory <GiB>] [--pids <n>] [--network <name>] [--runtime <name>] [--trusted-runtime <name>] [--trust <did>] [--resolv <file>] [--jobs <n>] [--pull-jobs <n>] [--owner-jobs <n>] [--allowance <minutes>] [--private-allowance <minutes>] [--cache-max <GiB>] [--disk-floor <GiB>] [--caches <dir>] [--paused <file>] [--keep-checks <days>] [--sweep <secs>] [--stop-wait <secs>] [--poll <secs>] [--pulls] [--pulls-only <handle>/<name>] [--unmetered <did>] [--queue [--batch <n>] [--depth 1|2] [--notice <secs>] [--review-wait <secs>]] [--discover] [<handle>/<name>...]";
#[derive(Debug, Clone, PartialEq)]
pub struct Options {
pub jetstream: String,
/// The repositories watched, as `<handle or did>/<name>`.
pub repos: Vec<String>,
/// 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<String>,
/// 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: `{"<handle>/<name>": {"NAME":
/// "value"}}`, or `{"value": "...", "refs": ["refs/heads/main"]}` for
/// one granted only to pushes to those refs.
pub secrets: Option<PathBuf>,
/// 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<PathBuf>,
/// The generation each ref was last run at. Defaults to one in the cache
/// directory.
pub state: Option<PathBuf>,
/// Where checkouts go while their steps run. Defaults to the system's
/// temporary directory.
pub work: Option<PathBuf>,
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<Duration>,
/// 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<u64>,
/// The same for private repositories' runs, counted apart.
pub private_allowance: Option<u64>,
/// 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<String>,
/// 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<String>,
/// 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<PathBuf>,
/// 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<PathBuf>,
/// 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<Duration>,
/// 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<String>) -> 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>) -> (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<Duration>) -> 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<Option<String>> {
if list.trim().is_empty() || owner == runner {
return Ok(None);
}
let list: HashMap<String, String> =
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 <days>` keeps the default branch's checks:
/// 0 keeps every check.
fn kept_for(days: u64) -> Option<Duration> {
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<std::time::Instant>, 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<usize, String> {
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<String, String> {
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<Duration, String> {
let secs = value.parse().map_err(|_| format!("{flag} takes seconds"))?;
Ok(Duration::from_secs(secs))
}
pub fn parse_args(args: &[String]) -> Result<Options, String> {
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 `<did>/<key>`; 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<String, BTreeMap<String, u64>>,
#[serde(default)]
pulls: BTreeMap<String, BTreeMap<String, String>>,
#[serde(default)]
reruns: BTreeMap<String, String>,
#[serde(default)]
asked: BTreeMap<String, BTreeSet<String>>,
/// 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<String, BTreeMap<String, BTreeSet<String>>>,
/// The trusted pulls' runs that passed, oldest first, by repository.
#[serde(default)]
passed: BTreeMap<String, Vec<Passed>>,
#[serde(default)]
dropped: BTreeMap<String, Dropped>,
/// The workflows asked for by name and not run yet, by repository and
/// ref.
#[serde(default)]
wanted: BTreeMap<String, BTreeMap<String, Asks>>,
/// 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<String, Used>,
/// The same of each owner's private repositories.
#[serde(default)]
used_private: BTreeMap<String, Used>,
/// The repositories found naming this runner, by label, to watch
/// again at a start, before the appview is asked.
#[serde(default)]
discovered: BTreeSet<String>,
/// Where the invites read so far end, to read on from.
#[serde(default, skip_serializing_if = "Option::is_none")]
invited: Option<String>,
/// Each repository's scheduled workflows, as its default branch had
/// them when it was last read.
#[serde(default)]
schedules: BTreeMap<String, Scheduled>,
/// 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<String, BTreeMap<String, u64>>,
}
/// 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<String, Vec<String>>,
}
/// 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<String, Used> {
if private {
&mut self.used_private
} else {
&mut self.used
}
}
}
/// An owner's milliseconds of steps, by day since the epoch.
type Used = BTreeMap<u64, u64>;
/// 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<u64>) -> 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<u64>, 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<u64>) -> Option<u64> {
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<u64>, 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<u64>, 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<caches::Caches>, 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<String, Vec<(String, String)>>;
/// 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<String, BTreeMap<String, BTreeSet<String>>>;
/// 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<knotless::records::PackRecord>,
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<Pull>,
/// 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<String>,
/// 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<Vec<String>>,
/// The merge queue's batch it is the tip of.
batch: Option<queue::Batch>,
/// 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<BTreeSet<String>>,
}
#[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<u64>,
/// The commit the head merged into the base makes, or the paths that
/// conflict.
merged: Result<String, Vec<String>>,
/// 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<String>,
/// 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<String>,
}
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<AtomicBool>,
/// 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<Mutex<&'static str>>;
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<State>,
heads: Mutex<Heads>,
/// One permit each run spawned holds, waiting or going, so
/// [`Runner::idle`] can wait for them all.
runs: Arc<Semaphore>,
/// The runs going, by repository, ref or pull, and workflow.
going: Mutex<HashMap<String, Going>>,
/// The jobs, `--jobs` of them.
turns: Arc<Turns>,
/// The caches' bases, and the runs' layers over them.
caches: Arc<caches::Caches>,
/// 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<Turns>,
/// 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<HashMap<String, (std::time::Instant, u8)>>,
/// When each watched repository's checks were last pruned, by
/// [`Watch::id`].
pruned: Mutex<HashMap<String, std::time::Instant>>,
/// Each merge queue's batches going, and what a failing one showed,
/// by `<did>/<key>\n<branch>`.
lines: Mutex<HashMap<String, queue::Line>>,
/// 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<Doors>,
/// 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<HashMap<(String, String), Watch>>,
/// 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
/// `<did>/<key>`, and its queue records as last looked at.
noticing: Mutex<HashMap<String, queue::Noticing>>,
/// Set once the runner is stopping: a run that has not started its
/// steps by then is dropped.
stopping: tokio::sync::watch::Sender<bool>,
/// 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<HashSet<String>>,
/// Where each run going of a workflow whose runs finish is at, as
/// [`Runner::dropping`] names it.
finishing: tokio::sync::watch::Sender<Vec<String>>,
/// The appview runs' output goes to as it is written, and writes in
/// private repositories' spaces come from ([`live`]).
appview: Option<Arc<live::Appview>>,
/// 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<HashMap<String, Arc<Notify>>>,
}
/// 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<Vec<String>>,
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<String>,
/// 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<Option<&'static str>>;
/// 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<Option<&'static str>>,
/// 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<Box<dyn Future<Output = turns::Turn> + 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<OnRef>,
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<Runner>, Vec<Watch>)> {
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<Watch> {
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<Runner>, Vec<Watch>)> {
// 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<String> = 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<Self>, 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<String> = 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<String> = 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<String> = open.map(|i| i.uri).collect();
let refs: HashSet<String> = 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<Self>,
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<String, String> = 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<Self>, 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<String> = 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<String> = 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<bool> {
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<String, String>) {
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::<serde_json::Value>(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::<RerunRecord>(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<Option<String>> {
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<String> {
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<String> = 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<Vec<(RefRecord, bool)>> {
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<String, u64> = (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<BTreeSet<String>> {
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<Self>,
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<String> {
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<String> {
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<String> {
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<checks::Check> {
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<String> = 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<Passed> {
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<bool> {
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<bool> {
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<OnRef>,
cancelled: &mut Cancelled,
) -> Option<Held> {
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<u64>) {
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<Option<String>> {
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<u64> {
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<Option<MinutesRecord>> {
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<String> {
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<Self>, watch: Watch) -> Arc<Notify> {
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<Self>, 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<Self>) -> 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<Option<String>> = 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<Self>) -> 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<Self>, 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<String> {
{
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::<String>())?;
}
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<String> {
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<String> = 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<Vec<CheckOutput>> {
let mut paths: Vec<PathBuf> = std::fs::read_dir(dir)?
.map(|e| e.map(|e| e.path()))
.collect::<std::io::Result<_>>()?;
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<String>, Vec<String>)> {
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<String, BTreeMap<String, Secret>> = 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<Vec<(&'a String, String)>> {
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<BTreeMap<String, serde_json::Value>> = 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<String> },
}
impl Secret {
/// Its value, if a push to `ref_name` is granted it.
fn to(self, ref_name: &str) -> Option<String> {
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<String> {
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<Vec<String>> {
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<Item = &'a (String, String)> {
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<Runner>, watches: Vec<Watch>) -> 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<Runner>, watch: Watch, door: Arc<Notify>) {
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<T>(
label: &str,
stage: &Stage,
every: Duration,
pass: impl Future<Output = T>,
) -> 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<Runner>, appview: Arc<live::Appview>) {
let mut cursor: Option<String> = 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<Runner>) {
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<Runner>) {
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<Runner>) {
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<Runner>) {
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<RepoRecord>>, 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<Listed>,
cursor: Option<String>,
}
#[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<Runner>, watch: Watch, door: Arc<Notify>) {
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<Runner>, watch: Watch, door: Arc<Notify>) {
// 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<std::time::Instant>, 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<Notify>>;
/// 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<Runner>) -> impl Future<Output = ()> {
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<Runner>,
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<turns::Turn>,
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<HashSet<String>>,
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<live::Progress>,
/// 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<Finishing<'a>>,
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<T>(cancelled: &mut Cancelled, f: impl Future<Output = T>) -> Option<T> {
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<u64> {
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<String, Vec<String>>| 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::<Vec<_>>();
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<Options, String> {
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<serde_json::Value>| 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::<Vec<_>>(), [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);
}
}