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

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

    /// How long the job is expected to take, from its recent passing runs,

    /// when the hub has learned it (#76): a runner leaving with some notice

    /// finishes a job whose rest fits it and hands back the others.

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub expected_ms: Option<u64>,

        expected_ms: Some(17_700),

    // The expected time goes with the claim, and an older hub's claim

    // without it still reads (#76).

    let json = serde_json::to_value(&old).unwrap();

    assert_eq!(json["expected_ms"], 17_700);

    let mut without = json.clone();

    without.as_object_mut().unwrap().remove("expected_ms");

    assert_eq!(serde_json::from_value::<crate::api::Claim>(without).unwrap().expected_ms, None);

/// A departure ([`leave_within`]): why, and when the runner must be gone.

#[derive(Clone, Debug)]

struct Leave {

    reason: String,

    by: Instant,

}

/// Set once the runner is leaving: a job whose expected rest fits before

/// the deadline finishes; the others are stopped and handed back

/// (oxygenflow#76).

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

    LEAVING.lock().unwrap_or_else(|p| p.into_inner()).as_ref().map(|l| l.reason.clone())

    leave_within(reason, Duration::ZERO);

}

/// [`leave`], with `notice`: a job the hub expects to finish within it

/// (the claim's `expected_ms`, less the time it has run) finishes and is

/// reported; the others, and any still running when the notice is up, are

/// handed back. For a phone taken off charge, a spot reclaim.

pub fn leave_within(reason: &str, notice: Duration) {

    *LEAVING.lock().unwrap_or_else(|p| p.into_inner()) = Some(Leave { reason: reason.to_string(), by: Instant::now() + notice });

    /// handed back, unless its expected rest fits the notice.

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

    /// When the job started, and how long the hub expects it to take.

    started: Instant,

    expected: Option<Duration>,

        let leave = self.leaving.lock().unwrap_or_else(|p| p.into_inner()).clone();

        if let Some(l) = leave {

            // Finished if what is left of it fits the notice; else handed

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

            let now = Instant::now();

            let fits = self.expected.is_some_and(|e| now + e.saturating_sub(self.started.elapsed()) <= l.by);

            if !fits || now >= l.by {

                self.flush();

                return true;

            }

    let mut sink = LogSink {

        hub,

        job: claim.job_id,

        buf: Vec::new(),

        last: Instant::now(),

        gone: false,

        leaving: &LEAVING,

        started: Instant::now(),

        expected: claim.expected_ms.map(Duration::from_millis),

    };

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

        let mut sink = LogSink {

            hub: &hub,

            job: 7,

            buf: Vec::new(),

            last: Instant::now(),

            gone: false,

            leaving: &leaving,

            started: Instant::now(),

            expected: None,

        };

        *leaving.lock().unwrap() = Some(Leave { reason: "off Wi-Fi".into(), by: Instant::now() });

    }

    /// oxygenflow#76: with notice, a job the hub expects to finish within

    /// it runs on; one that will not, or whose time is unknown, is handed

    /// back at once, and any job still running when the notice is up.

    #[test]

    fn a_job_that_fits_the_notice_finishes_and_the_others_go_back() {

        use microvm::Sink;

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

        let hub = hub_at(base);

        let leaving = std::sync::Mutex::new(Some(Leave { reason: "off charge".into(), by: Instant::now() + Duration::from_secs(60) }));

        let sink = |expected: Option<u64>, ran: u64| LogSink {

            hub: &hub,

            job: 1,

            buf: Vec::new(),

            last: Instant::now(),

            gone: false,

            leaving: &leaving,

            started: Instant::now() - Duration::from_secs(ran),

            expected: expected.map(Duration::from_secs),

        };

        assert!(!sink(Some(70), 30).cancelled(), "40 s left of it, within a minute: it finishes");

        assert!(sink(Some(600), 30).cancelled(), "nearly ten minutes left: handed back");

        assert!(sink(None, 0).cancelled(), "no idea how long: handed back");

        *leaving.lock().unwrap() = Some(Leave { reason: "off charge".into(), by: Instant::now() - Duration::from_secs(1) });

        assert!(sink(Some(70), 30).cancelled(), "the notice is up: handed back whatever was expected");

        let mut sink = LogSink {

            hub: &hub,

            job: 1,

            buf: Vec::new(),

            last: quiet,

            gone: false,

            leaving: &NOT_LEAVING,

            started: Instant::now(),

            expected: None,

        };

        let mut sink = LogSink {

            hub: &hub,

            job: 1,

            buf: Vec::new(),

            last: quiet,

            gone: false,

            leaving: &NOT_LEAVING,

            started: Instant::now(),

            expected: None,

        };

        let mut sink = LogSink {

            hub: &hub,

            job: 1,

            buf: Vec::new(),

            last: Instant::now(),

            gone: false,

            leaving: &NOT_LEAVING,

            started: Instant::now(),

            expected: None,

        };

        let mut sink = LogSink {

            hub: &hub,

            job: 1,

            buf: Vec::new(),

            last: Instant::now(),

            gone: false,

            leaving: &NOT_LEAVING,

            started: Instant::now(),

            expected: None,

        };

**Notice.** A departure may come with notice (a phone taken off charge

has a minute; a spot reclaim 30 seconds to 2 minutes). A claim carries

`expected_ms`, how long the job is expected to take, when the hub has

learned it: the median of the job's last five passing runs in that

repository (reused passes left out). A leaving runner finishes and

reports a job whose expected rest fits within the notice, and hands back

the others at once, and any job still running when the notice is up. A

job with no expected time is handed back at once.

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

`runner::leave(reason)` (no notice) and `runner::leave_within(reason,

notice)` beside `runner::stop()`: each claims nothing more and stops the

jobs it does not finish (their logs sent first), releasing them; if the

by finishing what it holds. Checkpoints (#77), so the next attempt

resumes rather than restarts, come next.