OXYGENHUB

A leaving runner finishes what fits its notice (refs #76) 1b160d372d9e

A claim carries expected_ms when the hub has learned how long the job
takes; runner::leave_within(reason, notice) finishes and reports a job
whose expected rest fits the notice and hands back the others, and any
still running when it is up. runner::leave is the same with no notice.
Spec 11.5.2.
Author
Andy McCall @andymccall <[email protected]>
via
agent agent:claude@andymccall <[email protected]> · keyless, under your grant for changes,refs,seal · key b92e2f89aee3
When
· 2026-10-25 21:27 UTC
Signature
◈ sealed by agent:claude@andymccall
Hash
1b160d372d9ec16fa18a654b8fe78e7d39db4d21a76872d895dc863142f6f281
Depends on
06a4c472b385 280ef951d193 46fe2c815cc3 89171fca079c 9e5fd161efc3 a083cc802892 c3db750ed96e c4285535b03a e0918b887850
Depended on by
f201e1bc5bcd
In bookmarks
main
In tags
v0.4.0 v0.5.0 v0.5.1 v0.5.2 v0.5.3
Edits
22 atoms, 0 tree operations, 5 files

Browse the tree at this change

modified CHANGELOG.md +1 −0
…
3 unchanged lines
4 4
5 5
## Unreleased
6 6
7 +
- **A leaving runner finishes what fits its notice (#76):** a claim carries `expected_ms` (the median of the job's recent passing runs, from OxygenHub 0.9.4); `runner::leave_within(reason, notice)` finishes and reports a job whose expected rest fits the notice and hands back the others, and any still running when it is up. `runner::leave` is the same with no notice. Spec §11.5.2.
7 8
- **OxygenFlow's own CI writes a test report and coverage (#82):** its test job runs under cargo-nextest and cargo-llvm-cov (both in the Rust image, #81), keeping a JUnit report (whether or not the tests pass) and LCOV coverage for the hub's Tests and Coverage tabs; `.config/nextest.toml` has the `ci` profile.
8 9
- **A runner that leaves hands its jobs back (#76):** `runner::leave(reason)` (the runner as a library, as the Android app runs it) stops claiming, stops the jobs it holds, sends what they logged and hands them back with `POST /api/v1/flow/jobs/{id}/release`; the hub (OxygenHub #312) queues them again ahead of new work for another runner, and hands back the jobs of a runner silent for 3 minutes the same way. Spec §11.5.2.
9 10
- **A runner reads its load and kernel without /proc:** `sysinfo` and `uname`, the same numbers; an Android app may not read `/proc/loadavg` or `/proc/sys/kernel/osrelease`, so the phone runner was refused, and logged, at every claim (oxygenhub-android#8).
…
91 unchanged lines
modified crates/flow-core/src/api.rs +5 −0
…
284 unchanged lines
285 285
    /// `/api/v1/flow/jobs/{job_id}/inputs/{index}` and checked against it.
286 286
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
287 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>,
288 293
}
289 294
290 295
/// An artifact a job takes (oxygenflow#58), as the hub hands it over.
…
90 unchanged lines
modified crates/flow-core/src/tests.rs +8 −0
…
444 unchanged lines
445 445
        env: [("CC".to_string(), "clang".to_string()), ("FLOW_REPO".to_string(), "spoofed".to_string())].into_iter().collect(),
446 446
        secrets: vec![],
447 447
        inputs: vec![],
448 +
        expected_ms: Some(17_700),
448 449
    };
449 450
    let env = job.with_flow_env(&claim.flow_env()).env;
450 451
    assert_eq!(env["CC"], "clang", "the hub's variables reach the job");
…
14 unchanged lines
465 466
    };
466 467
    let names: Vec<String> = old.flow_env().into_iter().map(|(k, _)| k).collect();
467 468
    assert_eq!(names, ["FLOW_REPO", "FLOW_HEADS", "FLOW_JOB"]);
469 +
    // The expected time goes with the claim, and an older hub's claim
470 +
    // without it still reads (#76).
471 +
    let json = serde_json::to_value(&old).unwrap();
472 +
    assert_eq!(json["expected_ms"], 17_700);
473 +
    let mut without = json.clone();
474 +
    without.as_object_mut().unwrap().remove("expected_ms");
475 +
    assert_eq!(serde_json::from_value::<crate::api::Claim>(without).unwrap().expected_ms, None);
468 476
}
469 477
470 478
#[test]
…
259 unchanged lines
modified crates/flow/src/runner.rs +124 −19
…
590 unchanged lines
591 591
    serve_until_drained(home)
592 592
}
593 593
594 -
/// Why the runner is leaving, once [`leave`] says so: the jobs it holds are
595 -
/// stopped and handed back rather than finished (oxygenflow#76).
596 -
static LEAVING: std::sync::Mutex<Option<String>> = std::sync::Mutex::new(None);
594 +
/// A departure ([`leave_within`]): why, and when the runner must be gone.
595 +
#[derive(Clone, Debug)]
596 +
struct Leave {
597 +
    reason: String,
598 +
    by: Instant,
599 +
}
600 +
601 +
/// Set once the runner is leaving: a job whose expected rest fits before
602 +
/// the deadline finishes; the others are stopped and handed back
603 +
/// (oxygenflow#76).
604 +
static LEAVING: std::sync::Mutex<Option<Leave>> = std::sync::Mutex::new(None);
597 605
598 606
fn leaving() -> Option<String> {
599 -
    LEAVING.lock().unwrap_or_else(|p| p.into_inner()).clone()
607 +
    LEAVING.lock().unwrap_or_else(|p| p.into_inner()).as_ref().map(|l| l.reason.clone())
600 608
}
601 609
602 610
/// Ask a runner started with [`serve_embedded`] to leave now: claim nothing
…
1 unchanged lines
604 612
/// them ahead of new work for another runner (oxygenflow#76), then return.
605 613
/// For a departure with no time to finish: a phone off the allowed network.
606 614
pub fn leave(reason: &str) {
607 -
    *LEAVING.lock().unwrap_or_else(|p| p.into_inner()) = Some(reason.to_string());
615 +
    leave_within(reason, Duration::ZERO);
616 +
}
617 +
618 +
/// [`leave`], with `notice`: a job the hub expects to finish within it
619 +
/// (the claim's `expected_ms`, less the time it has run) finishes and is
620 +
/// reported; the others, and any still running when the notice is up, are
621 +
/// handed back. For a phone taken off charge, a spot reclaim.
622 +
pub fn leave_within(reason: &str, notice: Duration) {
623 +
    *LEAVING.lock().unwrap_or_else(|p| p.into_inner()) = Some(Leave { reason: reason.to_string(), by: Instant::now() + notice });
608 624
    stop();
609 625
}
610 626
…
236 unchanged lines
847 863
    last: Instant,
848 864
    gone: bool,
849 865
    /// Set when the runner leaves ([`LEAVING`]): the job stops, to be
850 -
    /// handed back.
851 -
    leaving: &'a std::sync::Mutex<Option<String>>,
866 +
    /// handed back, unless its expected rest fits the notice.
867 +
    leaving: &'a std::sync::Mutex<Option<Leave>>,
868 +
    /// When the job started, and how long the hub expects it to take.
869 +
    started: Instant,
870 +
    expected: Option<Duration>,
852 871
}
853 872
854 873
/// The most log lines sent at once, as a busy job's sink flushes them.
…
11 unchanged lines
866 885
    }
867 886
868 887
    fn cancelled(&mut self) -> bool {
869 -
        if self.leaving.lock().unwrap_or_else(|p| p.into_inner()).is_some() {
870 -
            // Handed back as the runner leaves: what the job said so far first.
871 -
            self.flush();
872 -
            return true;
888 +
        let leave = self.leaving.lock().unwrap_or_else(|p| p.into_inner()).clone();
889 +
        if let Some(l) = leave {
890 +
            // Finished if what is left of it fits the notice; else handed
891 +
            // back as the runner leaves, what the job said so far first.
892 +
            let now = Instant::now();
893 +
            let fits = self.expected.is_some_and(|e| now + e.saturating_sub(self.started.elapsed()) <= l.by);
894 +
            if !fits || now >= l.by {
895 +
                self.flush();
896 +
                return true;
897 +
            }
873 898
        }
874 899
        if !self.gone && self.last.elapsed() >= HEARTBEAT {
875 900
            self.send();
…
100 unchanged lines
976 1001
977 1002
fn run_claim(home: &Home, hub: &Hub, cfg: &RunnerConfig, key: &RunnerKey, claim: &Claim) -> Result<Finish> {
978 1003
    let started = Instant::now();
979 -
    let mut sink = LogSink { hub, job: claim.job_id, buf: Vec::new(), last: Instant::now(), gone: false, leaving: &LEAVING };
1004 +
    let mut sink = LogSink {
1005 +
        hub,
1006 +
        job: claim.job_id,
1007 +
        buf: Vec::new(),
1008 +
        last: Instant::now(),
1009 +
        gone: false,
1010 +
        leaving: &LEAVING,
1011 +
        started: Instant::now(),
1012 +
        expected: claim.expected_ms.map(Duration::from_millis),
1013 +
    };
980 1014
    let note =
981 1015
        |sink: &mut LogSink, text: String| sink.push(Record { t: started.elapsed().as_millis() as u64, stream: "flow".into(), text });
982 1016
    let heads: Vec<ChangeHash> = claim.heads.iter().map(|h| ChangeHash::from_hex(h).context("bad head")).collect::<Result<_>>()?;
…
752 unchanged lines
1735 1769
        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 1, "a refusal is final");
1736 1770
    }
1737 1771
1738 -
    static NOT_LEAVING: std::sync::Mutex<Option<String>> = std::sync::Mutex::new(None);
1772 +
    static NOT_LEAVING: std::sync::Mutex<Option<Leave>> = std::sync::Mutex::new(None);
1739 1773
1740 1774
    /// oxygenflow#76: once the runner leaves, a job's sink says it is
1741 1775
    /// cancelled, having sent what the job said so far, and the job is
…
4 unchanged lines
1746 1780
        let leaving = std::sync::Mutex::new(None);
1747 1781
        let (base, seen) = fake_hub(vec![204, 204]);
1748 1782
        let hub = hub_at(base);
1749 -
        let mut sink = LogSink { hub: &hub, job: 7, buf: Vec::new(), last: Instant::now(), gone: false, leaving: &leaving };
1783 +
        let mut sink = LogSink {
1784 +
            hub: &hub,
1785 +
            job: 7,
1786 +
            buf: Vec::new(),
1787 +
            last: Instant::now(),
1788 +
            gone: false,
1789 +
            leaving: &leaving,
1790 +
            started: Instant::now(),
1791 +
            expected: None,
1792 +
        };
1750 1793
        sink.buf.push(Record { t: 1, stream: "out".into(), text: "step 1".into() });
1751 1794
        assert!(!sink.cancelled());
1752 -
        *leaving.lock().unwrap() = Some("off Wi-Fi".into());
1795 +
        *leaving.lock().unwrap() = Some(Leave { reason: "off Wi-Fi".into(), by: Instant::now() });
1753 1796
        assert!(sink.cancelled());
1754 1797
        assert!(sink.buf.is_empty(), "flushed before the job stops");
1755 1798
        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 1);
1756 1799
        hub.release(7, "off Wi-Fi").unwrap();
1757 1800
        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 2);
1801 +
    }
1802 +
1803 +
    /// oxygenflow#76: with notice, a job the hub expects to finish within
1804 +
    /// it runs on; one that will not, or whose time is unknown, is handed
1805 +
    /// back at once, and any job still running when the notice is up.
1806 +
    #[test]
1807 +
    fn a_job_that_fits_the_notice_finishes_and_the_others_go_back() {
1808 +
        use microvm::Sink;
1809 +
        let (base, _) = fake_hub(vec![204; 8]);
1810 +
        let hub = hub_at(base);
1811 +
        let leaving = std::sync::Mutex::new(Some(Leave { reason: "off charge".into(), by: Instant::now() + Duration::from_secs(60) }));
1812 +
        let sink = |expected: Option<u64>, ran: u64| LogSink {
1813 +
            hub: &hub,
1814 +
            job: 1,
1815 +
            buf: Vec::new(),
1816 +
            last: Instant::now(),
1817 +
            gone: false,
1818 +
            leaving: &leaving,
1819 +
            started: Instant::now() - Duration::from_secs(ran),
1820 +
            expected: expected.map(Duration::from_secs),
1821 +
        };
1822 +
        assert!(!sink(Some(70), 30).cancelled(), "40 s left of it, within a minute: it finishes");
1823 +
        assert!(sink(Some(600), 30).cancelled(), "nearly ten minutes left: handed back");
1824 +
        assert!(sink(None, 0).cancelled(), "no idea how long: handed back");
1825 +
        *leaving.lock().unwrap() = Some(Leave { reason: "off charge".into(), by: Instant::now() - Duration::from_secs(1) });
1826 +
        assert!(sink(Some(70), 30).cancelled(), "the notice is up: handed back whatever was expected");
1758 1827
    }
1759 1828
1760 1829
    /// oxygenflow#32: a quiet job asks the hub every few seconds whether
…
5 unchanged lines
1766 1835
        let quiet = Instant::now() - HEARTBEAT - Duration::from_secs(1);
1767 1836
        let (base, seen) = fake_hub(vec![204]);
1768 1837
        let hub = hub_at(base);
1769 -
        let mut sink = LogSink { hub: &hub, job: 1, buf: Vec::new(), last: quiet, gone: false, leaving: &NOT_LEAVING };
1838 +
        let mut sink = LogSink {
1839 +
            hub: &hub,
1840 +
            job: 1,
1841 +
            buf: Vec::new(),
1842 +
            last: quiet,
1843 +
            gone: false,
1844 +
            leaving: &NOT_LEAVING,
1845 +
            started: Instant::now(),
1846 +
            expected: None,
1847 +
        };
1770 1848
        assert!(!sink.cancelled());
1771 1849
        assert!(!sink.cancelled(), "asked again only after the heartbeat");
1772 1850
        assert_eq!(n(&seen), 1);
1773 1851
1774 1852
        let (base, seen) = fake_hub(vec![409]);
1775 1853
        let hub = hub_at(base);
1776 -
        let mut sink = LogSink { hub: &hub, job: 1, buf: Vec::new(), last: quiet, gone: false, leaving: &NOT_LEAVING };
1854 +
        let mut sink = LogSink {
1855 +
            hub: &hub,
1856 +
            job: 1,
1857 +
            buf: Vec::new(),
1858 +
            last: quiet,
1859 +
            gone: false,
1860 +
            leaving: &NOT_LEAVING,
1861 +
            started: Instant::now(),
1862 +
            expected: None,
1863 +
        };
1777 1864
        assert!(sink.cancelled());
1778 1865
        sink.record(Record { t: 0, stream: "out".into(), text: "late".into() });
1779 1866
        sink.flush();
…
13 unchanged lines
1793 1880
        let hub = hub_at(base.clone());
1794 1881
1795 1882
        // Log lines sent while it is away are kept, and go when it is back.
1796 -
        let mut sink = LogSink { hub: &hub, job: 1, buf: Vec::new(), last: Instant::now(), gone: false, leaving: &NOT_LEAVING };
1883 +
        let mut sink = LogSink {
1884 +
            hub: &hub,
1885 +
            job: 1,
1886 +
            buf: Vec::new(),
1887 +
            last: Instant::now(),
1888 +
            gone: false,
1889 +
            leaving: &NOT_LEAVING,
1890 +
            started: Instant::now(),
1891 +
            expected: None,
1892 +
        };
1797 1893
        for i in 0..450 {
1798 1894
            sink.buf.push(Record { t: i, stream: "out".into(), text: format!("line {i}") });
1799 1895
        }
…
43 unchanged lines
1843 1939
    fn unsent_log_lines_are_bounded() {
1844 1940
        let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
1845 1941
        let hub = hub_at(format!("http://127.0.0.1:{port}"));
1846 -
        let mut sink = LogSink { hub: &hub, job: 1, buf: Vec::new(), last: Instant::now(), gone: false, leaving: &NOT_LEAVING };
1942 +
        let mut sink = LogSink {
1943 +
            hub: &hub,
1944 +
            job: 1,
1945 +
            buf: Vec::new(),
1946 +
            last: Instant::now(),
1947 +
            gone: false,
1948 +
            leaving: &NOT_LEAVING,
1949 +
            started: Instant::now(),
1950 +
            expected: None,
1951 +
        };
1847 1952
        sink.buf = (0..MAX_UNSENT as u64 + 10).map(|i| Record { t: i, stream: "out".into(), text: String::new() }).collect();
1848 1953
        sink.send();
1849 1954
        assert_eq!(sink.buf.len(), MAX_UNSENT + 1);
…
19 unchanged lines
modified docs/spec.md +15 −6
…
821 unchanged lines
822 822
network, crashed) costs a restart of its job, not the job. A job handed
823 823
back more than 5 times ends as an error saying so.
824 824
825 -
As built (flow 0.3.2, OxygenHub 0.9.3): the runner as a library has
826 -
`runner::leave(reason)` beside `runner::stop()`: it claims nothing more,
827 -
stops the jobs it holds (their logs sent first) and releases them; if the
825 +
**Notice.** A departure may come with notice (a phone taken off charge
826 +
has a minute; a spot reclaim 30 seconds to 2 minutes). A claim carries
827 +
`expected_ms`, how long the job is expected to take, when the hub has
828 +
learned it: the median of the job's last five passing runs in that
829 +
repository (reused passes left out). A leaving runner finishes and
830 +
reports a job whose expected rest fits within the notice, and hands back
831 +
the others at once, and any job still running when the notice is up. A
832 +
job with no expected time is handed back at once.
833 +
834 +
As built (flow 0.3.2, OxygenHub 0.9.4): the runner as a library has
835 +
`runner::leave(reason)` (no notice) and `runner::leave_within(reason,
836 +
notice)` beside `runner::stop()`: each claims nothing more and stops the
837 +
jobs it does not finish (their logs sent first), releasing them; if the
828 838
release cannot be sent, the lease does it. A host's SIGTERM still drains
829 -
by finishing what it holds. Deciding from a job's expected time whether
830 -
to finish it within a departure's notice, and checkpoints (#77) so the
831 -
next attempt resumes rather than restarts, come next.
839 +
by finishing what it holds. Checkpoints (#77), so the next attempt
840 +
resumes rather than restarts, come next.
832 841
833 842
### 11.6 Conformance
834 843
…
11 unchanged lines