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

- **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.

    /// Hand back a job this runner holds as it leaves (oxygenflow#76): the

    /// hub queues it again ahead of new work for another runner.

    pub fn release(&self, job: i64, reason: &str) -> Result<()> {

        self.post_patiently(&format!("/flow/jobs/{job}/release"), &Decline { reason: reason.into() })?;

        Ok(())

    }

    *LEAVING.lock().unwrap_or_else(|p| p.into_inner()) = None;

}

/// Why the runner is leaving, once [`leave`] says so: the jobs it holds are

/// stopped and handed back rather than finished (oxygenflow#76).

static LEAVING: std::sync::Mutex<Option<String>> = std::sync::Mutex::new(None);

fn leaving() -> Option<String> {

    LEAVING.lock().unwrap_or_else(|p| p.into_inner()).clone()

}

/// Ask a runner started with [`serve_embedded`] to leave now: claim nothing

/// more, stop the jobs it holds and hand them back to the hub, which queues

/// them ahead of new work for another runner (oxygenflow#76), then return.

/// For a departure with no time to finish: a phone off the allowed network.

pub fn leave(reason: &str) {

    *LEAVING.lock().unwrap_or_else(|p| p.into_inner()) = Some(reason.to_string());

    stop();

                        match leaving() {

                            // Stopped as the runner leaves: another runner takes it.

                            Some(why) if cancelled => {

                                eprintln!("job {id}: handed back: {why}");

                                if let Err(e) = hub.release(id, &why) {

                                    eprintln!("job {id}: handing back: {e:#}; the hub requeues it once this runner's lease lapses");

                                }

                            }

                            _ if !cancelled => report_finish(home, hub, id, &finish),

                            _ => {}

    /// Set when the runner leaves ([`LEAVING`]): the job stops, to be

    /// handed back.

    leaving: &'a std::sync::Mutex<Option<String>>,

        if self.leaving.lock().unwrap_or_else(|p| p.into_inner()).is_some() {

            // Handed back as the runner leaves: what the job said so far first.

            self.flush();

            return true;

        }

    let mut sink = LogSink { hub, job: claim.job_id, buf: Vec::new(), last: Instant::now(), gone: false, leaving: &LEAVING };

    }

    static NOT_LEAVING: std::sync::Mutex<Option<String>> = std::sync::Mutex::new(None);

    /// oxygenflow#76: once the runner leaves, a job's sink says it is

    /// cancelled, having sent what the job said so far, and the job is

    /// handed back to the hub.

    #[test]

    fn a_leaving_runner_stops_its_job_and_hands_it_back() {

        use microvm::Sink;

        let leaving = std::sync::Mutex::new(None);

        let (base, seen) = fake_hub(vec![204, 204]);

        let hub = hub_at(base);

        let mut sink = LogSink { hub: &hub, job: 7, buf: Vec::new(), last: Instant::now(), gone: false, leaving: &leaving };

        sink.buf.push(Record { t: 1, stream: "out".into(), text: "step 1".into() });

        assert!(!sink.cancelled());

        *leaving.lock().unwrap() = Some("off Wi-Fi".into());

        assert!(sink.cancelled());

        assert!(sink.buf.is_empty(), "flushed before the job stops");

        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 1);

        hub.release(7, "off Wi-Fi").unwrap();

        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 2);

        let mut sink = LogSink { hub: &hub, job: 1, buf: Vec::new(), last: quiet, gone: false, leaving: &NOT_LEAVING };

        let mut sink = LogSink { hub: &hub, job: 1, buf: Vec::new(), last: quiet, gone: false, leaving: &NOT_LEAVING };

        let mut sink = LogSink { hub: &hub, job: 1, buf: Vec::new(), last: Instant::now(), gone: false, leaving: &NOT_LEAVING };

        let mut sink = LogSink { hub: &hub, job: 1, buf: Vec::new(), last: Instant::now(), gone: false, leaving: &NOT_LEAVING };

### 11.5.2 Handing back (oxygenflow#76)

A runner that leaves while it holds a job, with no time to finish it,

hands the job back: `POST /api/v1/flow/jobs/{id}/release` with

`{"reason": …}` (a phone off its allowed network, docs/handover.md §6).

The hub queues the job again **ahead of new work**, whatever its

priority, adds a line to its log saying who handed it back and why, and

lets the next attempt's log follow on. It is not a decline: the runner is

not held from the job and it does not count towards unplaceable.

**Leases.** Holding a job is a lease the runner keeps by being heard from:

claims, a heartbeat every 30 seconds while it claims nothing, its jobs'

logs. When the hub has not heard from a runner for 3 minutes, its jobs

are handed back the same way, so a runner that vanishes (out of every

network, crashed) costs a restart of its job, not the job. A job handed

back more than 5 times ends as an error saying so.

As built (flow 0.3.2, OxygenHub 0.9.3): the runner as a library has

`runner::leave(reason)` beside `runner::stop()`: it claims nothing more,

stops the jobs it holds (their logs sent first) and releases them; if the

release cannot be sent, the lease does it. A host's SIGTERM still drains

by finishing what it holds. Deciding from a job's expected time whether

to finish it within a departure's notice, and checkpoints (#77) so the

next attempt resumes rather than restarts, come next.