OXYGENHUB
# 92587de58cef oxygenflow / crates / flow-core / src / api.rs
13.7 KiB · Rust · last changed Artifacts between jobs: input { artifact "x" from "job"; path "/src/x" } 20 days ago source blame history raw
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 
}
56
57 
/// A runner's place in the image swarm: the address peers reach it on,
58 
/// the site it is in (peers are only ever in the same site), and the
59 
/// BLAKE3 of its TLS certificate, which peers check it by.
60 
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
61 
pub struct SwarmAddr {
62 
    pub addr: String,
63 
    #[serde(default)]
64 
    pub site: String,
65 
    #[serde(default)]
66 
    pub cert: String,
67 
}
68
69 
/// `POST /flow/swarm/announce`: a runner starting, continuing or ending a
70 
/// fetch of `image`.
71 
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
72 
pub struct Announce {
73 
    pub image: String,
74 
    /// `fetching`, `done` or `stopped`.
75 
    pub state: String,
76 
}
77
78 
/// The hub's answer to an announce: the peers to ask, and the ticket that
79 
/// admits this runner to them.
80 
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
81 
pub struct Announced {
82 
    pub ticket: String,
83 
    pub peers: Vec<SwarmPeer>,
84 
    /// Announce again after this many seconds while still fetching.
85 
    pub interval_s: u64,
86 
}
87
88 
/// A peer in the swarm for one image.
89 
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
90 
pub struct SwarmPeer {
91 
    pub runner: i64,
92 
    pub name: String,
93 
    pub addr: String,
94 
    /// The BLAKE3 of its TLS certificate: a peer that shows another is not
95 
    /// this runner.
96 
    pub cert: String,
97 
    /// It holds the whole image (rather than fetching it now).
98 
    pub complete: bool,
99 
}
100
101 
/// A runner's own measurements, sent with every claim (spec §11.1): cheap
102 
/// ones taken each time, the speed from a short benchmark run at start and
103 
/// again while idle. Whole numbers, so a claim compares exactly.
104 
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
105 
pub struct Facts {
106 
    /// Free disk where jobs' disks are made, in MiB.
107 
    #[serde(default, skip_serializing_if = "Option::is_none")]
108 
    pub free_disk_mib: Option<u64>,
109 
    /// The 1-minute load average, times 100.
110 
    #[serde(default, skip_serializing_if = "Option::is_none")]
111 
    pub load_x100: Option<u32>,
112 
    /// The benchmark's score: 100 on the reference machine (Earth), twice
113 
    /// as fast scoring 200.
114 
    #[serde(default, skip_serializing_if = "Option::is_none")]
115 
    pub speed: Option<u32>,
116 
    /// The speed class, 1 (slow) to 5 (fast); each class twice the one
117 
    /// below (spec §11.1).
118 
    #[serde(default, skip_serializing_if = "Option::is_none")]
119 
    pub speed_class: Option<u8>,
120 
    /// The kernel release.
121 
    #[serde(default, skip_serializing_if = "Option::is_none")]
122 
    pub kernel: Option<String>,
123 
}
124
125 
impl Facts {
126 
    /// The class of a speed score: 3 for the reference machine's 100, each
127 
    /// class a factor of two.
128 
    pub fn class_of(speed: u32) -> u8 {
129 
        match speed {
130 
            0..=34 => 1,
131 
            35..=69 => 2,
132 
            70..=139 => 3,
133 
            140..=279 => 4,
134 
            _ => 5,
135 
        }
136 
    }
137 
}
138
139 
/// A runner's description: shown on the hub's Runners page, and what
140 
/// scheduling by label and architecture will match.
141 
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
142 
pub struct RunnerInfo {
143 
    /// `x86_64`, `aarch64`, `riscv64`, …
144 
    pub arch: String,
145 
    pub os: String,
146 
    /// The `flow` version.
147 
    pub version: String,
148 
    /// From flow.toml `[runner] labels`.
149 
    #[serde(default)]
150 
    pub labels: Vec<String>,
151 
    /// Jobs at once, and the memory (MiB) and vCPUs they share.
152 
    pub max_jobs: u32,
153 
    #[serde(default)]
154 
    pub memory_mib: Option<u32>,
155 
    #[serde(default)]
156 
    pub cpus: Option<u32>,
157 
}
158
159 
impl RunnerInfo {
160 
    /// This machine and build, with the given labels and capacity.
161 
    pub fn here(labels: Vec<String>, max_jobs: u32, memory_mib: Option<u32>, cpus: Option<u32>) -> RunnerInfo {
162 
        RunnerInfo {
163 
            arch: std::env::consts::ARCH.into(),
164 
            os: std::env::consts::OS.into(),
165 
            version: env!("CARGO_PKG_VERSION").into(),
166 
            labels,
167 
            max_jobs,
168 
            memory_mib,
169 
            cpus,
170 
        }
171 
    }
172 
}
173
174 
/// `POST /api/v1/flow/runners/register`: exchange a one-time registration
175 
/// token from the hub for a runner credential, binding the runner's
176 
/// signing identity to it.
177 
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
178 
pub struct Register {
179 
    pub token: String,
180 
    pub name: String,
181 
    /// The runner's Oxygen identity, hex of its encoding: its key signs
182 
    /// every receipt, and the hub accepts no other for this runner.
183 
    pub identity: String,
184 
    pub info: RunnerInfo,
185 
}
186
187 
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
188 
pub struct Registered {
189 
    pub id: i64,
190 
    pub name: String,
191 
    /// The runner's bearer token from now on; shown once.
192 
    pub credential: String,
193 
}
194
195 
/// A pinned subrepo to put in the workspace: `repo`'s state `heads`, at `path`.
196 
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
197 
pub struct SubrepoInput {
198 
    pub path: String,
199 
    /// `owner/name` on the hub.
200 
    pub repo: String,
201 
    pub clone_url: String,
202 
    pub heads: Vec<String>,
203 
}
204
205 
/// A secret for a job, as the hub supplies it (spec §9): written by the
206 
/// runner at `path` in the sandbox, 0400, on an in-memory mount; never in
207 
/// the environment, the receipt or an artifact.
208 
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
209 
pub struct SecretInput {
210 
    pub name: String,
211 
    pub path: String,
212 
    pub value: String,
213 
}
214
215 
/// A claimed job: everything the runner needs, nothing more.
216 
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
217 
pub struct Claim {
218 
    pub job_id: i64,
219 
    pub run_id: i64,
220 
    /// `owner/name`.
221 
    pub repo: String,
222 
    /// Where to fetch the repository from (`oxy` wire protocol).
223 
    pub clone_url: String,
224 
    /// Read access to this repository only, until the job finishes.
225 
    pub token: String,
226 
    /// The state to build, as full change hashes.
227 
    pub heads: Vec<String>,
228 
    /// Hex BLAKE3 of the canonical pipeline.
229 
    pub pipeline_hash: String,
230 
    pub job: Job,
231 
    pub sandbox: Sandbox,
232 
    /// The hub's limit for this job.
233 
    pub timeout_s: u64,
234 
    /// Subrepos the state pins (Oxygen spec §7.8) that the job may read,
235 
    /// fetched with the same token and written at their paths.
236 
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
237 
    pub subrepos: Vec<SubrepoInput>,
238 
    /// The state's hash on the hub (what checks are keyed by).
239 
    #[serde(default, skip_serializing_if = "String::is_empty")]
240 
    pub state: String,
241 
    /// The bookmark the run is for.
242 
    #[serde(default, skip_serializing_if = "String::is_empty")]
243 
    pub bookmark: String,
244 
    /// The run's number within its repository.
245 
    #[serde(default)]
246 
    pub run_number: i64,
247 
    /// Variables the hub holds for the repository (its site's, its
248 
    /// organisation's and its own, inner overriding outer; oxygenhub#102).
249 
    /// Set under `FLOW_*` and under the pipeline's own `env`, so they are
250 
    /// part of the job's environment and of its receipt. Never secrets:
251 
    /// those are mounted, not put in the environment (spec §9).
252 
    #[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")]
253 
    pub env: std::collections::BTreeMap<String, String>,
254 
    /// The secrets the job declares (`secret` blocks), as the hub holds
255 
    /// them for the repository (oxygenhub#107).
256 
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
257 
    pub secrets: Vec<SecretInput>,
258 
    /// The job's inputs (oxygenflow#58), in the job's order, each with the
259 
    /// blob the producing job's receipt names: fetched from
260 
    /// `/api/v1/flow/jobs/{job_id}/inputs/{index}` and checked against it.
261 
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
262 
    pub inputs: Vec<ArtifactInput>,
263 
}
264
265 
/// An artifact a job takes (oxygenflow#58), as the hub hands it over.
266 
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
267 
pub struct ArtifactInput {
268 
    pub artifact: String,
269 
    /// The producing job.
270 
    pub from: String,
271 
    /// Where it goes, under `/src`.
272 
    pub path: String,
273 
    /// The Oxygen blob reference its receipt names, hex.
274 
    pub blob: String,
275 
}
276
277 
impl Claim {
278 
    /// What a job is told about what it builds (docs/m0.md §4):
279 
    /// `FLOW_REPO`, `FLOW_STATE`, `FLOW_HEADS` (full hashes joined with
280 
    /// `+`), `FLOW_BOOKMARK`, `FLOW_RUN` and `FLOW_JOB`. Unknown values are
281 
    /// left out.
282 
    pub fn flow_env(&self) -> Vec<(String, String)> {
283 
        let mut v = vec![
284 
            ("FLOW_REPO".to_string(), self.repo.clone()),
285 
            ("FLOW_HEADS".to_string(), self.heads.join("+")),
286 
            ("FLOW_JOB".to_string(), self.job.name.clone()),
287 
        ];
288 
        if !self.state.is_empty() {
289 
            v.push(("FLOW_STATE".into(), self.state.clone()));
290 
        }
291 
        if !self.bookmark.is_empty() {
292 
            v.push(("FLOW_BOOKMARK".into(), self.bookmark.clone()));
293 
        }
294 
        if self.run_number > 0 {
295 
            v.push(("FLOW_RUN".into(), self.run_number.to_string()));
296 
        }
297 
        for (k, val) in &self.env {
298 
            if !v.iter().any(|(n, _)| n == k) {
299 
                v.push((k.clone(), val.clone()));
300 
            }
301 
        }
302 
        v
303 
    }
304 
}
305
306 
/// One line of output (docs/m0.md §2).
307 
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
308 
pub struct Record {
309 
    /// Milliseconds since the job started.
310 
    pub t: u64,
311 
    /// `out`, `err`, `console` (the VM outside the job) or `flow` (the runner).
312 
    pub stream: String,
313 
    pub text: String,
314 
}
315
316 
/// `POST /api/v1/flow/jobs/{id}/decline` (spec §11.5): the runner will
317 
/// not run a job it was handed, and says why. The hub queues the job
318 
/// again and offers it to this runner no more until its profile changes.
319 
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
320 
pub struct Decline {
321 
    pub reason: String,
322 
}
323
324 
/// `POST /api/v1/flow/jobs/{id}/log`.
325 
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
326 
pub struct LogBatch {
327 
    pub records: Vec<Record>,
328 
}
329
330 
/// `POST /api/v1/flow/jobs/{id}/finish`.
331 
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
332 
pub struct Finish {
333 
    /// The job's exit code; absent when the runner could not run it.
334 
    pub exit_code: Option<i32>,
335 
    pub duration_ms: u64,
336 
    pub receipt: Option<Receipt>,
337 
    /// The receipt bound to the job's state and signed by the runner
338 
    /// (Oxygen spec §14.6), hex of the signed receipt's bytes.
339 
    #[serde(default, skip_serializing_if = "Option::is_none")]
340 
    pub signed_receipt: Option<String>,
341 
    /// Why the runner could not run the job (rootfs missing, VM failed).
342 
    pub error: Option<String>,
343 
    /// The most disk the job's disks and workspace took, in MiB, and the
344 
    /// most memory its sandbox held (oxygenflow#41): the hub learns from
345 
    /// them what the job's next run needs.
346 
    #[serde(default, skip_serializing_if = "Option::is_none")]
347 
    pub peak_disk_mib: Option<u64>,
348 
    #[serde(default, skip_serializing_if = "Option::is_none")]
349 
    pub peak_memory_mib: Option<u64>,
350 
}
351
352 
/// Longest text of one record the hub keeps; longer lines are cut.
353 
pub const MAX_RECORD_TEXT: usize = 16 * 1024;
354 
/// Most records the hub keeps per job; later ones are dropped with a note.
355 
pub const MAX_RECORDS: usize = 50_000;