OXYGENHUB
# c5b702e59c1f oxygenflow / crates / flow-core / src / api.rs
8.8 KiB · Rust · last changed Architecture awareness: image manifests, arch on a sandbox, archs on the claim 1 month 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 
    /// The jobs this runner holds now: running, or finished with a result
37 
    /// not yet reported. The hub ends any other job it has as running on
38 
    /// this runner, which the runner lost by restarting or never received
39 
    /// (oxygenhub#71). Absent: the runner does not say.
40 
    #[serde(default, skip_serializing_if = "Option::is_none")]
41 
    pub holding: Option<Vec<i64>>,
42 
}
43
44 
/// A runner's description: shown on the hub's Runners page, and what
45 
/// scheduling by label and architecture will match.
46 
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
47 
pub struct RunnerInfo {
48 
    /// `x86_64`, `aarch64`, `riscv64`, …
49 
    pub arch: String,
50 
    pub os: String,
51 
    /// The `flow` version.
52 
    pub version: String,
53 
    /// From flow.toml `[runner] labels`.
54 
    #[serde(default)]
55 
    pub labels: Vec<String>,
56 
    /// Jobs at once, and the memory (MiB) and vCPUs they share.
57 
    pub max_jobs: u32,
58 
    #[serde(default)]
59 
    pub memory_mib: Option<u32>,
60 
    #[serde(default)]
61 
    pub cpus: Option<u32>,
62 
}
63
64 
impl RunnerInfo {
65 
    /// This machine and build, with the given labels and capacity.
66 
    pub fn here(labels: Vec<String>, max_jobs: u32, memory_mib: Option<u32>, cpus: Option<u32>) -> RunnerInfo {
67 
        RunnerInfo {
68 
            arch: std::env::consts::ARCH.into(),
69 
            os: std::env::consts::OS.into(),
70 
            version: env!("CARGO_PKG_VERSION").into(),
71 
            labels,
72 
            max_jobs,
73 
            memory_mib,
74 
            cpus,
75 
        }
76 
    }
77 
}
78
79 
/// `POST /api/v1/flow/runners/register`: exchange a one-time registration
80 
/// token from the hub for a runner credential, binding the runner's
81 
/// signing identity to it.
82 
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
83 
pub struct Register {
84 
    pub token: String,
85 
    pub name: String,
86 
    /// The runner's Oxygen identity, hex of its encoding: its key signs
87 
    /// every receipt, and the hub accepts no other for this runner.
88 
    pub identity: String,
89 
    pub info: RunnerInfo,
90 
}
91
92 
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
93 
pub struct Registered {
94 
    pub id: i64,
95 
    pub name: String,
96 
    /// The runner's bearer token from now on; shown once.
97 
    pub credential: String,
98 
}
99
100 
/// A pinned subrepo to put in the workspace: `repo`'s state `heads`, at `path`.
101 
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
102 
pub struct SubrepoInput {
103 
    pub path: String,
104 
    /// `owner/name` on the hub.
105 
    pub repo: String,
106 
    pub clone_url: String,
107 
    pub heads: Vec<String>,
108 
}
109
110 
/// A secret for a job, as the hub supplies it (spec §9): written by the
111 
/// runner at `path` in the sandbox, 0400, on an in-memory mount; never in
112 
/// the environment, the receipt or an artifact.
113 
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
114 
pub struct SecretInput {
115 
    pub name: String,
116 
    pub path: String,
117 
    pub value: String,
118 
}
119
120 
/// A claimed job: everything the runner needs, nothing more.
121 
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
122 
pub struct Claim {
123 
    pub job_id: i64,
124 
    pub run_id: i64,
125 
    /// `owner/name`.
126 
    pub repo: String,
127 
    /// Where to fetch the repository from (`oxy` wire protocol).
128 
    pub clone_url: String,
129 
    /// Read access to this repository only, until the job finishes.
130 
    pub token: String,
131 
    /// The state to build, as full change hashes.
132 
    pub heads: Vec<String>,
133 
    /// Hex BLAKE3 of the canonical pipeline.
134 
    pub pipeline_hash: String,
135 
    pub job: Job,
136 
    pub sandbox: Sandbox,
137 
    /// The hub's limit for this job.
138 
    pub timeout_s: u64,
139 
    /// Subrepos the state pins (Oxygen spec §7.8) that the job may read,
140 
    /// fetched with the same token and written at their paths.
141 
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
142 
    pub subrepos: Vec<SubrepoInput>,
143 
    /// The state's hash on the hub (what checks are keyed by).
144 
    #[serde(default, skip_serializing_if = "String::is_empty")]
145 
    pub state: String,
146 
    /// The bookmark the run is for.
147 
    #[serde(default, skip_serializing_if = "String::is_empty")]
148 
    pub bookmark: String,
149 
    /// The run's number within its repository.
150 
    #[serde(default)]
151 
    pub run_number: i64,
152 
    /// Variables the hub holds for the repository (its site's, its
153 
    /// organisation's and its own, inner overriding outer; oxygenhub#102).
154 
    /// Set under `FLOW_*` and under the pipeline's own `env`, so they are
155 
    /// part of the job's environment and of its receipt. Never secrets:
156 
    /// those are mounted, not put in the environment (spec §9).
157 
    #[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")]
158 
    pub env: std::collections::BTreeMap<String, String>,
159 
    /// The secrets the job declares (`secret` blocks), as the hub holds
160 
    /// them for the repository (oxygenhub#107).
161 
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
162 
    pub secrets: Vec<SecretInput>,
163 
}
164
165 
impl Claim {
166 
    /// What a job is told about what it builds (docs/m0.md §4):
167 
    /// `FLOW_REPO`, `FLOW_STATE`, `FLOW_HEADS` (full hashes joined with
168 
    /// `+`), `FLOW_BOOKMARK`, `FLOW_RUN` and `FLOW_JOB`. Unknown values are
169 
    /// left out.
170 
    pub fn flow_env(&self) -> Vec<(String, String)> {
171 
        let mut v = vec![
172 
            ("FLOW_REPO".to_string(), self.repo.clone()),
173 
            ("FLOW_HEADS".to_string(), self.heads.join("+")),
174 
            ("FLOW_JOB".to_string(), self.job.name.clone()),
175 
        ];
176 
        if !self.state.is_empty() {
177 
            v.push(("FLOW_STATE".into(), self.state.clone()));
178 
        }
179 
        if !self.bookmark.is_empty() {
180 
            v.push(("FLOW_BOOKMARK".into(), self.bookmark.clone()));
181 
        }
182 
        if self.run_number > 0 {
183 
            v.push(("FLOW_RUN".into(), self.run_number.to_string()));
184 
        }
185 
        for (k, val) in &self.env {
186 
            if !v.iter().any(|(n, _)| n == k) {
187 
                v.push((k.clone(), val.clone()));
188 
            }
189 
        }
190 
        v
191 
    }
192 
}
193
194 
/// One line of output (docs/m0.md §2).
195 
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
196 
pub struct Record {
197 
    /// Milliseconds since the job started.
198 
    pub t: u64,
199 
    /// `out`, `err`, `console` (the VM outside the job) or `flow` (the runner).
200 
    pub stream: String,
201 
    pub text: String,
202 
}
203
204 
/// `POST /api/v1/flow/jobs/{id}/log`.
205 
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
206 
pub struct LogBatch {
207 
    pub records: Vec<Record>,
208 
}
209
210 
/// `POST /api/v1/flow/jobs/{id}/finish`.
211 
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
212 
pub struct Finish {
213 
    /// The job's exit code; absent when the runner could not run it.
214 
    pub exit_code: Option<i32>,
215 
    pub duration_ms: u64,
216 
    pub receipt: Option<Receipt>,
217 
    /// The receipt bound to the job's state and signed by the runner
218 
    /// (Oxygen spec §14.6), hex of the signed receipt's bytes.
219 
    #[serde(default, skip_serializing_if = "Option::is_none")]
220 
    pub signed_receipt: Option<String>,
221 
    /// Why the runner could not run the job (rootfs missing, VM failed).
222 
    pub error: Option<String>,
223 
}
224
225 
/// Longest text of one record the hub keeps; longer lines are cut.
226 
pub const MAX_RECORD_TEXT: usize = 16 * 1024;
227 
/// Most records the hub keeps per job; later ones are dropped with a note.
228 
pub const MAX_RECORDS: usize = 50_000;