pipeline.kdl: timeout on a job or a sandbox 02b127d2c721
timeout "90m" (hours, minutes, seconds, largest first; 1m to 24h) on a job, or on a sandbox as its jobs' default, compiled into the job as timeout_s; a job without one compiles and hashes as before. The hub takes it, capped at the site's maximum. Spec and tests. closes #23
- Author
- Andy McCall @andymccall <[email protected]>
- When
- · 2026-10-02 21:08 UTC
- Signature
- ✔ signed by Andy McCall
- Hash
02b127d2c721d85170666bf87f934bb9f927755a853ebb319ddb2f569d8f353c- Depends on
- 1daca0b6bb8c a083cc802892 a67b97e9a717 a72abeeedd78 c3db750ed96e e0918b887850 e576e08617dc fc10d9e80383
- Depended on by
- da5118d25db2 e87c1c138187 ea81796bb757 1abadd4a2fa6 c58efc6a74db bf7bb1b80fa3
- In bookmarks
- main newer-pipeline push-uses-login
- In tags
- v0.2.0 v0.3.0 v0.3.1 v0.4.0 v0.5.0 v0.5.1 v0.5.2 v0.5.3
- Edits
- 14 atoms, 0 tree operations, 5 files
Browse the tree at this change
modified CHANGELOG.md +4 −0
| … | 3 unchanged lines |
||
| 4 | 4 | ||
| 5 | 5 | ## Unreleased |
|
| 6 | 6 | ||
| 7 | + | - **A job says how long it may run (#23).** `timeout "90m"` on a job, or on a sandbox for its jobs (hours, minutes, seconds; a minute to a day). The hub caps it at the site's maximum. |
|
| 8 | + | ||
| 9 | + | - **A pinned file is re-read from the disk before its hash is called wrong (#22).** |
|
| 10 | + | ||
| 7 | 11 | - **A runner says which jobs it holds (oxygenhub#71).** Each claim lists the jobs the runner is running or still has a result for, so the hub ends at once any job it lost by restarting, instead of an hour later at the time limit. |
|
| 8 | 12 | - **A runner says which images it holds (oxygenhub#73).** Each claim lists the rootfs images in the runner's `cas/` (read each time, so an image imported while it runs counts), and the hub offers only jobs whose image is among them. plex-1 had failed metalgear's jobs for want of one. |
|
| 9 | 13 | ||
| … | 4 unchanged lines |
modified crates/flow-core/src/pipeline.rs +62 −1
| … | 72 unchanged lines |
||
| 73 | 73 | /// Labels a runner must have to run jobs here (`runs-on`), sorted. |
|
| 74 | 74 | #[serde(default, skip_serializing_if = "Vec::is_empty")] |
|
| 75 | 75 | pub runs_on: Vec<String>, |
|
| 76 | + | /// How long each job here may run, in seconds (`timeout "2h"`), when |
|
| 77 | + | /// the job does not say; the hub caps it at the site's maximum. |
|
| 78 | + | #[serde(default, skip_serializing_if = "Option::is_none")] |
|
| 79 | + | pub timeout_s: Option<u64>, |
|
| 76 | 80 | } |
|
| 77 | 81 | ||
| 78 | 82 | #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] |
|
| … | 17 unchanged lines |
||
| 96 | 100 | /// with its sandbox's, what the hub matches against a runner's. |
|
| 97 | 101 | #[serde(default, skip_serializing_if = "Vec::is_empty")] |
|
| 98 | 102 | pub runs_on: Vec<String>, |
|
| 103 | + | /// How long the job may run, in seconds (`timeout "90m"`); its |
|
| 104 | + | /// sandbox's when absent, else the site's default (oxygenflow#23). |
|
| 105 | + | #[serde(default, skip_serializing_if = "Option::is_none")] |
|
| 106 | + | pub timeout_s: Option<u64>, |
|
| 99 | 107 | } |
|
| 100 | 108 | ||
| 101 | 109 | impl Job { |
|
| … | 6 unchanged lines |
||
| 108 | 116 | job.runs_on.extend(sandbox.runs_on.iter().cloned()); |
|
| 109 | 117 | job.runs_on.sort(); |
|
| 110 | 118 | job.runs_on.dedup(); |
|
| 119 | + | job.timeout_s = job.timeout_s.or(sandbox.timeout_s); |
|
| 111 | 120 | job |
|
| 112 | 121 | } |
|
| 113 | 122 | ||
| … | 404 unchanged lines |
||
| 518 | 527 | let mut seen = BTreeSet::new(); |
|
| 519 | 528 | let mut env = BTreeMap::new(); |
|
| 520 | 529 | let mut runs_on = Vec::new(); |
|
| 530 | + | let mut timeout_s = None; |
|
| 521 | 531 | for c in block(ctx, node) { |
|
| 522 | 532 | let key = c.name().value(); |
|
| 523 | 533 | if key != "runs-on" && !seen.insert(key.to_string()) { |
|
| … | 82 unchanged lines |
||
| 606 | 616 | None => None, |
|
| 607 | 617 | } |
|
| 608 | 618 | } |
|
| 619 | + | "timeout" => timeout_s = parse_timeout(ctx, c), |
|
| 609 | 620 | other => ctx.err(c.span(), format!("unknown sandbox setting `{other}`")), |
|
| 610 | 621 | } |
|
| 611 | 622 | } |
|
| … | 3 unchanged lines |
||
| 615 | 626 | if !named { |
|
| 616 | 627 | return None; |
|
| 617 | 628 | } |
|
| 618 | - | Some(Sandbox { name, runtime: runtime?, rootfs: rootfs?, memory_mib: memory?, cpus: cpus?, network: network?, env, runs_on }) |
|
| 629 | + | Some(Sandbox { name, runtime: runtime?, rootfs: rootfs?, memory_mib: memory?, cpus: cpus?, network: network?, env, runs_on, timeout_s }) |
|
| 619 | 630 | } |
|
| 620 | 631 | ||
| 632 | + | /// The longest a pipeline may ask a job to run. |
|
| 633 | + | pub const MAX_TIMEOUT_S: u64 = 24 * 3600; |
|
| 634 | + | ||
| 635 | + | /// A duration: whole hours, minutes and seconds in that order, each |
|
| 636 | + | /// optional, as "90m", "2h", "1h30m" or "45s". |
|
| 637 | + | pub fn parse_duration(s: &str) -> Option<u64> { |
|
| 638 | + | let (mut total, mut num, mut last) = (0u64, String::new(), 0u8); |
|
| 639 | + | for ch in s.chars() { |
|
| 640 | + | if ch.is_ascii_digit() { |
|
| 641 | + | num.push(ch); |
|
| 642 | + | continue; |
|
| 643 | + | } |
|
| 644 | + | let (unit, mult) = match ch { |
|
| 645 | + | 'h' => (1, 3600), |
|
| 646 | + | 'm' => (2, 60), |
|
| 647 | + | 's' => (3, 1), |
|
| 648 | + | _ => return None, |
|
| 649 | + | }; |
|
| 650 | + | // Each unit once, largest first, each with its number. |
|
| 651 | + | if num.is_empty() || unit <= last { |
|
| 652 | + | return None; |
|
| 653 | + | } |
|
| 654 | + | last = unit; |
|
| 655 | + | total = total.checked_add(num.parse::<u64>().ok()?.checked_mul(mult)?)?; |
|
| 656 | + | num.clear(); |
|
| 657 | + | } |
|
| 658 | + | (num.is_empty() && total > 0).then_some(total) |
|
| 659 | + | } |
|
| 660 | + | ||
| 661 | + | /// A `timeout` line: one duration, from a minute to a day. |
|
| 662 | + | fn parse_timeout(ctx: &mut Ctx, c: &KdlNode) -> Option<u64> { |
|
| 663 | + | let t = ctx.one_string(c)?; |
|
| 664 | + | match parse_duration(&t) { |
|
| 665 | + | Some(secs) if (60..=MAX_TIMEOUT_S).contains(&secs) => Some(secs), |
|
| 666 | + | Some(_) => { |
|
| 667 | + | ctx.err(c.span(), format!("timeout {t:?} must be from 1m to 24h")); |
|
| 668 | + | None |
|
| 669 | + | } |
|
| 670 | + | None => { |
|
| 671 | + | ctx.err(c.span(), format!("timeout {t:?}: expected hours, minutes and seconds, as \"90m\" or \"1h30m\"")); |
|
| 672 | + | None |
|
| 673 | + | } |
|
| 674 | + | } |
|
| 675 | + | } |
|
| 676 | + | ||
| 621 | 677 | /// The labels of a `runs-on` line: lower-case letters, digits, `-`, `_` |
|
| 622 | 678 | /// and `.`, at most 64 bytes each; into `into`, sorted, once each. |
|
| 623 | 679 | fn parse_labels(ctx: &mut Ctx, c: &KdlNode, into: &mut Vec<String>) { |
|
| … | 165 unchanged lines |
||
| 789 | 845 | mounts: vec![], |
|
| 790 | 846 | outputs: vec![], |
|
| 791 | 847 | runs_on: vec![], |
|
| 848 | + | timeout_s: None, |
|
| 792 | 849 | }; |
|
| 793 | 850 | let mut seen = BTreeSet::new(); |
|
| 794 | 851 | let mut outputs_at = None; |
|
| … | 23 unchanged lines |
||
| 818 | 875 | "runs-on" => { |
|
| 819 | 876 | ctx.no_children(c); |
|
| 820 | 877 | parse_labels(ctx, c, &mut job.runs_on); |
|
| 878 | + | } |
|
| 879 | + | "timeout" => { |
|
| 880 | + | ctx.no_children(c); |
|
| 881 | + | job.timeout_s = parse_timeout(ctx, c); |
|
| 821 | 882 | } |
|
| 822 | 883 | "env" => parse_env(ctx, c, &mut job.env), |
|
| 823 | 884 | "exec" => { |
|
| … | 60 unchanged lines |
modified crates/flow-core/src/tests.rs +26 −0
| … | 386 unchanged lines |
||
| 387 | 387 | let none = problems(&doc("job \"y\" {\n sandbox \"alpine\"\n runs-on\n exec \"true\"\n}\n")); |
|
| 388 | 388 | assert!(none.iter().any(|(_, m)| m.contains("needs one or more labels")), "{none:?}"); |
|
| 389 | 389 | } |
|
| 390 | + | ||
| 391 | + | /// oxygenflow#23: `timeout` on a job, or on its sandbox for every job in |
|
| 392 | + | /// it, as a duration from a minute to a day. |
|
| 393 | + | #[test] |
|
| 394 | + | fn a_job_says_how_long_it_may_run() { |
|
| 395 | + | use crate::pipeline::{parse_duration, Pipeline}; |
|
| 396 | + | assert_eq!(parse_duration("90m"), Some(5400)); |
|
| 397 | + | assert_eq!(parse_duration("1h30m"), Some(5400)); |
|
| 398 | + | assert_eq!(parse_duration("2h"), Some(7200)); |
|
| 399 | + | assert_eq!(parse_duration("45s"), Some(45)); |
|
| 400 | + | for bad in ["", "90", "m", "30m1h", "1h1h", "2d", "-1h"] { |
|
| 401 | + | assert_eq!(parse_duration(bad), None, "{bad:?}"); |
|
| 402 | + | } |
|
| 403 | + | let root = "a".repeat(64); |
|
| 404 | + | let text = format!( |
|
| 405 | + | "schema-version 1\nsandbox \"s\" {{\n runtime \"microvm\"\n rootfs \"cas://blob/{root}\"\n memory-limit \"1GiB\"\n cpu-limit 1\n network \"isolated\"\n timeout \"2h\"\n}}\njob \"long\" {{\n sandbox \"s\"\n timeout \"3h30m\"\n exec \"true\"\n}}\njob \"plain\" {{\n sandbox \"s\"\n exec \"true\"\n}}\n" |
|
| 406 | + | ); |
|
| 407 | + | let p = Pipeline::parse(&text).unwrap(); |
|
| 408 | + | let job = |n: &str| p.jobs[n].in_sandbox(&p.sandboxes["s"]); |
|
| 409 | + | assert_eq!(job("long").timeout_s, Some(12600), "the job's own"); |
|
| 410 | + | assert_eq!(job("plain").timeout_s, Some(7200), "the sandbox's"); |
|
| 411 | + | let bad = text.replace("timeout \"3h30m\"", "timeout \"2d\""); |
|
| 412 | + | assert!(Pipeline::parse(&bad).unwrap_err().to_string().contains("timeout")); |
|
| 413 | + | let long = text.replace("timeout \"3h30m\"", "timeout \"25h\""); |
|
| 414 | + | assert!(Pipeline::parse(&long).unwrap_err().to_string().contains("from 1m to 24h")); |
|
| 415 | + | } |
modified crates/flow/src/outputs.rs +1 −0
| … | 75 unchanged lines |
||
| 76 | 76 | ], |
|
| 77 | 77 | outputs: vec![], |
|
| 78 | 78 | runs_on: vec![], |
|
| 79 | + | timeout_s: None, |
|
| 79 | 80 | } |
|
| 80 | 81 | } |
|
| 81 | 82 | ||
| … | 24 unchanged lines |
modified docs/spec.md +2 −0
| … | 160 unchanged lines |
||
| 161 | 161 | } |
|
| 162 | 162 | ||
| 163 | 163 | ||
| 164 | + | **Time limits.** A job may say how long it may run with `timeout "90m"` (hours, minutes and seconds, as `"2h"`, `"1h30m"` or `"45s"`, from a minute to a day); a sandbox's `timeout` is the default for its jobs. Without one, the hub's default applies, and the hub caps every job at the site's maximum (both set by an administrator on Admin → Runners). The runner stops a job at its limit, and the hub ends one its runner stopped reporting a few minutes after it. A timeout is part of the compiled job, so a job without one compiles as before. |
|
| 165 | + | ||
| 164 | 166 | **Runner labels.** A sandbox or a job may say `runs-on "LABEL" ...`: lower-case letters, digits, `-`, `_` and `.`, at most 64 bytes each, given on one or more lines. A job needs a runner that has every label its sandbox and it name. A runner reports its labels and architecture when it registers and on each claim (`flow runner register --label`). The hub hands a job only to a runner with all of them, within the runner's scope (OxygenHub spec 13.2), and a job no runner can take waits, saying so on the queue. Labels are part of the compiled job, so a job that asks for none compiles as before. |
|
| 165 | 167 | ||
| 166 | 168 | ### 3.2 Canonical Compilation |
|
| … | 248 unchanged lines |