feat(runner): restarts cost no job (closes #49, closes #50) 89171fca079c

- **Restarts cost no job (#49, #50, spec §11.5.1).** A job's uploads and declines wait out a hub that is away for up to three minutes (a deploy restarting it), through refused connections as well as server errors, and log lines it could not send go with the next batch instead of being dropped. A runner told to stop drains: it claims nothing more, finishes and reports the jobs it holds, then exits; a second signal stops it at once. With `KillMode=mixed` and a `TimeoutStopSec` longer than the longest job in its unit, deploying a runner no longer has to wait for it to be idle.

		sudo systemctl restart --no-block flow-runner; echo "flow-runner: restarting (it drains: the job it holds finishes first)"'

        let r = retrying(&url, delays, false, || {

            self.agent.post(&url).set("Authorization", &format!("Bearer {}", self.token)).send_json(body).map_err(Box::new)

        })?;

        Ok(if r.status() == 204 { None } else { Some(r) })

    }

    /// POST what a job leaves behind (oxygenflow#49): through a hub that is

    /// away for a few minutes, as while a deploy restarts it, so a job's

    /// declines and uploads outlast it.

    fn post_patiently(&self, path: &str, body: &impl serde::Serialize) -> Result<Option<ureq::Response>> {

        let url = format!("{}/api/v1{path}", self.base);

        let r = retrying(&url, &PATIENT_DELAYS, true, || {

            self.agent.post(&url).set("Authorization", &format!("Bearer {}", self.token)).send_json(body).map_err(Box::new)

        })?;

        Ok(if r.status() == 204 { None } else { Some(r) })

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

        let r = retrying(&url, &PATIENT_DELAYS, true, || {

            self.agent

                .set("Content-Type", "application/octet-stream")

                .send_bytes(bytes)

                .map_err(Box::new)

        })?;

        let v: serde_json::Value = r.into_json()?;

        v["blob"].as_str().map(String::from).context("the hub named no blob")

        let r = retrying(&url, &PATIENT_DELAYS, true, || {

            self.agent

                .set("X-Flow-Report-Path", path)

                .send_bytes(bytes)

                .map_err(Box::new)

        })?;

        let v: serde_json::Value = r.into_json()?;

        v["blob"].as_str().map(String::from).context("the hub named no blob")

/// Waits for what a job leaves behind (oxygenflow#49): three minutes in

/// all, longer than a hub takes to restart.

const PATIENT_DELAYS: [Duration; 15] = {

    let mut d = [Duration::from_secs(15); 15];

    d[0] = Duration::from_secs(1);

    d[1] = Duration::from_secs(2);

    d[2] = Duration::from_secs(4);

    d[3] = Duration::from_secs(8);

    d

};

/// Send a request until the hub answers, waiting `delays` in turn after a

/// server error (5xx: nothing was committed) and, when `connection`, after

/// the hub could not be reached at all. A refusal (4xx) is final, as a

/// [`Refused`] the caller can look at.

fn retrying(

    url: &str,

    delays: &[Duration],

    connection: bool,

    mut send: impl FnMut() -> Result<ureq::Response, Box<ureq::Error>>,

) -> Result<ureq::Response> {

    let mut delays = delays.iter();

    loop {

        let why = match send().map_err(|e| *e) {

            Ok(r) => return Ok(r),

            Err(ureq::Error::Status(code, r)) if code < 500 => {

                let text = r.into_string().unwrap_or_default();

                return Err(anyhow::Error::new(Refused { code, text }).context(url.to_string()));

            }

            Err(ureq::Error::Status(code, r)) => format!("HTTP {code}: {}", r.into_string().unwrap_or_default()),

            Err(e) if connection => e.to_string(),

            Err(e) => bail!("{url}: {e}"),

        };

        match delays.next() {

            Some(d) => std::thread::sleep(*d),

            None => bail!("{url}: {why}"),

        }

    }

}

    drain_on_stop();

        let mut said_draining = false;

            // Draining (oxygenflow#50): told to stop, claim nothing more and

            // stop once every job held is done and reported.

            if stopping() {

                let (jobs, holding) = (used.lock().unwrap().jobs, held.lock().unwrap().len());

                if jobs == 0 && holding == 0 {

                    eprintln!("flow runner {name}: drained, stopping");

                    break;

                }

                if !said_draining {

                    eprintln!("flow runner {name}: draining: claiming nothing more, stopping when the {jobs} job(s) held are done (stop again to stop now)");

                    said_draining = true;

                }

                std::thread::sleep(Duration::from_millis(500));

                continue;

            }

        Ok(())

}

/// How many times the runner was told to stop (SIGTERM or SIGINT).

static STOPS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);

extern "C" fn on_stop(_: libc::c_int) {

    // A second stop means now: only async-signal-safe calls here.

    if STOPS.fetch_add(1, std::sync::atomic::Ordering::SeqCst) >= 1 {

        unsafe { libc::_exit(143) };

    }

}

/// Drain instead of dying on the first SIGTERM or SIGINT (oxygenflow#50),

/// so restarting a runner never kills the job it runs: the unit's

/// `KillMode=mixed` sends the signal to the runner alone, not its microVMs.

fn drain_on_stop() {

    let handler = on_stop as extern "C" fn(libc::c_int) as libc::sighandler_t;

    unsafe {

        libc::signal(libc::SIGTERM, handler);

        libc::signal(libc::SIGINT, handler);

    }

fn stopping() -> bool {

    STOPS.load(std::sync::atomic::Ordering::SeqCst) > 0

}

/// The most log lines sent at once, as a busy job's sink flushes them.

const BATCH: usize = 200;

/// The most log lines kept while the hub cannot be reached.

const MAX_UNSENT: usize = 50_000;

        if self.buf.len() >= BATCH || self.last.elapsed() >= Duration::from_secs(1) {

    /// Send what is buffered, even nothing, in batches of at most

    /// `BATCH` lines.

        loop {

            if self.gone {

                return;

            }

            let n = self.buf.len().min(BATCH);

            let batch: Vec<Record> = self.buf.drain(..n).collect();

            match self.hub.log(self.job, batch.clone()) {

                Ok(()) => {}

                Err(e) if job_gone(&e) => self.gone = true,

                Err(e) => {

                    // The hub is away (a restart, say): keep the lines for

                    // the next batch, up to a bound, so the log loses none

                    // (oxygenflow#49).

                    eprintln!("job {}: log: {e:#}; kept for the next batch", self.job);

                    let mut kept = batch;

                    kept.append(&mut self.buf);

                    let over = kept.len().saturating_sub(MAX_UNSENT);

                    if over > 0 {

                        kept.drain(..over);

                        kept.insert(

                            0,

                            Record {

                                t: 0,

                                stream: "flow".into(),

                                text: format!("({over} log lines lost while the hub could not be reached)"),

                            },

                        );

                    }

                    self.buf = kept;

                    break;

                }

            }

            if self.buf.is_empty() {

                break;

            }

        fake_hub_on(std::net::TcpListener::bind("127.0.0.1:0").unwrap(), codes)

    }

    /// A hub at `listener` answering each request with the next of `codes`.

    fn fake_hub_on(listener: std::net::TcpListener, codes: Vec<u16>) -> (String, std::sync::Arc<std::sync::atomic::AtomicUsize>) {

    }

    /// oxygenflow#49: a hub that is away for a while (a deploy restarting

    /// it) costs a job's uploads a wait, not the upload, and costs its log

    /// no lines.

    #[test]

    fn a_job_outlasts_a_hub_restart() {

        let n = |seen: &std::sync::Arc<std::sync::atomic::AtomicUsize>| seen.load(std::sync::atomic::Ordering::SeqCst);

        // A port nobody listens on yet: the hub is restarting.

        let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();

        let base = format!("http://127.0.0.1:{port}");

        let hub = hub_at(base.clone());

        // Log lines sent while it is away are kept, and go when it is back.

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

        for i in 0..450 {

            sink.buf.push(Record { t: i, stream: "out".into(), text: format!("line {i}") });

        }

        sink.send();

        assert_eq!(sink.buf.len(), 450, "nothing lost while the hub is away");

        let (_, seen) = fake_hub_on(std::net::TcpListener::bind(("127.0.0.1", port)).unwrap(), vec![204, 204, 204]);

        sink.send();

        assert!(sink.buf.is_empty() && !sink.gone);

        assert_eq!(n(&seen), 3, "in batches of 200: 200, 200 and 50");

        // An upload waits for the hub to come back, through refused

        // connections and server errors alike.

        let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();

        let url = format!("http://127.0.0.1:{port}/x");

        let agent = ureq::AgentBuilder::new().timeout(Duration::from_secs(5)).build();

        let later = std::thread::spawn(move || {

            std::thread::sleep(Duration::from_millis(150));

            fake_hub_on(std::net::TcpListener::bind(("127.0.0.1", port)).unwrap(), vec![502, 204])

        });

        let quick = [Duration::from_millis(50); 20];

        let r = retrying(&url, &quick, true, || agent.post(&url).send_string("{}").map_err(Box::new)).unwrap();

        assert_eq!(r.status(), 204);

        assert_eq!(n(&later.join().unwrap().1), 2, "a 502 from the proxy, then the hub");

        // Without `connection`, a hub that cannot be reached ends it at once.

        let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();

        let url = format!("http://127.0.0.1:{port}/x");

        assert!(retrying(&url, &quick, false, || agent.post(&url).send_string("{}").map_err(Box::new)).is_err());

        assert!(PATIENT_DELAYS.iter().sum::<Duration>() >= Duration::from_secs(180), "longer than a hub takes to restart");

    }

    /// oxygenflow#50: the first stop drains (the second, not called here,

    /// exits at once).

    #[test]

    fn a_stop_drains_the_runner() {

        assert!(!stopping());

        on_stop(libc::SIGTERM);

        assert!(stopping(), "told to stop, it claims nothing more");

    }

    /// Lines kept for a hub that stays away are bounded, saying how many went.

    #[test]

    fn unsent_log_lines_are_bounded() {

        let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();

        let hub = hub_at(format!("http://127.0.0.1:{port}"));

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

        sink.buf = (0..MAX_UNSENT as u64 + 10).map(|i| Record { t: i, stream: "out".into(), text: String::new() }).collect();

        sink.send();

        assert_eq!(sink.buf.len(), MAX_UNSENT + 1);

        assert_eq!(sink.buf[0].text, "(10 log lines lost while the hub could not be reached)");

        assert_eq!(sink.buf[1].t, 10, "the oldest go first");

`https://earth.andymccall.xyz:7343`. The unit has `KillMode=mixed` and

`TimeoutStopSec=75min`, so a restart drains the runner instead of killing

its job (spec §11.5.1).

### 11.5.1 Restarts (oxygenflow#49, #50)

Neither a hub restart nor a runner restart costs a job anything.

- **The hub away.** A job's uploads (artifacts, report files) and declines

  are tried again through refused connections and server errors (5xx,

  as a proxy answers while the hub restarts) for three minutes, waiting

  1, 2, 4, 8 and then 15 seconds; a refusal (4xx) is final. Log lines

  the hub could not take are kept and sent with the next batch, 200 lines

  at a time, up to 50,000 (beyond that the oldest go, and the log says how

  many). A job's result is kept on disk and offered for two hours, as

  before. Claims are not retried through a lost connection, so a job is

  never claimed twice; the runner simply asks again.

- **The runner restarted.** The first SIGTERM or SIGINT drains the runner:

  it claims nothing more and stops once every job it holds has ended and

  been reported. A second signal stops it at once. Its systemd unit says

  `KillMode=mixed`, so only the runner is signalled and its microVMs run

  on, and `TimeoutStopSec` longer than the longest job, after which

  systemd kills what is left. A new binary is installed by replacing it

  and restarting the unit (`systemctl restart --no-block`): the old runner

  finishes its job, and `Restart=always` starts the new one.