crates/flow-core/src/api.rs at v0.5.2 · andymccall/oxygenflow

◆ v0.5.2 oxygenflow / crates / flow-core / src / api.rs
19.0 KiB · Rust · last changed A job's finish report says where its time went: image, prepare, run, finish (refs #35) 8 days ago source blame history raw
//! 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<String>,
    /// 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<u32>,
    /// vCPUs the runner has free for another job. Absent: no limit.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub free_cpus: Option<u32>,
    /// What the runner is, for the hub's Runners page.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub info: Option<RunnerInfo>,
    /// 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<Vec<String>>,
    /// 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<std::collections::BTreeMap<String, String>>,
    /// 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<Vec<i64>>,
    /// 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<Facts>,
    /// 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<SwarmAddr>,
    /// 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<Vec<String>>,
    /// 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<std::collections::BTreeMap<String, u64>>,
}

/// 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<SwarmPeer>,
    /// 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<bool>,
}

/// `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<String>,
}

/// 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<String>,
}

/// 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<u64>,
    /// The 1-minute load average, times 100.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub load_x100: Option<u32>,
    /// 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<u32>,
    /// 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<u8>,
    /// The kernel release.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub kernel: Option<String>,
}

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<String>,
    /// Jobs at once, and the memory (MiB) and vCPUs they share.
    pub max_jobs: u32,
    #[serde(default)]
    pub memory_mib: Option<u32>,
    #[serde(default)]
    pub cpus: Option<u32>,
}

impl RunnerInfo {
    /// This machine and build, with the given labels and capacity.
    pub fn here(labels: Vec<String>, max_jobs: u32, memory_mib: Option<u32>, cpus: Option<u32>) -> 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<String>,
}

/// 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<String>,
    /// 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<SubrepoInput>,
    /// 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<String, String>,
    /// 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<SecretInput>,
    /// 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<ArtifactInput>,
    /// 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<u64>,
    /// 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<String>,
}

/// 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<UpdateOffer>,
    /// 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<String>,
}

/// 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<Record>,
}

/// `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<i32>,
    pub duration_ms: u64,
    pub receipt: Option<Receipt>,
    /// 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<String>,
    /// Why the runner could not run the job (rootfs missing, VM failed).
    pub error: Option<String>,
    /// 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<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub peak_memory_mib: Option<u64>,
    /// Where the job's time went, in milliseconds (oxygenflow#35): `image`
    /// (fetching its image, when the runner lacked it), `prepare` (its
    /// workspace and disks), `run` (the sandbox, from start to stop) and
    /// `finish` (its outputs, caches and receipt).
    #[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")]
    pub phases: std::collections::BTreeMap<String, u64>,
}

/// 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;
`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet
Keyboard
