OXYGENHUB
# f98a8f7881e0 oxygenflow / crates / flow / src / runner.rs
22.1 KiB · Rust · last changed Register runners with a token; describe them in claims 1 month ago source blame history raw
1 
//! `flow runner` (docs/m0.md §5): claim a job from OxygenHub, fetch the
2 
//! repository into a cache with the job's token, write the state into a
3 
//! workspace, run it in its tier, stream the log, report the receipt.
4 
//! One job at a time.
5
6 
use crate::home::{Home, RunnerConfig, RunnerKey};
7 
use crate::{microvm, workspace};
8 
use anyhow::{bail, Context, Result};
9 
use flow_core::api::{Claim, ClaimRequest, Finish, LogBatch, Record, Register, Registered, RunnerInfo, MAX_RECORD_TEXT};
10 
use flow_core::{scope_tree_hash, Receipt};
11 
use oxy_core::hash::ChangeHash;
12 
use oxy_core::repo::Repo;
13 
use std::path::PathBuf;
14 
use std::time::{Duration, Instant};
15
16 
pub struct Hub {
17 
    base: String,
18 
    token: String,
19 
    agent: ureq::Agent,
20 
}
21
22 
impl Hub {
23 
    pub fn new(cfg: &RunnerConfig) -> Result<Hub> {
24 
        let tls = oxy_core::protocol::tls_client_config(cfg.ca.as_deref()).map_err(|e| anyhow::anyhow!("{e}"))?;
25 
        let agent = ureq::AgentBuilder::new().tls_config(tls).timeout(Duration::from_secs(60)).build();
26 
        Ok(Hub { base: cfg.hub.trim_end_matches('/').to_string(), token: cfg.token.clone(), agent })
27 
    }
28
29 
    /// POST to the hub. A server error (5xx) means nothing was committed,
30 
    /// so it is tried again after 1, 2 and 4 seconds before giving up: a
31 
    /// busy hub costs a job a few seconds, not its log lines or its claim.
32 
    fn post(&self, path: &str, body: &impl serde::Serialize) -> Result<Option<ureq::Response>> {
33 
        self.post_with(path, body, &RETRY_DELAYS)
34 
    }
35
36 
    fn post_with(&self, path: &str, body: &impl serde::Serialize, delays: &[Duration]) -> Result<Option<ureq::Response>> {
37 
        let url = format!("{}/api/v1{path}", self.base);
38 
        let mut delays = delays.iter();
39 
        loop {
40 
            match self.agent.post(&url).set("Authorization", &format!("Bearer {}", self.token)).send_json(body) {
41 
                Ok(r) if r.status() == 204 => return Ok(None),
42 
                Ok(r) => return Ok(Some(r)),
43 
                Err(ureq::Error::Status(code, r)) => {
44 
                    let text = r.into_string().unwrap_or_default();
45 
                    match delays.next() {
46 
                        Some(d) if code >= 500 => std::thread::sleep(*d),
47 
                        _ => bail!("{url}: HTTP {code}: {text}"),
48 
                    }
49 
                }
50 
                Err(e) => bail!("{url}: {e}"),
51 
            }
52 
        }
53 
    }
54
55 
    pub fn claim(&self, req: &ClaimRequest) -> Result<Option<Claim>> {
56 
        match self.post("/flow/claim", req)? {
57 
            Some(r) => Ok(Some(r.into_json()?)),
58 
            None => Ok(None),
59 
        }
60 
    }
61
62 
    pub fn log(&self, job: i64, records: Vec<Record>) -> Result<()> {
63 
        self.post(&format!("/flow/jobs/{job}/log"), &LogBatch { records })?;
64 
        Ok(())
65 
    }
66
67 
    pub fn finish(&self, job: i64, f: &Finish) -> Result<()> {
68 
        self.post(&format!("/flow/jobs/{job}/finish"), f)?;
69 
        Ok(())
70 
    }
71 
}
72
73 
/// `flow runner register`: exchange a one-time registration token from
74 
/// the hub for this runner's credential, binding the runner's signing key,
75 
/// and keep hub, credential and name in runner.toml.
76 
pub fn register(home: &Home, hub: &str, token: &str, name: &str, ca: Option<PathBuf>, labels: Vec<String>) -> Result<Registered> {
77 
    if home.config.runner.is_some() && !home.root.join(crate::home::RUNNER_TOML).is_file() {
78 
        bail!("flow.toml already has a [runner] section; remove it to register this runner instead");
79 
    }
80 
    let key = home.runner_key(name)?;
81 
    let hub = hub.trim_end_matches('/').to_string();
82 
    let tls = oxy_core::protocol::tls_client_config(ca.as_deref()).map_err(|e| anyhow::anyhow!("{e}"))?;
83 
    let agent = ureq::AgentBuilder::new().tls_config(tls).timeout(Duration::from_secs(60)).build();
84 
    let req = Register {
85 
        token: token.trim().to_string(),
86 
        name: name.to_string(),
87 
        identity: hex::encode(key.identity.encode()),
88 
        info: RunnerInfo::here(labels.clone(), 1, None, None),
89 
    };
90 
    let url = format!("{hub}/api/v1/flow/runners/register");
91 
    let got: Registered = match agent.post(&url).send_json(&req) {
92 
        Ok(r) => r.into_json()?,
93 
        Err(ureq::Error::Status(code, r)) => bail!("{url}: HTTP {code}: {}", r.into_string().unwrap_or_default()),
94 
        Err(e) => bail!("{url}: {e}"),
95 
    };
96 
    let mut text = format!(
97 
        "# Written by `flow runner register`; the credential is secret.\nhub = {hub:?}\ntoken = {:?}\nname = {:?}\n",
98 
        got.credential, got.name
99 
    );
100 
    if let Some(ca) = &ca {
101 
        text.push_str(&format!("ca = {:?}\n", ca.display().to_string()));
102 
    }
103 
    if !labels.is_empty() {
104 
        text.push_str(&format!("labels = {labels:?}\n"));
105 
    }
106 
    let path = home.root.join(crate::home::RUNNER_TOML);
107 
    {
108 
        use std::io::Write;
109 
        use std::os::unix::fs::OpenOptionsExt;
110 
        let tmp = path.with_extension("toml.tmp");
111 
        let mut f = std::fs::OpenOptions::new().write(true).create(true).truncate(true).mode(0o600).open(&tmp)?;
112 
        f.write_all(text.as_bytes())?;
113 
        f.sync_all()?;
114 
        std::fs::rename(&tmp, &path)?;
115 
    }
116 
    Ok(got)
117 
}
118
119 
/// Waits before trying a hub request again after a server error.
120 
const RETRY_DELAYS: [Duration; 3] = [Duration::from_secs(1), Duration::from_secs(2), Duration::from_secs(4)];
121
122 
/// What the runner may run at once (flow.toml `[runner]`): at most
123 
/// `max_jobs` jobs, whose sandboxes together fit `memory_mib` and `cpus`.
124 
#[derive(Debug, Default, PartialEq, Eq)]
125 
pub struct Budget {
126 
    pub max_jobs: usize,
127 
    pub memory_mib: Option<u32>,
128 
    pub cpus: Option<u32>,
129 
}
130
131 
/// What is in use now.
132 
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
133 
pub struct Used {
134 
    pub jobs: usize,
135 
    pub memory_mib: u32,
136 
    pub cpus: u32,
137 
}
138
139 
impl Budget {
140 
    pub fn of(cfg: &RunnerConfig) -> Budget {
141 
        Budget { max_jobs: cfg.max_jobs.unwrap_or(1).max(1), memory_mib: cfg.memory_mib, cpus: cfg.cpus }
142 
    }
143
144 
    /// The room to ask the hub for, or `None` when no job more can start.
145 
    /// A lone job may use the whole budget even if it asks for more, so a
146 
    /// large sandbox is never starved.
147 
    pub fn room(&self, used: Used) -> Option<(Option<u32>, Option<u32>)> {
148 
        if used.jobs >= self.max_jobs {
149 
            return None;
150 
        }
151 
        if used.jobs == 0 {
152 
            return Some((None, None));
153 
        }
154 
        let mem = self.memory_mib.map(|m| m.saturating_sub(used.memory_mib));
155 
        let cpus = self.cpus.map(|c| c.saturating_sub(used.cpus));
156 
        if mem == Some(0) || cpus == Some(0) {
157 
            return None;
158 
        }
159 
        Some((mem, cpus))
160 
    }
161 
}
162
163 
pub fn serve(home: &Home) -> Result<()> {
164 
    let cfg = home.config.runner.clone().context("flow.toml has no [runner] section")?;
165 
    let hub = Hub::new(&cfg)?;
166 
    let name = cfg.name.clone().unwrap_or_else(|| "flow".into());
167 
    let budget = Budget::of(&cfg);
168 
    let key = home.runner_key(&name)?;
169 
    eprintln!(
170 
        "flow runner {name}: polling {} (up to {} jobs at once), signing receipts as {}",
171 
        cfg.hub,
172 
        budget.max_jobs,
173 
        key.identity.hash().to_hex()
174 
    );
175 
    let used = std::sync::Mutex::new(Used::default());
176 
    let mut idle = Duration::from_secs(2);
177 
    std::thread::scope(|scope| loop {
178 
        let room = budget.room(*used.lock().unwrap());
179 
        let Some((free_memory_mib, free_cpus)) = room else {
180 
            std::thread::sleep(Duration::from_millis(500));
181 
            continue;
182 
        };
183 
        let req = ClaimRequest {
184 
            runner: name.clone(),
185 
            runtimes: vec!["microvm".into()],
186 
            free_memory_mib,
187 
            free_cpus,
188 
            info: Some(RunnerInfo::here(cfg.labels.clone(), budget.max_jobs as u32, cfg.memory_mib, cfg.cpus)),
189 
        };
190 
        match hub.claim(&req) {
191 
            Ok(Some(claim)) => {
192 
                idle = Duration::from_secs(2);
193 
                let (mem, cpus) = (claim.sandbox.memory_mib, claim.sandbox.cpus);
194 
                {
195 
                    let mut u = used.lock().unwrap();
196 
                    u.jobs += 1;
197 
                    u.memory_mib += mem;
198 
                    u.cpus += cpus;
199 
                }
200 
                let (home, hub, cfg, used, key) = (home, &hub, &cfg, &used, &key);
201 
                scope.spawn(move || {
202 
                    let id = claim.job_id;
203 
                    eprintln!("job {id}: {} {} on {}", claim.repo, claim.job.name, claim.heads.first().map(|h| &h[..12]).unwrap_or("-"));
204 
                    let finish = run_claim(home, hub, cfg, key, &claim).unwrap_or_else(|e| Finish {
205 
                        exit_code: None,
206 
                        duration_ms: 0,
207 
                        receipt: None,
208 
                        signed_receipt: None,
209 
                        error: Some(format!("{e:#}")),
210 
                    });
211 
                    eprintln!(
212 
                        "job {id}: exit {:?}{}",
213 
                        finish.exit_code,
214 
                        finish.error.as_deref().map(|e| format!(" ({e})")).unwrap_or_default()
215 
                    );
216 
                    if let Err(e) = hub.finish(id, &finish) {
217 
                        eprintln!("job {id}: reporting failed: {e:#}");
218 
                    }
219 
                    let mut u = used.lock().unwrap();
220 
                    u.jobs -= 1;
221 
                    u.memory_mib -= mem;
222 
                    u.cpus -= cpus;
223 
                });
224 
            }
225 
            Ok(None) => std::thread::sleep(idle),
226 
            Err(e) => {
227 
                eprintln!("claim: {e:#}");
228 
                std::thread::sleep(idle);
229 
                idle = (idle * 2).min(Duration::from_secs(60));
230 
            }
231 
        }
232 
    })
233 
}
234
235 
/// Batches records to the hub every second (or every 200 lines).
236 
struct LogSink<'a> {
237 
    hub: &'a Hub,
238 
    job: i64,
239 
    buf: Vec<Record>,
240 
    last: Instant,
241 
}
242
243 
impl LogSink<'_> {
244 
    fn push(&mut self, mut r: Record) {
245 
        if r.text.len() > MAX_RECORD_TEXT {
246 
            let mut cut = MAX_RECORD_TEXT;
247 
            while !r.text.is_char_boundary(cut) {
248 
                cut -= 1;
249 
            }
250 
            r.text.truncate(cut);
251 
        }
252 
        self.buf.push(r);
253 
        if self.buf.len() >= 200 || self.last.elapsed() >= Duration::from_secs(1) {
254 
            self.flush();
255 
        }
256 
    }
257
258 
    fn flush(&mut self) {
259 
        if !self.buf.is_empty() {
260 
            if let Err(e) = self.hub.log(self.job, std::mem::take(&mut self.buf)) {
261 
                eprintln!("job {}: log: {e:#}", self.job);
262 
            }
263 
        }
264 
        self.last = Instant::now();
265 
    }
266 
}
267
268 
/// Held while a job's workspace is prepared (see `run_claim`).
269 
static PREPARE: std::sync::Mutex<()> = std::sync::Mutex::new(());
270
271 
fn run_claim(home: &Home, hub: &Hub, cfg: &RunnerConfig, key: &RunnerKey, claim: &Claim) -> Result<Finish> {
272 
    let started = Instant::now();
273 
    let mut sink = LogSink { hub, job: claim.job_id, buf: Vec::new(), last: Instant::now() };
274 
    let note =
275 
        |sink: &mut LogSink, text: String| sink.push(Record { t: started.elapsed().as_millis() as u64, stream: "flow".into(), text });
276 
    let heads: Vec<ChangeHash> = claim.heads.iter().map(|h| ChangeHash::from_hex(h).context("bad head")).collect::<Result<_>>()?;
277 
    note(&mut sink, format!("fetching {} for state {}", claim.repo, claim.heads.iter().map(|h| &h[..12]).collect::<Vec<_>>().join("+")));
278 
    // Preparing a workspace (the cached repositories, the export, the
279 
    // image) is one job at a time: fetches into a cached repository must not
280 
    // overlap, and the first clone passes its token through the process
281 
    // environment. Only the microVMs run side by side.
282 
    let mut preparing = Some(PREPARE.lock().unwrap_or_else(|p| p.into_inner()));
283 
    let repo = cached_repo(home, cfg, claim)?;
284 
    let scratch = home.root.join("work").join(format!("job-{}", claim.job_id));
285 
    let _ = std::fs::remove_dir_all(&scratch);
286 
    std::fs::create_dir_all(scratch.join("tree"))?;
287 
    let result = (|| -> Result<Finish> {
288 
        let files = workspace::export_state(&repo, &heads, &scratch.join("tree"))?;
289 
        let scope_hash = scope_tree_hash(&claim.job.scope, files.iter().map(|(p, h)| (p.as_str(), *h)));
290 
        workspace::image(&scratch.join("tree"), &scratch.join("workspace.ext4"))?;
291 
        std::fs::remove_dir_all(scratch.join("tree"))?;
292 
        preparing.take();
293 
        note(
294 
            &mut sink,
295 
            format!(
296 
                "booting a microVM: {} vCPU, {} MiB, rootfs {}",
297 
                claim.sandbox.cpus,
298 
                claim.sandbox.memory_mib,
299 
                &claim.sandbox.rootfs[..12]
300 
            ),
301 
        );
302 
        let outcome = microvm::run(
303 
            home,
304 
            &claim.sandbox,
305 
            &claim.job,
306 
            &scratch.join("workspace.ext4"),
307 
            &scratch,
308 
            Duration::from_secs(claim.timeout_s),
309 
            &mut |r| sink.push(r),
310 
        )?;
311 
        let receipt = outcome.exit_code.map(|code| Receipt {
312 
            job_name: claim.job.name.clone(),
313 
            pipeline_hash: hex32(&claim.pipeline_hash),
314 
            scope_tree_hash: scope_hash,
315 
            inputs: vec![],
316 
            environment: outcome.environment.hash(),
317 
            outputs: vec![],
318 
            exit_code: code as u32,
319 
            duration_ms: outcome.duration_ms,
320 
        });
321 
        let signed_receipt = match &receipt {
322 
            Some(r) => Some(hex::encode(sign_receipt(&repo, &heads, cfg, key, claim, r)?.encode())),
323 
            None => None,
324 
        };
325 
        Ok(Finish {
326 
            exit_code: outcome.exit_code,
327 
            duration_ms: started.elapsed().as_millis() as u64,
328 
            receipt,
329 
            signed_receipt,
330 
            error: outcome.timed_out.then(|| "stopped at the time limit".to_string()),
331 
        })
332 
    })();
333 
    let _ = std::fs::remove_dir_all(&scratch);
334 
    sink.flush();
335 
    result
336 
}
337
338 
/// Bind `receipt` to the state of `heads` in `repo` and sign it as the
339 
/// runner (Oxygen spec §14.6). The state is computed here, not taken from
340 
/// the claim, so the runner vouches only for what it fetched.
341 
fn sign_receipt(
342 
    repo: &Repo,
343 
    heads: &[ChangeHash],
344 
    cfg: &RunnerConfig,
345 
    key: &RunnerKey,
346 
    claim: &Claim,
347 
    receipt: &Receipt,
348 
) -> Result<oxy_core::receipt::SignedReceipt> {
349 
    let state = state_of(heads);
350 
    let run = format!("{}/{}/flow/{}", cfg.hub.trim_end_matches('/'), claim.repo, claim.run_number);
351 
    Ok(oxy_core::receipt::SignedReceipt::sign(receipt.to_oxy(), repo.id()?, state, &run, &key.identity, &key.secret))
352 
}
353
354 
/// The ObjectRef of the state whose heads are `heads` (Oxygen spec §10.1).
355 
pub fn state_of(heads: &[ChangeHash]) -> oxy_core::hash::ObjectRef {
356 
    use oxy_core::hash::{hash_object, ObjectType};
357 
    hash_object(ObjectType::State, &oxy_core::objects::StateObject::new(heads.to_vec()).encode())
358 
}
359
360 
fn hex32(s: &str) -> [u8; 32] {
361 
    let mut out = [0; 32];
362 
    if let Ok(v) = hex::decode(s) {
363 
        if v.len() == 32 {
364 
            out.copy_from_slice(&v);
365 
        }
366 
    }
367 
    out
368 
}
369
370 
/// The runner's copy of a repository, fetched up to date with the job's token.
371 
fn cached_repo(home: &Home, cfg: &RunnerConfig, claim: &Claim) -> Result<Repo> {
372 
    let dir: PathBuf = home.root.join("repos").join(claim.repo.replace('/', "__"));
373 
    let set_remote = |repo: &mut Repo| -> Result<()> {
374 
        repo.config.set("remote.origin.url", &claim.clone_url).map_err(|e| anyhow::anyhow!("{e}"))?;
375 
        repo.config.set("remote.origin.token", &claim.token).map_err(|e| anyhow::anyhow!("{e}"))?;
376 
        if let Some(ca) = &cfg.ca {
377 
            repo.config.set("remote.origin.ca", &ca.to_string_lossy()).map_err(|e| anyhow::anyhow!("{e}"))?;
378 
        }
379 
        repo.save_config().map_err(|e| anyhow::anyhow!("{e}"))
380 
    };
381 
    if dir.join(".oxy").is_dir() {
382 
        let mut repo = Repo::open(&dir).map_err(|e| anyhow::anyhow!("{e}"))?;
383 
        set_remote(&mut repo)?;
384 
        repo.fetch("origin").map_err(|e| anyhow::anyhow!("fetch {}: {e}", claim.clone_url))?;
385 
        return Ok(repo);
386 
    }
387 
    // First job for this repository: clone with the job's token.
388 
    std::env::set_var("OXY_TOKEN", &claim.token);
389 
    if let Some(ca) = &cfg.ca {
390 
        std::env::set_var("OXY_CA", ca);
391 
    }
392 
    let cloned = Repo::clone(&claim.clone_url, &dir);
393 
    std::env::remove_var("OXY_TOKEN");
394 
    std::env::remove_var("OXY_CA");
395 
    let mut repo = cloned.map_err(|e| {
396 
        let _ = std::fs::remove_dir_all(&dir);
397 
        anyhow::anyhow!("clone {}: {e}", claim.clone_url)
398 
    })?;
399 
    set_remote(&mut repo)?;
400 
    Ok(repo)
401 
}
402
403 
#[cfg(test)]
404 
mod tests {
405 
    use super::*;
406
407 
    /// A hub that answers each request with the next of `codes`, counting them.
408 
    fn fake_hub(codes: Vec<u16>) -> (String, std::sync::Arc<std::sync::atomic::AtomicUsize>) {
409 
        use std::io::{Read, Write};
410 
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
411 
        let addr = listener.local_addr().unwrap();
412 
        let seen = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
413 
        let count = seen.clone();
414 
        std::thread::spawn(move || {
415 
            for code in codes {
416 
                let (mut s, _) = listener.accept().unwrap();
417 
                let mut buf = [0u8; 8192];
418 
                let _ = s.read(&mut buf);
419 
                count.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
420 
                let reason = if code == 204 { "No Content" } else { "Error" };
421 
                let _ = write!(s, "HTTP/1.1 {code} {reason}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n");
422 
            }
423 
        });
424 
        (format!("http://{addr}"), seen)
425 
    }
426
427 
    /// A hub that answers one request with `body` (status 200) and hands
428 
    /// back what it was sent.
429 
    fn fake_json_hub(body: &'static str) -> (String, std::sync::mpsc::Receiver<String>) {
430 
        use std::io::{Read, Write};
431 
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
432 
        let addr = listener.local_addr().unwrap();
433 
        let (tx, rx) = std::sync::mpsc::channel();
434 
        std::thread::spawn(move || {
435 
            let (mut s, _) = listener.accept().unwrap();
436 
            let mut req = Vec::new();
437 
            let mut buf = [0u8; 65536];
438 
            loop {
439 
                let n = s.read(&mut buf).unwrap_or(0);
440 
                if n == 0 {
441 
                    break;
442 
                }
443 
                req.extend_from_slice(&buf[..n]);
444 
                let text = String::from_utf8_lossy(&req);
445 
                if let Some(end) = text.find("\r\n\r\n") {
446 
                    let len = text[..end]
447 
                        .lines()
448 
                        .find_map(|l| {
449 
                            l.to_ascii_lowercase().strip_prefix("content-length:").map(|v| v.trim().parse::<usize>().unwrap_or(0))
450 
                        })
451 
                        .unwrap_or(0);
452 
                    if req.len() >= end + 4 + len {
453 
                        break;
454 
                    }
455 
                }
456 
            }
457 
            tx.send(String::from_utf8_lossy(&req).into_owned()).unwrap();
458 
            let _ = write!(
459 
                s,
460 
                "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
461 
                body.len()
462 
            );
463 
        });
464 
        (format!("http://{addr}"), rx)
465 
    }
466
467 
    #[test]
468 
    fn registering_binds_the_key_and_keeps_the_credential_private() {
469 
        use std::os::unix::fs::PermissionsExt;
470 
        let dir = tempfile::tempdir().unwrap();
471 
        let home = Home::locate(Some(dir.path())).unwrap();
472 
        let (hub, sent) = fake_json_hub(r#"{"id":7,"name":"plex","credential":"flowrunner_secret"}"#);
473 
        let got = register(&home, &hub, " reg-token ", "plex", None, vec!["gpu".into()]).unwrap();
474 
        assert_eq!((got.id, got.name.as_str()), (7, "plex"));
475 
        let req = sent.recv().unwrap();
476 
        assert!(req.starts_with("POST /api/v1/flow/runners/register"), "{req}");
477 
        let body: Register = serde_json::from_str(&req[req.find("\r\n\r\n").unwrap() + 4..]).unwrap();
478 
        assert_eq!((body.token.as_str(), body.info.labels.clone()), ("reg-token", vec!["gpu".to_string()]));
479 
        let key = home.runner_key("plex").unwrap();
480 
        assert_eq!(body.identity, hex::encode(key.identity.encode()), "the runner's signing identity");
481 
        let path = dir.path().join(crate::home::RUNNER_TOML);
482 
        assert_eq!(std::fs::metadata(&path).unwrap().permissions().mode() & 0o777, 0o600);
483 
        // The runner then starts from runner.toml.
484 
        let again = Home::locate(Some(dir.path())).unwrap();
485 
        let cfg = again.config.runner.unwrap();
486 
        assert_eq!((cfg.hub.as_str(), cfg.token.as_str(), cfg.name.as_deref()), (hub.as_str(), "flowrunner_secret", Some("plex")));
487 
        assert_eq!(cfg.labels, ["gpu"]);
488 
    }
489
490 
    fn hub_at(base: String) -> Hub {
491 
        Hub { base, token: "t".into(), agent: ureq::AgentBuilder::new().timeout(Duration::from_secs(5)).build() }
492 
    }
493
494 
    #[test]
495 
    fn server_errors_are_tried_again_and_client_errors_are_not() {
496 
        let quick = [Duration::from_millis(10); 3];
497 
        let (base, seen) = fake_hub(vec![500, 503, 204]);
498 
        let r = hub_at(base).post_with("/flow/jobs/1/log", &serde_json::json!({}), &quick).unwrap();
499 
        assert!(r.is_none());
500 
        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 3, "two retries, then accepted");
501
502 
        let (base, seen) = fake_hub(vec![500, 500, 500, 500]);
503 
        let e = hub_at(base).post_with("/flow/claim", &serde_json::json!({}), &quick).unwrap_err().to_string();
504 
        assert!(e.contains("HTTP 500"), "{e}");
505 
        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 4, "gives up after the last delay");
506
507 
        let (base, seen) = fake_hub(vec![401]);
508 
        assert!(hub_at(base).post_with("/flow/claim", &serde_json::json!({}), &quick).is_err());
509 
        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 1, "a refusal is final");
510 
    }
511
512 
    #[test]
513 
    fn the_budget_admits_jobs_that_fit() {
514 
        let b = Budget { max_jobs: 3, memory_mib: Some(24_576), cpus: Some(8) };
515 
        let used = |jobs, memory_mib, cpus| Used { jobs, memory_mib, cpus };
516 
        assert_eq!(b.room(used(0, 0, 0)), Some((None, None)), "a lone job may take everything");
517 
        assert_eq!(b.room(used(1, 8192, 4)), Some((Some(16_384), Some(4))));
518 
        assert_eq!(b.room(used(2, 16_384, 8)), None, "no vCPU left");
519 
        assert_eq!(b.room(used(3, 1024, 1)), None, "at max_jobs");
520 
        let one = Budget { max_jobs: 1, memory_mib: None, cpus: None };
521 
        assert_eq!(one.room(used(0, 0, 0)), Some((None, None)));
522 
        assert_eq!(one.room(used(1, 0, 0)), None, "the default runs one job at a time");
523 
        let open = Budget { max_jobs: 2, memory_mib: None, cpus: None };
524 
        assert_eq!(open.room(used(1, 8192, 4)), Some((None, None)), "no limits configured");
525 
    }
526 
}