//! The runner API between OxygenHub and `flow runner` (docs/m0.md §5). //! Both sides use these types, so they cannot drift. use crate::pipeline::{Job, Sandbox}; use crate::receipt::Receipt; use serde::{Deserialize, Serialize}; /// `POST /api/v1/flow/claim`. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct ClaimRequest { /// The runner's name, shown on job pages. pub runner: String, /// Tiers this runner can run (`microvm`, `syscall`). pub runtimes: Vec, /// Memory the runner has free for another job, in MiB: the hub hands /// out only a job whose sandbox fits. Absent: no limit. #[serde(default, skip_serializing_if = "Option::is_none")] pub free_memory_mib: Option, /// vCPUs the runner has free for another job. Absent: no limit. #[serde(default, skip_serializing_if = "Option::is_none")] pub free_cpus: Option, /// What the runner is, for the hub's Runners page. #[serde(default, skip_serializing_if = "Option::is_none")] pub info: Option, /// The rootfs images the runner holds (BLAKE3 hex, as pipelines pin /// them): the hub offers only jobs whose image is among them /// (oxygenhub#73). Absent: the runner does not say, and any job may be /// offered. #[serde(default, skip_serializing_if = "Option::is_none")] pub images: Option>, /// The architecture of each image it holds, by hash, as the image's /// manifest records it (spec §11.2): the hub offers a job only to a /// runner whose architecture its image is for. #[serde(default, skip_serializing_if = "Option::is_none")] pub image_archs: Option>, /// Whether the runner fetches a job's image from the hub when it lacks /// it (oxygenflow#39): the hub may then offer it jobs whose image only /// the hub holds. #[serde(default, skip_serializing_if = "std::ops::Not::not")] pub fetches_images: bool, /// The jobs this runner holds now: running, or finished with a result /// not yet reported. The hub ends any other job it has as running on /// this runner, which the runner lost by restarting or never received /// (oxygenhub#71). Absent: the runner does not say. #[serde(default, skip_serializing_if = "Option::is_none")] pub holding: Option>, /// What the runner measured about itself just now (spec §11.1, /// docs/jetstream.md §6). Absent: an older runner. #[serde(default, skip_serializing_if = "Option::is_none")] pub facts: Option, /// Where this runner serves image chunks to its peers (docs/jetstream.md /// §8.1). Absent: it is not in a swarm. #[serde(default, skip_serializing_if = "Option::is_none")] pub swarm: Option, /// The build caches this runner holds a disk of (oxygenflow#51), each /// [`cache_slot`] of a repository and job, so the hub can send a job /// where its cache is. Absent: an older runner. #[serde(default, skip_serializing_if = "Option::is_none")] pub caches: Option>, /// When each of [`ClaimRequest::caches`] last kept a disk, Unix ms, so /// the hub can send a job to the runner whose copy is freshest when /// several hold one (oxygenflow#54). Absent: an older runner. #[serde(default, skip_serializing_if = "Option::is_none")] pub cache_times: Option>, } /// A name safe as one path segment: kept readable, with a short hash so /// two names that clean up alike stay apart. The build cache's pool is laid /// out by these (oxygenflow#28). pub fn segment(s: &str) -> String { let clean: String = s.chars().map(|c| if c.is_ascii_alphanumeric() || "._-".contains(c) { c } else { '_' }).take(48).collect(); format!("{clean}-{}", &blake3::hash(s.as_bytes()).to_hex()[..8]) } /// Where a repository's job keeps its build caches, as runners report /// them in [`ClaimRequest::caches`] (oxygenflow#51): `repo/job`, each part /// a [`segment`]. pub fn cache_slot(repo: &str, job: &str) -> String { format!("{}/{}", segment(repo), segment(job)) } /// A runner's place in the image swarm: the address peers reach it on, /// the site it is in (peers are only ever in the same site), and the /// BLAKE3 of its TLS certificate, which peers check it by. #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct SwarmAddr { pub addr: String, #[serde(default)] pub site: String, #[serde(default)] pub cert: String, } /// `POST /flow/swarm/announce`: a runner starting, continuing or ending a /// fetch of `image`. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct Announce { pub image: String, /// `fetching`, `done` or `stopped`. pub state: String, } /// The hub's answer to an announce: the peers to ask, and the ticket that /// admits this runner to them. #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct Announced { pub ticket: String, pub peers: Vec, /// Announce again after this many seconds while still fetching. pub interval_s: u64, /// Whether the hub holds the image's chunks and so may be asked for /// them (jetstream.md §8.1.1): `Some(false)` for an image only runners /// hold; absent from a hub that predates it, which always seeds. #[serde(default, skip_serializing_if = "Option::is_none")] pub seed: Option, } /// `POST /flow/images/held`: the images a runner holds, by hash /// (oxygenflow#94). #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct HeldImages { pub images: Vec, } /// The hub's answer to [`HeldImages`]: those it keeps no manifest for, which /// the runner sends with `PUT /flow/images/{hash}/held`. #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct WantedManifests { pub wanted: Vec, } /// A peer in the swarm for one image. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct SwarmPeer { pub runner: i64, pub name: String, pub addr: String, /// The BLAKE3 of its TLS certificate: a peer that shows another is not /// this runner. pub cert: String, /// It holds the whole image (rather than fetching it now). pub complete: bool, } /// A runner's own measurements, sent with every claim (spec §11.1): cheap /// ones taken each time, the speed from a short benchmark run at start and /// again while idle. Whole numbers, so a claim compares exactly. #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct Facts { /// Free disk where jobs' disks are made, in MiB. #[serde(default, skip_serializing_if = "Option::is_none")] pub free_disk_mib: Option, /// The 1-minute load average, times 100. #[serde(default, skip_serializing_if = "Option::is_none")] pub load_x100: Option, /// The benchmark's score: 100 on the reference machine (Earth), twice /// as fast scoring 200. #[serde(default, skip_serializing_if = "Option::is_none")] pub speed: Option, /// The speed class, 1 (slow) to 5 (fast); each class twice the one /// below (spec §11.1). #[serde(default, skip_serializing_if = "Option::is_none")] pub speed_class: Option, /// The kernel release. #[serde(default, skip_serializing_if = "Option::is_none")] pub kernel: Option, } impl Facts { /// The class of a speed score: 3 for the reference machine's 100, each /// class a factor of two. pub fn class_of(speed: u32) -> u8 { match speed { 0..=34 => 1, 35..=69 => 2, 70..=139 => 3, 140..=279 => 4, _ => 5, } } } /// A runner's description: shown on the hub's Runners page, and what /// scheduling by label and architecture will match. #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct RunnerInfo { /// `x86_64`, `aarch64`, `riscv64`, … pub arch: String, pub os: String, /// The `flow` version. pub version: String, /// From flow.toml `[runner] labels`. #[serde(default)] pub labels: Vec, /// Jobs at once, and the memory (MiB) and vCPUs they share. pub max_jobs: u32, #[serde(default)] pub memory_mib: Option, #[serde(default)] pub cpus: Option, } impl RunnerInfo { /// This machine and build, with the given labels and capacity. pub fn here(labels: Vec, max_jobs: u32, memory_mib: Option, cpus: Option) -> RunnerInfo { RunnerInfo { arch: std::env::consts::ARCH.into(), os: std::env::consts::OS.into(), version: env!("CARGO_PKG_VERSION").into(), labels, max_jobs, memory_mib, cpus, } } } /// `POST /api/v1/flow/runners/register`: exchange a one-time registration /// token from the hub for a runner credential, binding the runner's /// signing identity to it. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct Register { pub token: String, pub name: String, /// The runner's Oxygen identity, hex of its encoding: its key signs /// every receipt, and the hub accepts no other for this runner. pub identity: String, pub info: RunnerInfo, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct Registered { pub id: i64, pub name: String, /// The runner's bearer token from now on; shown once. pub credential: String, } /// A pinned subrepo to put in the workspace: `repo`'s state `heads`, at `path`. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct SubrepoInput { pub path: String, /// `owner/name` on the hub. pub repo: String, pub clone_url: String, pub heads: Vec, } /// A secret for a job, as the hub supplies it (spec §9): written by the /// runner at `path` in the sandbox, 0400, on an in-memory mount; never in /// the environment, the receipt or an artifact. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct SecretInput { pub name: String, pub path: String, pub value: String, } /// A claimed job: everything the runner needs, nothing more. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct Claim { pub job_id: i64, pub run_id: i64, /// `owner/name`. pub repo: String, /// Where to fetch the repository from (`oxy` wire protocol). pub clone_url: String, /// Read access to this repository only, until the job finishes. pub token: String, /// The state to build, as full change hashes. pub heads: Vec, /// Hex BLAKE3 of the canonical pipeline. pub pipeline_hash: String, pub job: Job, pub sandbox: Sandbox, /// The hub's limit for this job. pub timeout_s: u64, /// Subrepos the state pins (Oxygen spec §7.8) that the job may read, /// fetched with the same token and written at their paths. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub subrepos: Vec, /// The state's hash on the hub (what checks are keyed by). #[serde(default, skip_serializing_if = "String::is_empty")] pub state: String, /// The bookmark the run is for. #[serde(default, skip_serializing_if = "String::is_empty")] pub bookmark: String, /// The run's number within its repository. #[serde(default)] pub run_number: i64, /// Variables the hub holds for the repository (its site's, its /// organisation's and its own, inner overriding outer; oxygenhub#102). /// Set under `FLOW_*` and under the pipeline's own `env`, so they are /// part of the job's environment and of its receipt. Never secrets: /// those are mounted, not put in the environment (spec §9). #[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")] pub env: std::collections::BTreeMap, /// The secrets the job declares (`secret` blocks), as the hub holds /// them for the repository (oxygenhub#107). #[serde(default, skip_serializing_if = "Vec::is_empty")] pub secrets: Vec, /// The job's inputs (oxygenflow#58), in the job's order, each with the /// blob the producing job's receipt names: fetched from /// `/api/v1/flow/jobs/{job_id}/inputs/{index}` and checked against it. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub inputs: Vec, /// How long the job is expected to take, from its recent passing runs, /// when the hub has learned it (#76): a runner leaving with some notice /// finishes a job whose rest fits it and hands back the others. #[serde(default, skip_serializing_if = "Option::is_none")] pub expected_ms: Option, /// The checkpoint an earlier attempt saved before its runner handed the /// job back (oxygenflow#77): the blob to fetch from /// `/api/v1/flow/jobs/{job_id}/checkpoint` and check, then restore /// before the job runs. #[serde(default, skip_serializing_if = "Option::is_none")] pub checkpoint: Option, } /// An artifact a job takes (oxygenflow#58), as the hub hands it over. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct ArtifactInput { pub artifact: String, /// The producing job. pub from: String, /// Where it goes, under `/src`. pub path: String, /// The Oxygen blob reference its receipt names, hex. pub blob: String, } impl Claim { /// What a job is told about what it builds (docs/m0.md §4): /// `FLOW_REPO`, `FLOW_STATE`, `FLOW_HEADS` (full hashes joined with /// `+`), `FLOW_BOOKMARK`, `FLOW_RUN` and `FLOW_JOB`. Unknown values are /// left out. pub fn flow_env(&self) -> Vec<(String, String)> { let mut v = vec![ ("FLOW_REPO".to_string(), self.repo.clone()), ("FLOW_HEADS".to_string(), self.heads.join("+")), ("FLOW_JOB".to_string(), self.job.name.clone()), ]; if !self.state.is_empty() { v.push(("FLOW_STATE".into(), self.state.clone())); } if !self.bookmark.is_empty() { v.push(("FLOW_BOOKMARK".into(), self.bookmark.clone())); } if self.run_number > 0 { v.push(("FLOW_RUN".into(), self.run_number.to_string())); } for (k, val) in &self.env { if !v.iter().any(|(n, _)| n == k) { v.push((k.clone(), val.clone())); } } v } } /// One line of output (docs/m0.md §2). #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct Record { /// Milliseconds since the job started. pub t: u64, /// `out`, `err`, `console` (the VM outside the job) or `flow` (the runner). pub stream: String, pub text: String, } /// `POST /api/v1/flow/jobs/{id}/decline` (spec §11.5): the runner will /// not run a job it was handed, and says why. The hub queues the job /// again and offers it to this runner no more until its profile changes. #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct Decline { pub reason: String, } /// `GET /api/v1/flow/runner/update` (docs/jetstream.md §8.2, oxygenhub#322): /// the update policy that applies to the asking runner, and the build it /// should run when that is not the one it runs. #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct UpdateAnswer { pub policy: UpdatePolicy, /// The build to install now; none when updates are off, the runner is /// up to date, it is outside its window, or no build suits it. #[serde(default)] pub offer: Option, /// Why there is no offer, or what the offer is for, in words. #[serde(default)] pub note: String, } /// A runner's update policy once every level is applied (nearest wins). #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct UpdatePolicy { /// Off unless some level turned it on. pub updates: bool, /// `stable`, `canary`, or a version such as `0.3.4`. pub channel: String, /// Hours (UTC) in which it may update, from and to, when limited. #[serde(default)] pub window: Option<(u8, u8)>, /// Fetch and check, say what would happen, do not swap. #[serde(default)] pub dry_run: bool, /// Where each setting came from: `site`, `owner acme`, `group canary`, /// `runner`, or `default`, in the order updates, channel, window, dry run. #[serde(default)] pub from: Vec, } /// A build to install: a release asset of the site's runner releases. #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct UpdateOffer { pub version: String, pub tag: String, /// The asset's name, `flow-linux-x86_64`. pub asset: String, /// Where to download it, relative to the hub. pub url: String, /// Its BLAKE3, as the hub stores and attests it: the download must match. pub blake3: String, pub size: u64, /// Older than the runner's build: a pinned channel going back. #[serde(default)] pub older: bool, } /// `POST /api/v1/flow/runner/update`: what the runner did about its update, /// shown on its page. #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct UpdateReport { pub note: String, } /// `POST /api/v1/flow/jobs/{id}/log`. #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct LogBatch { pub records: Vec, } /// `POST /api/v1/flow/jobs/{id}/finish`. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct Finish { /// The job's exit code; absent when the runner could not run it. pub exit_code: Option, pub duration_ms: u64, pub receipt: Option, /// The receipt bound to the job's state and signed by the runner /// (Oxygen spec §14.6), hex of the signed receipt's bytes. #[serde(default, skip_serializing_if = "Option::is_none")] pub signed_receipt: Option, /// Why the runner could not run the job (rootfs missing, VM failed). pub error: Option, /// The most disk the job's disks and workspace took, in MiB, and the /// most memory its sandbox held (oxygenflow#41): the hub learns from /// them what the job's next run needs. #[serde(default, skip_serializing_if = "Option::is_none")] pub peak_disk_mib: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub peak_memory_mib: Option, } /// Longest text of one record the hub keeps; longer lines are cut. pub const MAX_RECORD_TEXT: usize = 16 * 1024; /// Most records the hub keeps per job; later ones are dropped with a note. pub const MAX_RECORDS: usize = 50_000;