OXYGENHUB

A runner that leaves hands its jobs back (refs #76) 06a4c472b385

runner::leave(reason) stops claiming, stops the jobs it holds with what
they logged sent first, and hands them back with POST
/flow/jobs/{id}/release (OxygenHub #312), which queues them ahead of new
work for another runner. If the release cannot be sent, the hub's lease
does it after 3 minutes. 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 18e62edc58df
When
· 2026-10-25 17:50 UTC
Signature
◈ sealed by agent:claude@andymccall
Hash
06a4c472b385a86d927329debd9dac71a432292c516682b20ad73af06849e7dd
Depends on
1b2229dab1a4 280ef951d193 337afb8f6043 4be5e4cd0af6 5cf1e8aefbb6 89171fca079c a083cc802892 a4b3f8cb9aae b11e30f5031d c3db750ed96e e0918b887850 fd2d4822f7f3
Depended on by
85275e1ef3b7 1b160d372d9e 9e5fd161efc3
In bookmarks
main
In tags
v0.4.0 v0.5.0 v0.5.1 v0.5.2 v0.5.3
Edits
15 atoms, 0 tree operations, 3 files

Browse the tree at this change

modified CHANGELOG.md +1 −0
…
3 unchanged lines
4 4
5 5
## Unreleased
6 6
7 +
- **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.
7 8
- **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).
8 9
- **Built on oxygen's main with #140 and #141:** the runner's store puts objects in place where hard links are refused (Android, so the phone runner can clone), and the `deps/oxygen` pin conflict two replacements in one change left on main is settled with `oxy resolve deps/oxygen`.
9 10
- **The WASM tier (#75):** `runtime "wasm"` runs WASI preview 1 modules with wasmtime 49 on any runner (`flow tiers` offers `wasm` everywhere): no rootfs, no network, the job's files at `/src`, its mounts at their targets, the memory limit enforced by wasmtime. `exec` names a module or a glob of them (`target/wasm32-wasip1/debug/deps/*.wasm` runs each test binary in turn); the receipt names wasmtime's version and every module's hash. The tier phones and browsers will run (docs/handover.md). Secrets and identities are not offered in it yet.
…
89 unchanged lines
modified crates/flow/src/runner.rs +70 −7
…
101 unchanged lines
102 102
        Ok(())
103 103
    }
104 104
105 +
    /// Hand back a job this runner holds as it leaves (oxygenflow#76): the
106 +
    /// hub queues it again ahead of new work for another runner.
107 +
    pub fn release(&self, job: i64, reason: &str) -> Result<()> {
108 +
        self.post_patiently(&format!("/flow/jobs/{job}/release"), &Decline { reason: reason.into() })?;
109 +
        Ok(())
110 +
    }
111 +
105 112
    /// Say this runner is there while it claims nothing: its slots are full,
106 113
    /// or it is draining (oxygenhub#243). Once, no retries: the next comes
107 114
    /// soon. A hub without the endpoint (404) is not an error.
…
472 unchanged lines
580 587
/// it holds are done and reported. It may be started again afterwards.
581 588
pub fn serve_embedded(home: &Home) -> Result<()> {
582 589
    STOPS.store(0, std::sync::atomic::Ordering::SeqCst);
590 +
    *LEAVING.lock().unwrap_or_else(|p| p.into_inner()) = None;
583 591
    serve_until_drained(home)
592 +
}
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);
597 +
598 +
fn leaving() -> Option<String> {
599 +
    LEAVING.lock().unwrap_or_else(|p| p.into_inner()).clone()
600 +
}
601 +
602 +
/// Ask a runner started with [`serve_embedded`] to leave now: claim nothing
603 +
/// more, stop the jobs it holds and hand them back to the hub, which queues
604 +
/// them ahead of new work for another runner (oxygenflow#76), then return.
605 +
/// For a departure with no time to finish: a phone off the allowed network.
606 +
pub fn leave(reason: &str) {
607 +
    *LEAVING.lock().unwrap_or_else(|p| p.into_inner()) = Some(reason.to_string());
608 +
    stop();
584 609
}
585 610
586 611
/// Ask a runner started with [`serve_embedded`] to drain: claim nothing
…
170 unchanged lines
757 782
                        crate::metrics::RUNNING.fetch_sub(1, std::sync::atomic::Ordering::Relaxed);
758 783
                        let cancelled = finish.error.as_deref() == Some(CANCELLED);
759 784
                        crate::metrics::finished(finish.exit_code, finish.error.as_deref(), finish.duration_ms, cancelled);
760 -
                        if !cancelled {
761 -
                            report_finish(home, hub, id, &finish);
785 +
                        match leaving() {
786 +
                            // Stopped as the runner leaves: another runner takes it.
787 +
                            Some(why) if cancelled => {
788 +
                                eprintln!("job {id}: handed back: {why}");
789 +
                                if let Err(e) = hub.release(id, &why) {
790 +
                                    eprintln!("job {id}: handing back: {e:#}; the hub requeues it once this runner's lease lapses");
791 +
                                }
792 +
                            }
793 +
                            _ if !cancelled => report_finish(home, hub, id, &finish),
794 +
                            _ => {}
762 795
                        }
763 796
                        held.lock().unwrap().remove(&id);
764 797
                        let mut u = used.lock().unwrap();
…
48 unchanged lines
813 846
    buf: Vec<Record>,
814 847
    last: Instant,
815 848
    gone: bool,
849 +
    /// Set when the runner leaves ([`LEAVING`]): the job stops, to be
850 +
    /// handed back.
851 +
    leaving: &'a std::sync::Mutex<Option<String>>,
816 852
}
817 853
818 854
/// The most log lines sent at once, as a busy job's sink flushes them.
…
11 unchanged lines
830 866
    }
831 867
832 868
    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;
873 +
        }
833 874
        if !self.gone && self.last.elapsed() >= HEARTBEAT {
834 875
            self.send();
835 876
        }
…
99 unchanged lines
935 976
936 977
fn run_claim(home: &Home, hub: &Hub, cfg: &RunnerConfig, key: &RunnerKey, claim: &Claim) -> Result<Finish> {
937 978
    let started = Instant::now();
938 -
    let mut sink = LogSink { hub, job: claim.job_id, buf: Vec::new(), last: Instant::now(), gone: false };
979 +
    let mut sink = LogSink { hub, job: claim.job_id, buf: Vec::new(), last: Instant::now(), gone: false, leaving: &LEAVING };
939 980
    let note =
940 981
        |sink: &mut LogSink, text: String| sink.push(Record { t: started.elapsed().as_millis() as u64, stream: "flow".into(), text });
941 982
    let heads: Vec<ChangeHash> = claim.heads.iter().map(|h| ChangeHash::from_hex(h).context("bad head")).collect::<Result<_>>()?;
…
750 unchanged lines
1692 1733
        let (base, seen) = fake_hub(vec![401]);
1693 1734
        assert!(hub_at(base).post_with("/flow/claim", &serde_json::json!({}), &quick).is_err());
1694 1735
        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 1, "a refusal is final");
1736 +
    }
1737 +
1738 +
    static NOT_LEAVING: std::sync::Mutex<Option<String>> = std::sync::Mutex::new(None);
1739 +
1740 +
    /// oxygenflow#76: once the runner leaves, a job's sink says it is
1741 +
    /// cancelled, having sent what the job said so far, and the job is
1742 +
    /// handed back to the hub.
1743 +
    #[test]
1744 +
    fn a_leaving_runner_stops_its_job_and_hands_it_back() {
1745 +
        use microvm::Sink;
1746 +
        let leaving = std::sync::Mutex::new(None);
1747 +
        let (base, seen) = fake_hub(vec![204, 204]);
1748 +
        let hub = hub_at(base);
1749 +
        let mut sink = LogSink { hub: &hub, job: 7, buf: Vec::new(), last: Instant::now(), gone: false, leaving: &leaving };
1750 +
        sink.buf.push(Record { t: 1, stream: "out".into(), text: "step 1".into() });
1751 +
        assert!(!sink.cancelled());
1752 +
        *leaving.lock().unwrap() = Some("off Wi-Fi".into());
1753 +
        assert!(sink.cancelled());
1754 +
        assert!(sink.buf.is_empty(), "flushed before the job stops");
1755 +
        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 1);
1756 +
        hub.release(7, "off Wi-Fi").unwrap();
1757 +
        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 2);
1695 1758
    }
1696 1759
1697 1760
    /// oxygenflow#32: a quiet job asks the hub every few seconds whether
…
5 unchanged lines
1703 1766
        let quiet = Instant::now() - HEARTBEAT - Duration::from_secs(1);
1704 1767
        let (base, seen) = fake_hub(vec![204]);
1705 1768
        let hub = hub_at(base);
1706 -
        let mut sink = LogSink { hub: &hub, job: 1, buf: Vec::new(), last: quiet, gone: false };
1769 +
        let mut sink = LogSink { hub: &hub, job: 1, buf: Vec::new(), last: quiet, gone: false, leaving: &NOT_LEAVING };
1707 1770
        assert!(!sink.cancelled());
1708 1771
        assert!(!sink.cancelled(), "asked again only after the heartbeat");
1709 1772
        assert_eq!(n(&seen), 1);
1710 1773
1711 1774
        let (base, seen) = fake_hub(vec![409]);
1712 1775
        let hub = hub_at(base);
1713 -
        let mut sink = LogSink { hub: &hub, job: 1, buf: Vec::new(), last: quiet, gone: false };
1776 +
        let mut sink = LogSink { hub: &hub, job: 1, buf: Vec::new(), last: quiet, gone: false, leaving: &NOT_LEAVING };
1714 1777
        assert!(sink.cancelled());
1715 1778
        sink.record(Record { t: 0, stream: "out".into(), text: "late".into() });
1716 1779
        sink.flush();
…
13 unchanged lines
1730 1793
        let hub = hub_at(base.clone());
1731 1794
1732 1795
        // Log lines sent while it is away are kept, and go when it is back.
1733 -
        let mut sink = LogSink { hub: &hub, job: 1, buf: Vec::new(), last: Instant::now(), gone: false };
1796 +
        let mut sink = LogSink { hub: &hub, job: 1, buf: Vec::new(), last: Instant::now(), gone: false, leaving: &NOT_LEAVING };
1734 1797
        for i in 0..450 {
1735 1798
            sink.buf.push(Record { t: i, stream: "out".into(), text: format!("line {i}") });
1736 1799
        }
…
43 unchanged lines
1780 1843
    fn unsent_log_lines_are_bounded() {
1781 1844
        let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
1782 1845
        let hub = hub_at(format!("http://127.0.0.1:{port}"));
1783 -
        let mut sink = LogSink { hub: &hub, job: 1, buf: Vec::new(), last: Instant::now(), gone: false };
1846 +
        let mut sink = LogSink { hub: &hub, job: 1, buf: Vec::new(), last: Instant::now(), gone: false, leaving: &NOT_LEAVING };
1784 1847
        sink.buf = (0..MAX_UNSENT as u64 + 10).map(|i| Record { t: i, stream: "out".into(), text: String::new() }).collect();
1785 1848
        sink.send();
1786 1849
        assert_eq!(sink.buf.len(), MAX_UNSENT + 1);
…
19 unchanged lines
modified docs/spec.md +25 −0
…
804 unchanged lines
805 805
  and restarting the unit (`systemctl restart --no-block`): the old runner
806 806
  finishes its job, and `Restart=always` starts the new one.
807 807
808 +
### 11.5.2 Handing back (oxygenflow#76)
809 +
810 +
A runner that leaves while it holds a job, with no time to finish it,
811 +
hands the job back: `POST /api/v1/flow/jobs/{id}/release` with
812 +
`{"reason": …}` (a phone off its allowed network, docs/handover.md §6).
813 +
The hub queues the job again **ahead of new work**, whatever its
814 +
priority, adds a line to its log saying who handed it back and why, and
815 +
lets the next attempt's log follow on. It is not a decline: the runner is
816 +
not held from the job and it does not count towards unplaceable.
817 +
818 +
**Leases.** Holding a job is a lease the runner keeps by being heard from:
819 +
claims, a heartbeat every 30 seconds while it claims nothing, its jobs'
820 +
logs. When the hub has not heard from a runner for 3 minutes, its jobs
821 +
are handed back the same way, so a runner that vanishes (out of every
822 +
network, crashed) costs a restart of its job, not the job. A job handed
823 +
back more than 5 times ends as an error saying so.
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
828 +
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.
832 +
808 833
### 11.6 Conformance
809 834
810 835
1. Architecture Safety: a job whose image is `x86_64` MUST never be
…
10 unchanged lines