api.rs 100.0%

48 of 48 lines run, 7 of 14 functions · run #185 on main · run missed

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

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

/// 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

1		//! The runner API between OxygenHub and `flow runner` (docs/m0.md §5).
2		//! Both sides use these types, so they cannot drift.
3
4		use crate::pipeline::{Job, Sandbox};
5		use crate::receipt::Receipt;
6		use serde::{Deserialize, Serialize};
7
8		/// `POST /api/v1/flow/claim`.
9		#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
10		pub struct ClaimRequest {
11		/// The runner's name, shown on job pages.
12		pub runner: String,
13		/// Tiers this runner can run (`microvm`, `syscall`).
14		pub runtimes: Vec<String>,
15		/// Memory the runner has free for another job, in MiB: the hub hands
16		/// out only a job whose sandbox fits. Absent: no limit.
17		#[serde(default, skip_serializing_if = "Option::is_none")]
18		pub free_memory_mib: Option<u32>,
19		/// vCPUs the runner has free for another job. Absent: no limit.
20		#[serde(default, skip_serializing_if = "Option::is_none")]
21		pub free_cpus: Option<u32>,
22		/// What the runner is, for the hub's Runners page.
23		#[serde(default, skip_serializing_if = "Option::is_none")]
24		pub info: Option<RunnerInfo>,
25		/// The rootfs images the runner holds (BLAKE3 hex, as pipelines pin
26		/// them): the hub offers only jobs whose image is among them
27		/// (oxygenhub#73). Absent: the runner does not say, and any job may be
28		/// offered.
29		#[serde(default, skip_serializing_if = "Option::is_none")]
30		pub images: Option<Vec<String>>,
31		/// The architecture of each image it holds, by hash, as the image's
32		/// manifest records it (spec §11.2): the hub offers a job only to a
33		/// runner whose architecture its image is for.
34		#[serde(default, skip_serializing_if = "Option::is_none")]
35		pub image_archs: Option<std::collections::BTreeMap<String, String>>,
36		/// Whether the runner fetches a job's image from the hub when it lacks
37		/// it (oxygenflow#39): the hub may then offer it jobs whose image only
38		/// the hub holds.
39		#[serde(default, skip_serializing_if = "std::ops::Not::not")]
40		pub fetches_images: bool,
41		/// The jobs this runner holds now: running, or finished with a result
42		/// not yet reported. The hub ends any other job it has as running on
43		/// this runner, which the runner lost by restarting or never received
44		/// (oxygenhub#71). Absent: the runner does not say.
45		#[serde(default, skip_serializing_if = "Option::is_none")]
46		pub holding: Option<Vec<i64>>,
47		/// What the runner measured about itself just now (spec §11.1,
48		/// docs/jetstream.md §6). Absent: an older runner.
49		#[serde(default, skip_serializing_if = "Option::is_none")]
50		pub facts: Option<Facts>,
51		/// Where this runner serves image chunks to its peers (docs/jetstream.md
52		/// §8.1). Absent: it is not in a swarm.
53		#[serde(default, skip_serializing_if = "Option::is_none")]
54		pub swarm: Option<SwarmAddr>,
55		/// The build caches this runner holds a disk of (oxygenflow#51), each
56		/// [`cache_slot`] of a repository and job, so the hub can send a job
57		/// where its cache is. Absent: an older runner.
58		#[serde(default, skip_serializing_if = "Option::is_none")]
59		pub caches: Option<Vec<String>>,
60		/// When each of [`ClaimRequest::caches`] last kept a disk, Unix ms, so
61		/// the hub can send a job to the runner whose copy is freshest when
62		/// several hold one (oxygenflow#54). Absent: an older runner.
63		#[serde(default, skip_serializing_if = "Option::is_none")]
64		pub cache_times: Option<std::collections::BTreeMap<String, u64>>,
65		}
66
67		/// A name safe as one path segment: kept readable, with a short hash so
68		/// two names that clean up alike stay apart. The build cache's pool is laid
69		/// out by these (oxygenflow#28).
70	186×	pub fn segment(s: &str) -> String {
71	987×	let clean: String = s.chars().map(\|c\| if c.is_ascii_alphanumeric() \|\| "._-".contains(c) { c } else { '_' }).take(48).collect();
72	186×	format!("{clean}-{}", &blake3::hash(s.as_bytes()).to_hex()[..8])
73	186×	}
74
75		/// Where a repository's job keeps its build caches, as runners report
76		/// them in [`ClaimRequest::caches`] (oxygenflow#51): `repo/job`, each part
77		/// a [`segment`].
78	3×	pub fn cache_slot(repo: &str, job: &str) -> String {
79	3×	format!("{}/{}", segment(repo), segment(job))
80	3×	}
81
82		/// A runner's place in the image swarm: the address peers reach it on,
83		/// the site it is in (peers are only ever in the same site), and the
84		/// BLAKE3 of its TLS certificate, which peers check it by.
85		#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
86		pub struct SwarmAddr {
87		pub addr: String,
88		#[serde(default)]
89		pub site: String,
90		#[serde(default)]
91		pub cert: String,
92		}
93
94		/// `POST /flow/swarm/announce`: a runner starting, continuing or ending a
95		/// fetch of `image`.
96		#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
97		pub struct Announce {
98		pub image: String,
99		/// `fetching`, `done` or `stopped`.
100		pub state: String,
101		}
102
103		/// The hub's answer to an announce: the peers to ask, and the ticket that
104		/// admits this runner to them.
105		#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
106		pub struct Announced {
107		pub ticket: String,
108		pub peers: Vec<SwarmPeer>,
109		/// Announce again after this many seconds while still fetching.
110		pub interval_s: u64,
111		}
112
113		/// A peer in the swarm for one image.
114		#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
115		pub struct SwarmPeer {
116		pub runner: i64,
117		pub name: String,
118		pub addr: String,
119		/// The BLAKE3 of its TLS certificate: a peer that shows another is not
120		/// this runner.
121		pub cert: String,
122		/// It holds the whole image (rather than fetching it now).
123		pub complete: bool,
124		}
125
126		/// A runner's own measurements, sent with every claim (spec §11.1): cheap
127		/// ones taken each time, the speed from a short benchmark run at start and
128		/// again while idle. Whole numbers, so a claim compares exactly.
129		#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
130		pub struct Facts {
131		/// Free disk where jobs' disks are made, in MiB.
132		#[serde(default, skip_serializing_if = "Option::is_none")]
133		pub free_disk_mib: Option<u64>,
134		/// The 1-minute load average, times 100.
135		#[serde(default, skip_serializing_if = "Option::is_none")]
136		pub load_x100: Option<u32>,
137		/// The benchmark's score: 100 on the reference machine (Earth), twice
138		/// as fast scoring 200.
139		#[serde(default, skip_serializing_if = "Option::is_none")]
140		pub speed: Option<u32>,
141		/// The speed class, 1 (slow) to 5 (fast); each class twice the one
142		/// below (spec §11.1).
143		#[serde(default, skip_serializing_if = "Option::is_none")]
144		pub speed_class: Option<u8>,
145		/// The kernel release.
146		#[serde(default, skip_serializing_if = "Option::is_none")]
147		pub kernel: Option<String>,
148		}
149
150		impl Facts {
151		/// The class of a speed score: 3 for the reference machine's 100, each
152		/// class a factor of two.
153	5×	pub fn class_of(speed: u32) -> u8 {
154	5×	match speed {
155	5×	0..=34 => 1,
156	4×	35..=69 => 2,
157	3×	70..=139 => 3,
158	2×	140..=279 => 4,
159	1×	_ => 5,
160		}
161	5×	}
162		}
163
164		/// A runner's description: shown on the hub's Runners page, and what
165		/// scheduling by label and architecture will match.
166		#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
167		pub struct RunnerInfo {
168		/// `x86_64`, `aarch64`, `riscv64`, …
169		pub arch: String,
170		pub os: String,
171		/// The `flow` version.
172		pub version: String,
173		/// From flow.toml `[runner] labels`.
174		#[serde(default)]
175		pub labels: Vec<String>,
176		/// Jobs at once, and the memory (MiB) and vCPUs they share.
177		pub max_jobs: u32,
178		#[serde(default)]
179		pub memory_mib: Option<u32>,
180		#[serde(default)]
181		pub cpus: Option<u32>,
182		}
183
184		impl RunnerInfo {
185		/// This machine and build, with the given labels and capacity.
186	1×	pub fn here(labels: Vec<String>, max_jobs: u32, memory_mib: Option<u32>, cpus: Option<u32>) -> RunnerInfo {
187	1×	RunnerInfo {
188	1×	arch: std::env::consts::ARCH.into(),
189	1×	os: std::env::consts::OS.into(),
190	1×	version: env!("CARGO_PKG_VERSION").into(),
191	1×	labels,
192	1×	max_jobs,
193	1×	memory_mib,
194	1×	cpus,
195	1×	}
196	1×	}
197		}
198
199		/// `POST /api/v1/flow/runners/register`: exchange a one-time registration
200		/// token from the hub for a runner credential, binding the runner's
201		/// signing identity to it.
202		#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
203		pub struct Register {
204		pub token: String,
205		pub name: String,
206		/// The runner's Oxygen identity, hex of its encoding: its key signs
207		/// every receipt, and the hub accepts no other for this runner.
208		pub identity: String,
209		pub info: RunnerInfo,
210		}
211
212		#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
213		pub struct Registered {
214		pub id: i64,
215		pub name: String,
216		/// The runner's bearer token from now on; shown once.
217		pub credential: String,
218		}
219
220		/// A pinned subrepo to put in the workspace: `repo`'s state `heads`, at `path`.
221		#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
222		pub struct SubrepoInput {
223		pub path: String,
224		/// `owner/name` on the hub.
225		pub repo: String,
226		pub clone_url: String,
227		pub heads: Vec<String>,
228		}
229
230		/// A secret for a job, as the hub supplies it (spec §9): written by the
231		/// runner at `path` in the sandbox, 0400, on an in-memory mount; never in
232		/// the environment, the receipt or an artifact.
233		#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
234		pub struct SecretInput {
235		pub name: String,
236		pub path: String,
237		pub value: String,
238		}
239
240		/// A claimed job: everything the runner needs, nothing more.
241		#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
242		pub struct Claim {
243		pub job_id: i64,
244		pub run_id: i64,
245		/// `owner/name`.
246		pub repo: String,
247		/// Where to fetch the repository from (`oxy` wire protocol).
248		pub clone_url: String,
249		/// Read access to this repository only, until the job finishes.
250		pub token: String,
251		/// The state to build, as full change hashes.
252		pub heads: Vec<String>,
253		/// Hex BLAKE3 of the canonical pipeline.
254		pub pipeline_hash: String,
255		pub job: Job,
256		pub sandbox: Sandbox,
257		/// The hub's limit for this job.
258		pub timeout_s: u64,
259		/// Subrepos the state pins (Oxygen spec §7.8) that the job may read,
260		/// fetched with the same token and written at their paths.
261		#[serde(default, skip_serializing_if = "Vec::is_empty")]
262		pub subrepos: Vec<SubrepoInput>,
263		/// The state's hash on the hub (what checks are keyed by).
264		#[serde(default, skip_serializing_if = "String::is_empty")]
265		pub state: String,
266		/// The bookmark the run is for.
267		#[serde(default, skip_serializing_if = "String::is_empty")]
268		pub bookmark: String,
269		/// The run's number within its repository.
270		#[serde(default)]
271		pub run_number: i64,
272		/// Variables the hub holds for the repository (its site's, its
273		/// organisation's and its own, inner overriding outer; oxygenhub#102).
274		/// Set under `FLOW_*` and under the pipeline's own `env`, so they are
275		/// part of the job's environment and of its receipt. Never secrets:
276		/// those are mounted, not put in the environment (spec §9).
277		#[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")]
278		pub env: std::collections::BTreeMap<String, String>,
279		/// The secrets the job declares (`secret` blocks), as the hub holds
280		/// them for the repository (oxygenhub#107).
281		#[serde(default, skip_serializing_if = "Vec::is_empty")]
282		pub secrets: Vec<SecretInput>,
283		/// The job's inputs (oxygenflow#58), in the job's order, each with the
284		/// blob the producing job's receipt names: fetched from
285		/// `/api/v1/flow/jobs/{job_id}/inputs/{index}` and checked against it.
286		#[serde(default, skip_serializing_if = "Vec::is_empty")]
287		pub inputs: Vec<ArtifactInput>,
288		/// How long the job is expected to take, from its recent passing runs,
289		/// when the hub has learned it (#76): a runner leaving with some notice
290		/// finishes a job whose rest fits it and hands back the others.
291		#[serde(default, skip_serializing_if = "Option::is_none")]
292		pub expected_ms: Option<u64>,
293		/// The checkpoint an earlier attempt saved before its runner handed the
294		/// job back (oxygenflow#77): the blob to fetch from
295		/// `/api/v1/flow/jobs/{job_id}/checkpoint` and check, then restore
296		/// before the job runs.
297		#[serde(default, skip_serializing_if = "Option::is_none")]
298		pub checkpoint: Option<String>,
299		}
300
301		/// An artifact a job takes (oxygenflow#58), as the hub hands it over.
302		#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
303		pub struct ArtifactInput {
304		pub artifact: String,
305		/// The producing job.
306		pub from: String,
307		/// Where it goes, under `/src`.
308		pub path: String,
309		/// The Oxygen blob reference its receipt names, hex.
310		pub blob: String,
311		}
312
313		impl Claim {
314		/// What a job is told about what it builds (docs/m0.md §4):
315		/// `FLOW_REPO`, `FLOW_STATE`, `FLOW_HEADS` (full hashes joined with
316		/// `+`), `FLOW_BOOKMARK`, `FLOW_RUN` and `FLOW_JOB`. Unknown values are
317		/// left out.
318	2×	pub fn flow_env(&self) -> Vec<(String, String)> {
319	2×	let mut v = vec![
320	2×	("FLOW_REPO".to_string(), self.repo.clone()),
321	2×	("FLOW_HEADS".to_string(), self.heads.join("+")),
322	2×	("FLOW_JOB".to_string(), self.job.name.clone()),
323		];
324	2×	if !self.state.is_empty() {
325	1×	v.push(("FLOW_STATE".into(), self.state.clone()));
326	1×	}
327	2×	if !self.bookmark.is_empty() {
328	1×	v.push(("FLOW_BOOKMARK".into(), self.bookmark.clone()));
329	1×	}
330	2×	if self.run_number > 0 {
331	1×	v.push(("FLOW_RUN".into(), self.run_number.to_string()));
332	1×	}
333	2×	for (k, val) in &self.env {
334	7×	if !v.iter().any(\|(n, _)\| n == k) {
335	1×	v.push((k.clone(), val.clone()));
336	1×	}
337		}
338	2×	v
339	2×	}
340		}
341
342		/// One line of output (docs/m0.md §2).
343		#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
344		pub struct Record {
345		/// Milliseconds since the job started.
346		pub t: u64,
347		/// `out`, `err`, `console` (the VM outside the job) or `flow` (the runner).
348		pub stream: String,
349		pub text: String,
350		}
351
352		/// `POST /api/v1/flow/jobs/{id}/decline` (spec §11.5): the runner will
353		/// not run a job it was handed, and says why. The hub queues the job
354		/// again and offers it to this runner no more until its profile changes.
355		#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
356		pub struct Decline {
357		pub reason: String,
358		}
359
360		/// `GET /api/v1/flow/runner/update` (docs/jetstream.md §8.2, oxygenhub#322):
361		/// the update policy that applies to the asking runner, and the build it
362		/// should run when that is not the one it runs.
363		#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
364		pub struct UpdateAnswer {
365		pub policy: UpdatePolicy,
366		/// The build to install now; none when updates are off, the runner is
367		/// up to date, it is outside its window, or no build suits it.
368		#[serde(default)]
369		pub offer: Option<UpdateOffer>,
370		/// Why there is no offer, or what the offer is for, in words.
371		#[serde(default)]
372		pub note: String,
373		}
374
375		/// A runner's update policy once every level is applied (nearest wins).
376		#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
377		pub struct UpdatePolicy {
378		/// Off unless some level turned it on.
379		pub updates: bool,
380		/// `stable`, `canary`, or a version such as `0.3.4`.
381		pub channel: String,
382		/// Hours (UTC) in which it may update, from and to, when limited.
383		#[serde(default)]
384		pub window: Option<(u8, u8)>,
385		/// Fetch and check, say what would happen, do not swap.
386		#[serde(default)]
387		pub dry_run: bool,
388		/// Where each setting came from: `site`, `owner acme`, `group canary`,
389		/// `runner`, or `default`, in the order updates, channel, window, dry run.
390		#[serde(default)]
391		pub from: Vec<String>,
392		}
393
394		/// A build to install: a release asset of the site's runner releases.
395		#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
396		pub struct UpdateOffer {
397		pub version: String,
398		pub tag: String,
399		/// The asset's name, `flow-linux-x86_64`.
400		pub asset: String,
401		/// Where to download it, relative to the hub.
402		pub url: String,
403		/// Its BLAKE3, as the hub stores and attests it: the download must match.
404		pub blake3: String,
405		pub size: u64,
406		/// Older than the runner's build: a pinned channel going back.
407		#[serde(default)]
408		pub older: bool,
409		}
410
411		/// `POST /api/v1/flow/runner/update`: what the runner did about its update,
412		/// shown on its page.
413		#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
414		pub struct UpdateReport {
415		pub note: String,
416		}
417
418		/// `POST /api/v1/flow/jobs/{id}/log`.
419		#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
420		pub struct LogBatch {
421		pub records: Vec<Record>,
422		}
423
424		/// `POST /api/v1/flow/jobs/{id}/finish`.
425		#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
426		pub struct Finish {
427		/// The job's exit code; absent when the runner could not run it.
428		pub exit_code: Option<i32>,
429		pub duration_ms: u64,
430		pub receipt: Option<Receipt>,
431		/// The receipt bound to the job's state and signed by the runner
432		/// (Oxygen spec §14.6), hex of the signed receipt's bytes.
433		#[serde(default, skip_serializing_if = "Option::is_none")]
434		pub signed_receipt: Option<String>,
435		/// Why the runner could not run the job (rootfs missing, VM failed).
436		pub error: Option<String>,
437		/// The most disk the job's disks and workspace took, in MiB, and the
438		/// most memory its sandbox held (oxygenflow#41): the hub learns from
439		/// them what the job's next run needs.
440		#[serde(default, skip_serializing_if = "Option::is_none")]
441		pub peak_disk_mib: Option<u64>,
442		#[serde(default, skip_serializing_if = "Option::is_none")]
443		pub peak_memory_mib: Option<u64>,
444		}
445
446		/// Longest text of one record the hub keeps; longer lines are cut.
447		pub const MAX_RECORD_TEXT: usize = 16 * 1024;
448		/// Most records the hub keeps per job; later ones are dropped with a note.
449		pub const MAX_RECORDS: usize = 50_000;

Coverage / crates/flow-core/src/api.rs 100.0%

Keyboard